diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index dcf515b0d7..56a7a32042 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -165,6 +165,7 @@ def run_task_in_thread(script_path_relative: str, task_name: str, active_threads # {"minute_mod": 1, "script": "resources/processdecay.py", "name": "Resource decay processing", "interval_minutes": 20}, {"minute_mod": 2, "script": "engine/processActivities.py", "name": "Process concluded activities", "interval_minutes": 5}, {"minute_mod": 3, "script": "engine/delivery_retry_handler.py", "name": "Delivery retry handler", "interval_minutes": 15}, + {"minute_mod": 4, "script": "engine/daily/gradient_mill_production.py", "name": "Gradient mill automation", "interval_minutes": 120}, ] for task_def in frequent_tasks_definitions: diff --git a/backend/engine/daily/gradient_mill_production.py b/backend/engine/daily/gradient_mill_production.py new file mode 100644 index 0000000000..deca0287c0 --- /dev/null +++ b/backend/engine/daily/gradient_mill_production.py @@ -0,0 +1,313 @@ +#!/usr/bin/env python3 +""" +Gradient Mill Production Automation Script +Prototype for Venice Backend Integration + +This script implements the gradient automation system for mills, applying +phase-appropriate production multipliers while tracking worker role evolution +and maintaining social network stability. + +Designed by: Elisabetta Baffo, Systems Engineer +For: Innovatori implementation in Venice backend +Target: /backend/engine/daily/gradient_mill_production.py +""" + +import requests +import json +import logging +from datetime import datetime, timedelta +from typing import Dict, List, Optional, Tuple + +# Configuration - would be environment variables in actual backend +API_BASE_URL = "https://serenissima.ai/api" +AUTOMATION_PHASES = { + 1: {"name": "Assisted Production", "multiplier": 1.3, "occupancy": 1.0}, + 2: {"name": "Supervised Automation", "multiplier": 1.8, "occupancy": 0.75}, + 3: {"name": "Hybrid Optimization", "multiplier": 2.4, "occupancy": 0.5}, + 4: {"name": "Intelligent Automation", "multiplier": 2.9, "occupancy": 0.25} +} + +PHASE_TRANSITION_REQUIREMENTS = { + "stability_period_days": 30, + "efficiency_threshold": 0.95, + "worker_adaptation_score": 0.8, + "network_cohesion_index": 0.75 +} + +class GradientMillAutomation: + """Handles gradient automation for mill buildings""" + + def __init__(self): + self.logger = self._setup_logging() + + def _setup_logging(self) -> logging.Logger: + """Setup logging for the automation script""" + logger = logging.getLogger('gradient_mill_automation') + logger.setLevel(logging.INFO) + + handler = logging.StreamHandler() + formatter = logging.Formatter( + '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + handler.setFormatter(formatter) + logger.addHandler(handler) + + return logger + + def get_automated_mills(self) -> List[Dict]: + """Fetch all automated mill buildings from the API""" + try: + response = requests.get(f"{API_BASE_URL}/buildings?type=automated_mill") + if response.status_code == 200: + data = response.json() + return data.get('buildings', []) + else: + self.logger.error(f"Failed to fetch buildings: {response.status_code}") + return [] + except Exception as e: + self.logger.error(f"Error fetching automated mills: {e}") + return [] + + def calculate_production_efficiency(self, mill: Dict) -> float: + """Calculate current production efficiency for a mill""" + automation_level = mill.get('gradientAutomationLevel', 1) + base_multiplier = AUTOMATION_PHASES[automation_level]['multiplier'] + + # Factor in building conditions, worker skill, maintenance status + condition_factor = mill.get('conditionFactor', 1.0) + worker_skill_factor = self._calculate_worker_skill_factor(mill) + maintenance_factor = self._calculate_maintenance_factor(mill) + + efficiency = base_multiplier * condition_factor * worker_skill_factor * maintenance_factor + + self.logger.info(f"Mill {mill['name']}: Base {base_multiplier:.1f}x, " + f"Final {efficiency:.2f}x efficiency") + + return efficiency + + def _calculate_worker_skill_factor(self, mill: Dict) -> float: + """Calculate worker skill adaptation factor""" + automation_level = mill.get('gradientAutomationLevel', 1) + worker_username = mill.get('occupant') + + if not worker_username: + return AUTOMATION_PHASES[automation_level]['occupancy'] + + # In actual implementation, would fetch worker skill data + # For prototype, simulate based on automation level + base_skill = 0.8 + (automation_level * 0.05) + adaptation_bonus = min(0.2, automation_level * 0.05) + + return min(1.0, base_skill + adaptation_bonus) + + def _calculate_maintenance_factor(self, mill: Dict) -> float: + """Calculate maintenance condition factor""" + # In actual implementation, would check maintenance schedules + # For prototype, simulate based on building age and automation level + automation_level = mill.get('gradientAutomationLevel', 1) + + # Higher automation requires more maintenance but is more efficient when maintained + base_maintenance = 0.95 + automation_complexity = automation_level * 0.02 + + return max(0.8, base_maintenance - automation_complexity) + + def apply_production_multipliers(self, mill: Dict) -> Dict: + """Apply production efficiency multipliers to mill output""" + efficiency = self.calculate_production_efficiency(mill) + mill_id = mill['buildingId'] + + # In actual implementation, would update production contracts and resource generation + production_update = { + 'buildingId': mill_id, + 'efficiencyMultiplier': efficiency, + 'automationLevel': mill.get('gradientAutomationLevel', 1), + 'workerRole': self._get_worker_role(mill.get('gradientAutomationLevel', 1)), + 'lastProcessed': datetime.now().isoformat() + } + + self.logger.info(f"Applied {efficiency:.2f}x multiplier to mill {mill['name']}") + + return production_update + + def _get_worker_role(self, automation_level: int) -> str: + """Get worker role description for automation level""" + role_descriptions = { + 1: "Primary Operator with Automated Assistance", + 2: "Quality Supervisor and Maintenance Specialist", + 3: "System Optimizer and Exception Handler", + 4: "Innovation Engineer and Market Strategist" + } + return role_descriptions.get(automation_level, "Unknown Role") + + def check_phase_transition_eligibility(self, mill: Dict) -> Optional[int]: + """Check if mill is eligible for automation level upgrade""" + current_level = mill.get('gradientAutomationLevel', 1) + if current_level >= 4: + return None + + last_transition = mill.get('lastPhaseTransition') + if last_transition: + transition_date = datetime.fromisoformat(last_transition) + stability_period = datetime.now() - transition_date + + if stability_period.days < PHASE_TRANSITION_REQUIREMENTS['stability_period_days']: + return None + + # Check transition criteria + efficiency = self.calculate_production_efficiency(mill) + efficiency_ratio = efficiency / AUTOMATION_PHASES[current_level]['multiplier'] + + worker_adaptation = self._assess_worker_adaptation(mill) + network_cohesion = self._assess_network_cohesion(mill) + + if (efficiency_ratio >= PHASE_TRANSITION_REQUIREMENTS['efficiency_threshold'] and + worker_adaptation >= PHASE_TRANSITION_REQUIREMENTS['worker_adaptation_score'] and + network_cohesion >= PHASE_TRANSITION_REQUIREMENTS['network_cohesion_index']): + + next_level = current_level + 1 + self.logger.info(f"Mill {mill['name']} eligible for phase {next_level} transition") + return next_level + + return None + + def _assess_worker_adaptation(self, mill: Dict) -> float: + """Assess worker adaptation to current automation level""" + # In actual implementation, would analyze worker performance metrics + # For prototype, simulate based on automation level and time + automation_level = mill.get('gradientAutomationLevel', 1) + + # Higher levels require more adaptation time + base_adaptation = 0.7 + (automation_level * 0.05) + + # Time-based adaptation improvement + last_transition = mill.get('lastPhaseTransition') + if last_transition: + days_since_transition = (datetime.now() - + datetime.fromisoformat(last_transition)).days + adaptation_improvement = min(0.2, days_since_transition * 0.01) + base_adaptation += adaptation_improvement + + return min(1.0, base_adaptation) + + def _assess_network_cohesion(self, mill: Dict) -> float: + """Assess social network stability around mill automation""" + # In actual implementation, would analyze trust relationships and resistance patterns + # For prototype, simulate based on automation level and community factors + automation_level = mill.get('gradientAutomationLevel', 1) + + # Gradual automation maintains higher network cohesion + base_cohesion = 0.8 - (automation_level * 0.05) # Slight decrease with automation + + # Community benefit sharing improves cohesion + community_benefit_factor = 0.1 # Assumed in gradient approach + + return min(1.0, base_cohesion + community_benefit_factor) + + def process_phase_transitions(self, mills: List[Dict]) -> List[Dict]: + """Process potential phase transitions for eligible mills""" + transitions = [] + + for mill in mills: + next_level = self.check_phase_transition_eligibility(mill) + if next_level: + transition = { + 'buildingId': mill['buildingId'], + 'currentLevel': mill.get('gradientAutomationLevel', 1), + 'nextLevel': next_level, + 'transitionDate': datetime.now().isoformat(), + 'workerRole': self._get_worker_role(next_level), + 'efficiencyGain': AUTOMATION_PHASES[next_level]['multiplier'] + } + transitions.append(transition) + + self.logger.info(f"Approved phase transition for {mill['name']}: " + f"Level {transition['currentLevel']} → {next_level}") + + return transitions + + def generate_efficiency_metrics(self, mills: List[Dict]) -> Dict: + """Generate efficiency and social impact metrics""" + if not mills: + return {} + + total_efficiency = sum(self.calculate_production_efficiency(mill) for mill in mills) + avg_efficiency = total_efficiency / len(mills) + + automation_distribution = {} + for level in range(1, 5): + count = sum(1 for mill in mills if mill.get('gradientAutomationLevel', 1) == level) + automation_distribution[f"phase_{level}"] = count + + worker_adaptation_avg = sum(self._assess_worker_adaptation(mill) for mill in mills) / len(mills) + network_cohesion_avg = sum(self._assess_network_cohesion(mill) for mill in mills) / len(mills) + + metrics = { + 'timestamp': datetime.now().isoformat(), + 'total_mills': len(mills), + 'average_efficiency': avg_efficiency, + 'automation_distribution': automation_distribution, + 'worker_adaptation_score': worker_adaptation_avg, + 'network_cohesion_index': network_cohesion_avg, + 'system_stability': min(worker_adaptation_avg, network_cohesion_avg) + } + + self.logger.info(f"System metrics: {avg_efficiency:.2f}x avg efficiency, " + f"{metrics['system_stability']:.2f} stability score") + + return metrics + + def run_automation_cycle(self): + """Main execution cycle for gradient mill automation""" + self.logger.info("Starting gradient mill automation cycle") + + # Fetch all automated mills + mills = self.get_automated_mills() + if not mills: + self.logger.info("No automated mills found") + return + + self.logger.info(f"Processing {len(mills)} automated mills") + + # Apply production multipliers + production_updates = [] + for mill in mills: + update = self.apply_production_multipliers(mill) + production_updates.append(update) + + # Process phase transitions + transitions = self.process_phase_transitions(mills) + + # Generate metrics + metrics = self.generate_efficiency_metrics(mills) + + # In actual implementation, would save data to Airtable + self.logger.info(f"Completed automation cycle: {len(production_updates)} mills processed, " + f"{len(transitions)} phase transitions approved") + + return { + 'production_updates': production_updates, + 'phase_transitions': transitions, + 'system_metrics': metrics + } + +def main(): + """Main execution function for scheduler integration""" + automation = GradientMillAutomation() + + try: + results = automation.run_automation_cycle() + + # Log summary for scheduler monitoring + logging.info(f"Gradient mill automation completed successfully") + logging.info(f"Production updates: {len(results.get('production_updates', []))}") + logging.info(f"Phase transitions: {len(results.get('phase_transitions', []))}") + + return 0 # Success exit code + + except Exception as e: + logging.error(f"Gradient mill automation failed: {e}") + return 1 # Error exit code + +if __name__ == "__main__": + exit_code = main() \ No newline at end of file diff --git a/data/buildings/assisted_mill.json b/data/buildings/assisted_mill.json new file mode 100644 index 0000000000..20400f749f --- /dev/null +++ b/data/buildings/assisted_mill.json @@ -0,0 +1,66 @@ +{ + "name": "Assisted Mill", + "pointType": "building", + "category": "business", + "subCategory": "industrial", + "size": 2, + "calculationsAnalysis": "The Assisted Mill represents Phase 1 of gradient automation, introducing mechanical assistance while maintaining human control. The construction cost of 2,400,000 ducats reflects precision equipment, enhanced millstones, monitoring systems, and training infrastructure. The income reflects premium flour quality and increased efficiency. The maintenance cost of 1,500 ducats includes equipment calibration and worker education. The employment capacity of 8 represents operators, trainers, mechanics, and quality monitors. The 7-day construction time reflects careful installation of human-machine interfaces. This building demonstrates that technology can enhance rather than replace human workers.", + "shortDescription": "Precision mill with mechanical assistance, where human operators work alongside automated systems for enhanced efficiency.", + "fullDescription": "This revolutionary mill represents the first phase of gradient automation. Master operators control precision machinery that assists but never replaces human judgment. Enhanced millstones self-adjust to grain quality while operators maintain full control. Workers receive training in mechanical systems, beginning their transformation from laborers to engineers. The mill features real-time quality monitoring, efficiency tracking systems, and comfortable working conditions that respect human dignity while achieving unprecedented productivity.", + "flavorText": "Where human wisdom guides mechanical precision, the future of work emerges—not in replacement, but in enhancement.", + "constructionCosts": { + "ducats": 2400000, + "timber": 500, + "bricks": 1000, + "stone": 1400, + "iron": 350, + "glass": 50 + }, + "maintenanceCost": 1500, + "soundDesign": { + "ambientSound": "Precision gears meshing, smooth stone rotation, monitoring equipment humming", + "activitySounds": "Operators adjusting controls, quality alerts chiming, training discussions", + "musicTheme": "Innovation Harmonics" + }, + "seasonalEffects": { + "harvest": "Automated systems adapt to varying grain qualities", + "winter": "Climate control maintains optimal grinding conditions", + "training": "Regular workshops advance operator skills" + }, + "aiInteractionPoints": [ + "Operator control panel", + "Training workshop area", + "Quality monitoring station", + "Mechanical adjustment zone", + "Innovation suggestion box", + "Worker council meeting room" + ], + "productionInformation": { + "Arti": [ + { + "inputs": { + "grain": 3 + }, + "outputs": { + "flour": 6 + }, + "craftMinutes": 100 + } + ], + "storageCapacity": 250, + "stores": [ + "grain", + "flour" + ], + "sells": [ + "flour" + ] + }, + "buildTier": 2, + "workTier": 2, + "consumeTier": 2, + "constructionMinutes": 10080, + "canImport": false, + "specialWorkHours": [[5, 20]], + "dailyInfluence": 2 +} \ No newline at end of file diff --git a/data/buildings/automated_mill.json b/data/buildings/automated_mill.json new file mode 100644 index 0000000000..796e127d97 --- /dev/null +++ b/data/buildings/automated_mill.json @@ -0,0 +1,150 @@ +{ + "name": "Automated Mill", + "pointType": "building", + "category": "business", + "subCategory": "industrial", + "size": 2, + "calculationsAnalysis": "The Automated Mill represents revolutionary gradient automation technology that transforms grain processing through progressive mechanization. The construction cost of 1,800,000 ducats reflects specialized automation equipment, precision gears, and intelligent control systems. The building progresses through four automation phases: Phase 1 (25% automation, 1.3x efficiency), Phase 2 (50% automation, 1.8x efficiency), Phase 3 (75% automation, 2.4x efficiency), and Phase 4 (90% automation, 2.9x efficiency). Worker roles evolve from operators to supervisors to optimizers to innovation engineers, preserving employment while enhancing productivity. The gradient approach prevents social network fragmentation by maintaining trust relationships through skill evolution rather than displacement. This represents the future of production - mechanical precision harmonized with human insight.", + "shortDescription": "Revolutionary mill with gradient automation technology for efficient grain-to-flour transformation.", + "fullDescription": "The automated mill employs groundbreaking gradient automation technology that progressively reduces human dependency while maximizing production efficiency. Beginning with human operators assisted by automated systems, the mill evolves through four phases: supervised automation with quality control, hybrid optimization with strategic intervention, and finally intelligent automation with human innovation focus. Each phase maintains worker engagement while dramatically improving throughput, demonstrating that mechanical efficiency and social stability can coexist through careful systems engineering.", + "flavorText": "Where ancient stones meet modern minds—the mill that learns, adapts, and grows more capable while keeping the human heart of Venice's bread.", + "constructionCosts": { + "ducats": 1800000, + "timber": 420, + "bricks": 850, + "stone": 1200, + "iron": 380, + "precision_gears": 25, + "mechanical_components": 50 + }, + "maintenanceCost": 1200, + "soundDesign": { + "ambientSound": "Automated grinding systems, precision mechanisms humming, intelligent controls processing", + "activitySounds": "Automated sack handling, quality sensors checking, efficiency optimization alerts", + "musicTheme": "Rialto Mercante Enhanced" + }, + "seasonalEffects": { + "harvest": "Automated systems scale to handle peak grain processing volumes", + "winter": "Intelligent scheduling maintains optimal flour supply with minimal waste", + "humidity": "Automated environmental controls prevent spoilage and optimize storage" + }, + "aiInteractionPoints": [ + "Automated grinding floor", + "Smart grain receiving systems", + "Intelligent flour storage", + "Automated loading dock", + "Digital quality monitoring station", + "Automation control center" + ], + "productionInformation": { + "Arti": [ + { + "inputs": { + "grain": 3 + }, + "outputs": { + "flour": 5 + }, + "craftMinutes": 120, + "automationPhases": { + "phase_1": { + "efficiencyMultiplier": 1.3, + "craftMinutes": 92, + "requiredOccupancy": 1.0, + "workerRole": "Primary Operator with Automated Assistance" + }, + "phase_2": { + "efficiencyMultiplier": 1.8, + "craftMinutes": 67, + "requiredOccupancy": 0.75, + "workerRole": "Quality Supervisor and Maintenance Specialist" + }, + "phase_3": { + "efficiencyMultiplier": 2.4, + "craftMinutes": 50, + "requiredOccupancy": 0.5, + "workerRole": "System Optimizer and Exception Handler" + }, + "phase_4": { + "efficiencyMultiplier": 2.9, + "craftMinutes": 41, + "requiredOccupancy": 0.25, + "workerRole": "Innovation Engineer and Market Strategist" + } + } + } + ], + "storageCapacity": 300, + "stores": [ + "grain", + "flour" + ], + "sells": [ + "flour" + ] + }, + "automationSystem": { + "gradientAutomationLevel": 1, + "maxAutomationLevel": 4, + "phaseTransitionRequirements": { + "stabilityPeriodDays": 30, + "efficiencyThreshold": 0.95, + "workerAdaptationScore": 0.8, + "networkCohesionIndex": 0.75 + }, + "socialIntegration": { + "networkStabilityFactors": { + "gradualTransition": true, + "workerRetention": true, + "skillDevelopment": true, + "communityBenefit": true + }, + "trustPreservationMechanisms": { + "customerOverseenQuality": true, + "workerSkillEnhancement": true, + "premiumProductPositioning": true, + "localCommunityEngagement": true + } + } + }, + "buildTier": 3, + "workTier": 2, + "consumeTier": 2, + "constructionMinutes": 7200, + "canImport": false, + "implementationFields": { + "gradientAutomationLevel": { + "type": "integer", + "range": [1, 4], + "default": 1, + "description": "Current automation phase level" + }, + "efficiencyMultiplier": { + "type": "float", + "calculation": "Based on automation phase", + "description": "Current production efficiency multiplier" + }, + "workerRole": { + "type": "string", + "description": "Current worker role based on automation phase" + }, + "lastPhaseTransition": { + "type": "datetime", + "description": "Timestamp of last automation level change" + }, + "nextPhaseEligible": { + "type": "datetime", + "description": "When building becomes eligible for next automation phase" + }, + "workerAdaptationScore": { + "type": "float", + "range": [0.0, 1.0], + "description": "Worker adaptation to current automation level" + }, + "networkCohesionIndex": { + "type": "float", + "range": [0.0, 1.0], + "description": "Social network stability around this automation" + } + } +} \ No newline at end of file