File size: 7,827 Bytes
baf834b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 | """
Production Scheduling Adjustment Module
Author: Antigravity AI
Date: August 2026
This module handles closed-loop production schedule adjustments.
It evaluates a production schedule against ML predictions (failure probabilities
and Remaining Useful Life), identifies at-risk assets, and dynamically reschedules/reallocates
jobs to healthy backup machines of the same model.
"""
import os
import logging
import pandas as pd
try:
from utils.project_paths import LOGS_DIR
except ImportError:
from project_paths import LOGS_DIR
# Setup logging
os.makedirs(LOGS_DIR, exist_ok=True)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.StreamHandler(),
logging.FileHandler(os.path.join(LOGS_DIR, "scheduler.log"))
]
)
logger = logging.getLogger(__name__)
def generate_mock_schedule(output_path: str):
"""Generate a realistic baseline production schedule for an FMCG manufacturing plant."""
logger.info(f"Generating mock baseline production schedule at {output_path}...")
# 50 production jobs running on machines 1 to 20
jobs = []
for i in range(1, 51):
mid = (i % 20) + 1 # machines 1 to 20
job_id = f"JOB_FMCG_{i:03d}"
duration = float((i * 7) % 24 + 4) # 4 to 27 hours
priority = "High" if (i % 3 == 0) else "Medium" if (i % 3 == 1) else "Low"
jobs.append({
'job_id': job_id,
'machine_id': mid,
'duration_hours': duration,
'priority': priority,
'status': 'Scheduled',
'notes': 'Original schedule assignment.'
})
df = pd.DataFrame(jobs)
os.makedirs(os.path.dirname(output_path), exist_ok=True)
df.to_csv(output_path, index=False)
logger.info("Mock schedule created successfully.")
class ProductionScheduler:
def __init__(self, schedule_path: str, machines_metadata_path: str):
self.schedule_path = schedule_path
self.machines_path = machines_metadata_path
if not os.path.exists(self.schedule_path):
generate_mock_schedule(self.schedule_path)
if not os.path.exists(self.machines_path):
raise FileNotFoundError(f"Missing machine metadata file: {self.machines_path}")
self.schedule = pd.read_csv(self.schedule_path)
self.machines = pd.read_csv(self.machines_path)
def adjust_schedule(self, predictions: dict, output_adjusted_path: str,
failure_threshold: float = 0.7, min_rul_hours: float = 48.0) -> pd.DataFrame:
"""
Adjust schedule by rerouting jobs from at-risk machines to healthy backup machines.
Parameters:
- predictions: dict like { machine_id: { 'failure_probability': float, 'rul': float } }
- output_adjusted_path: file path to save the modified schedule.
"""
logger.info("Starting production schedule adjustment run...")
adjusted_df = self.schedule.copy()
# Build machine health lookup maps
at_risk_machines = set()
healthy_machines = {} # model -> list of (machine_id, failure_prob, RUL)
# Initialize model groups from machines metadata
model_groups = {}
for _, row in self.machines.iterrows():
mid = row['machineID']
model = row['model']
model_groups[mid] = model
for _, row in self.machines.iterrows():
mid = row['machineID']
model = row['model']
# Retrieve predictions (default to healthy if no predictions available)
pred = predictions.get(mid, {'failure_probability': 0.0, 'rul': 500.0})
f_prob = pred.get('failure_probability', 0.0)
rul = pred.get('rul', 500.0)
# Determine if machine is at risk
is_risk = False
reasons = []
if f_prob > failure_threshold:
is_risk = True
reasons.append(f"high failure probability ({f_prob*100:.1f}%)")
if rul is not None and rul < min_rul_hours:
is_risk = True
reasons.append(f"low remaining useful life ({rul:.1f} hours)")
if is_risk:
at_risk_machines.add(mid)
logger.warning(f"Machine {mid} ({model}) is AT RISK: {', '.join(reasons)}.")
else:
if model not in healthy_machines:
healthy_machines[model] = []
healthy_machines[model].append((mid, f_prob, rul))
# Sort healthy backup machines by lowest failure probability, then highest RUL
for model in healthy_machines:
healthy_machines[model].sort(key=lambda x: (x[1], -x[2] if x[2] is not None else 0))
reassigned_count = 0
# Round-robin counters per model for load balancing
rr_counters = {model: 0 for model in healthy_machines}
# Adjust assignments
for idx, row in adjusted_df.iterrows():
mid = row['machine_id']
if mid in at_risk_machines:
model = model_groups.get(mid, 'Unknown')
backups = healthy_machines.get(model, [])
if backups:
# Round-robin: pick next healthy machine in rotation
rr_idx = rr_counters.get(model, 0) % len(backups)
backup_id, backup_f_prob, backup_rul = backups[rr_idx]
rr_counters[model] = rr_idx + 1
adjusted_df.at[idx, 'machine_id'] = backup_id
adjusted_df.at[idx, 'status'] = 'Rescheduled'
adjusted_df.at[idx, 'notes'] = (
f"Reallocated from at-risk Machine {mid} to healthy backup Machine {backup_id} "
f"(Model: {model}). Risk mitigation active."
)
logger.info(f"Job {row['job_id']} reassigned from Machine {mid} to Machine {backup_id}.")
reassigned_count += 1
else:
adjusted_df.at[idx, 'status'] = 'Risk Alert'
adjusted_df.at[idx, 'notes'] = (
f"Machine {mid} is at risk, but no healthy backup machines of model {model} were available!"
)
logger.error(f"Failed to reassign Job {row['job_id']}: No healthy backup machines available for model {model}!")
os.makedirs(os.path.dirname(output_adjusted_path), exist_ok=True)
adjusted_df.to_csv(output_adjusted_path, index=False)
logger.info(f"Production schedule adjustment complete. Reallocated {reassigned_count} jobs.")
logger.info(f"Adjusted schedule saved to {output_adjusted_path}")
return adjusted_df
if __name__ == '__main__':
# Test block
base_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
sch_path = os.path.join(base_dir, 'data', 'raw', 'production_schedule.csv')
mach_path = os.path.join(base_dir, 'data', 'raw', 'PdM_machines.csv')
adj_path = os.path.join(base_dir, 'data', 'processed', 'adjusted_schedule.csv')
# Mock some machine risk predictions
mock_preds = {
3: {'failure_probability': 0.85, 'rul': 12.0}, # At-risk
8: {'failure_probability': 0.12, 'rul': 350.0}, # Healthy
12: {'failure_probability': 0.92, 'rul': 4.0} # At-risk
}
scheduler = ProductionScheduler(sch_path, mach_path)
scheduler.adjust_schedule(mock_preds, adj_path)
|