""" Module 5: Resource & Budget Optimizer La "Secret Sauce" - Adapte protocole aux contraintes réelles """ from typing import List, Dict, Any, Optional from app.models.schemas import ExperimentalProtocol class ResourceOptimizer: \"\"\"Optimise protocole expérimental en fonction des contraintes\"\"\" def __init__(self): self.optimization_prompts = self._init_optimization_prompts() self.material_equivalents = self._init_material_equivalents() def _init_optimization_prompts(self) -> Dict[str, str]: \"\"\"K2 Think prompts pour optimisation ressources\"\"\" return { "resource_mapping": \"\"\"Mappe les ressources requises aux ressources disponibles: 1. Matériaux/réactifs: identifier alternatives validées 2. Équipement: remplacer par équivalents disponibles 3. Personnel: adapter chronoprogramme 4. Locaux: adapter mise en place Pour chaque adaptation: impact sur validité (0-1) Proposer alternatives seulement si validité conservée\"\"\", "budget_optimization": \"\"\"Optimise budget sans perdre rigueur: 1. Identifier dépenses non-critiques 2. Proposer alternatives moins chères validées 3. Prioriser expériences clés 4. Stagner si nécessaire Retour: protocole ajusté + budget minimum + recommandations\"\"\", "timeline_adjustment": \"\"\"Adapte calendrier aux contraintes temps: 1. Identifier étapes parallélisables 2. Compresser sans compromis validité 3. Proposer timeline alternatives 4. Indiquer impact sur conclusions\"\"\", } def _init_material_equivalents(self) -> Dict[str, List[Dict]]: \"\"\"Database d'équivalents matériaux validés\"\"\" return { "HPLC_system": [ { "original": "High-performance liquid chromatography system", "alternative": "Gas chromatography", "cost_ratio": 0.6, "validity_impact": 0.95, "conditions": "If compounds are volatile" }, { "original": "High-performance liquid chromatography system", "alternative": "UV-Visible spectrophotometry", "cost_ratio": 0.3, "validity_impact": 0.75, "conditions": "For qualitative confirmation only" } ], "cell_culture": [ { "original": "Primary cell culture from patient biopsies", "alternative": "Established cell lines", "cost_ratio": 0.2, "validity_impact": 0.80, "conditions": "If relevant cell line exists" }, { "original": "3D tissue culture system", "alternative": "2D monolayer culture", "cost_ratio": 0.4, "validity_impact": 0.70, "conditions": "For initial screening only" } ], "animal_model": [ { "original": "Primate model", "alternative": "Rodent model with human-equivalent genetics", "cost_ratio": 0.1, "validity_impact": 0.85, "conditions": "For mechanism study; translational risk" }, { "original": "In-vivo animal model", "alternative": "Organoid model", "cost_ratio": 0.6, "validity_impact": 0.75, "conditions": "For disease phenotyping" } ] } def optimize_protocol( self, protocol: ExperimentalProtocol, constraints: Dict[str, Any] ) -> Dict[str, Any]: \"\"\" Optimise protocole selon contraintes Args: protocol: Protocole initial constraints: { "budget_usd": float, "duration_days": float, "unavailable_materials": [str], "available_equipment": [str], "location_type": str, "team_size": int } \"\"\" optimization = { "original_protocol": { "estimated_cost": protocol.estimated_budget_usd, "estimated_duration": protocol.estimated_duration_days, "sample_size": self._estimate_sample_size(protocol), }, "constraints": constraints, "adaptations": [], "optimized_protocol": None, "optimized_cost": None, "validity_impact_score": 1.0, "recommendations": [] } # Optimization round 1: Material substitutions material_adaptations = self._adapt_materials( protocol, constraints.get("unavailable_materials", []) ) optimization["adaptations"].extend(material_adaptations) # Optimization round 2: Timeline compression if constraints.get("duration_days") and \ constraints["duration_days"] < protocol.estimated_duration_days: timeline_adapt = self._compress_timeline( protocol, constraints["duration_days"] ) optimization["adaptations"].append(timeline_adapt) # Optimization round 3: Budget reduction if constraints.get("budget_usd"): budget_adapt = self._reduce_budget( protocol, constraints["budget_usd"], material_adaptations ) optimization["adaptations"].append(budget_adapt) optimization["optimized_cost"] = budget_adapt.get("new_cost") # Optimization round 4: Sample size adjustment if constraints.get("budget_usd"): sample_adapt = self._adjust_sample_size( protocol, optimization["optimized_cost"] or protocol.estimated_budget_usd ) optimization["adaptations"].append(sample_adapt) # Calculate overall validity impact validity_impacts = [a.get("validity_impact", 1.0) for a in optimization["adaptations"]] optimization["validity_impact_score"] = min(validity_impacts) if validity_impacts else 1.0 # Generate recommendations optimization["recommendations"] = self._generate_recommendations( optimization, constraints ) return optimization def _adapt_materials( self, protocol: ExperimentalProtocol, unavailable_materials: List[str] ) -> List[Dict[str, Any]]: \"\"\"Substitue matériaux non-disponibles\"\"\" adaptations = [] for unavailable in unavailable_materials: # Find equivalents in database if unavailable in self.material_equivalents: alternatives = self.material_equivalents[unavailable] best_alt = alternatives[0] # Best match adaptation = { "type": "material_substitution", "original": unavailable, "proposed_alternative": best_alt["alternative"], "cost_ratio": best_alt["cost_ratio"], "validity_impact": best_alt["validity_impact"], "conditions": best_alt["conditions"], "status": "validated" if best_alt["validity_impact"] > 0.85 else "requires_pilot" } adaptations.append(adaptation) return adaptations def _compress_timeline( self, protocol: ExperimentalProtocol, target_duration_days: float ) -> Dict[str, Any]: \"\"\"Compresse calendrier\"\"\" original_steps = len(protocol.steps) parallelizable_steps = [s for s in protocol.steps if self._is_parallelizable(s)] compression = { "type": "timeline_compression", "original_duration_days": protocol.estimated_duration_days, "target_duration_days": target_duration_days, "compression_ratio": target_duration_days / protocol.estimated_duration_days, "parallelizable_steps": len(parallelizable_steps), "recommended_measures": [ f"Parallelize {len(parallelizable_steps)} non-dependent steps", "Add team members to compress serial steps", "Use overnight incubations for waiting periods" ], "validity_impact": 0.95 if target_duration_days / protocol.estimated_duration_days > 0.7 else 0.80, "risk_increase": "medium" if target_duration_days / protocol.estimated_duration_days < 0.7 else "low" } return compression def _reduce_budget( self, protocol: ExperimentalProtocol, target_budget_usd: float, material_adaptations: List[Dict] ) -> Dict[str, Any]: \"\"\"Réduit budget tout en conservant rigueur\"\"\" original_cost = protocol.estimated_budget_usd or 15000 cost_savings = original_cost - target_budget_usd savings_percent = (cost_savings / original_cost) * 100 reduction = { "type": "budget_reduction", "original_budget": original_cost, "target_budget": target_budget_usd, "required_savings": cost_savings, "savings_percent": savings_percent, "potential_reductions": [ { "category": "Equipment rental", "potential_saving": "20-40%", "method": "Use cheaper alternatives or institutional equipment" }, { "category": "Materials", "potential_saving": "30-50%", "method": "Bulk purchase; negotiate with suppliers" }, { "category": "Personnel", "potential_saving": "10-20%", "method": "Use graduate students for routine tasks" } ], "material_savings": self._calculate_material_savings(material_adaptations), "new_cost": target_budget_usd, "validity_impact": 0.90 if savings_percent < 50 else 0.75, "feasibility": "high" if savings_percent < 40 else "medium" if savings_percent < 60 else "low" } return reduction def _calculate_material_savings(self, adaptations: List[Dict]) -> float: \"\"\"Calcule économies matériaux\"\"\" total_savings = sum( 1 - a.get("cost_ratio", 1) for a in adaptations if a.get("type") == "material_substitution" ) return total_savings * 100 def _adjust_sample_size( self, protocol: ExperimentalProtocol, available_budget: float ) -> Dict[str, Any]: \"\"\"Ajuste taille d'échantillon en fonction du budget\"\"\" # Cost per participant (simplified) cost_per_participant = (available_budget or 15000) / 50 max_participants = int(available_budget / cost_per_participant) if available_budget else 50 return { "type": "sample_size_adjustment", "cost_per_participant": cost_per_participant, "maximum_sample_size": max_participants, "statistical_power_impact": "Reduced to ~70-75% if N < 40", "recommendation": f"Prioritize primary outcome; accept reduced power for secondary" } def _is_parallelizable(self, step: Any) -> bool: \"\"\"Détermine si étape peut être parallélisée\"\"\" non_parallelizable = [ "randomization", "baseline measurements", "data analysis" ] return not any(term in step.description.lower() for term in non_parallelizable) def _generate_recommendations( self, optimization: Dict, constraints: Dict ) -> List[str]: \"\"\"Généère recommandations basées sur optimisations\"\"\" recommendations = [] if optimization["validity_impact_score"] < 0.85: recommendations.append( "WARNING: Proposed adaptations reduce protocol validity. " "Consider: (1) requesting additional resources, (2) narrowing research scope, " "(3) using pilot study for feasibility validation" ) if len(optimization["adaptations"]) > 3: recommendations.append( "Multiple constraint adaptations required. Protocol fundamentally changed. " "Recommend consulting with methodology expert/biostatistician." ) recommendations.extend([ "Validate all material substitutions with pilot data before main study", "Pre-register protocol adaptations before study start", "Monitor validity indicators during conduct; adjust if needed" ]) return recommendations