AMCNOTIFY-3 / src /model_predictor.py
Abhi10000's picture
Update src/model_predictor.py (#4)
4897445 verified
import datetime # Added full datetime module import
import logging
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def predict_next_service(last_service_date, usage_hours, days_per_week, equipment_type):
"""
Predict the next AMC (Annual Maintenance Contract) date based on usage and equipment type.
Args:
last_service_date (datetime.date): The date of the last service.
usage_hours (int/float): Average usage hours per day.
days_per_week (int): Number of operational days per week.
equipment_type (str): Type of equipment (e.g., 'Analyzer', 'Centrifuge', 'Incubator').
Returns:
datetime.date: The predicted next service date.
Raises:
ValueError: If inputs are invalid (e.g., negative usage hours, invalid equipment type).
"""
try:
# Validate inputs
if not isinstance(last_service_date, datetime.date):
raise ValueError("last_service_date must be a datetime.date object")
if not isinstance(usage_hours, (int, float)) or usage_hours <= 0:
raise ValueError("usage_hours must be a positive number")
if not isinstance(days_per_week, int) or not 1 <= days_per_week <= 7:
raise ValueError("days_per_week must be an integer between 1 and 7")
if not isinstance(equipment_type, str):
raise ValueError("equipment_type must be a string")
# Log input parameters
logger.info(
f"Predicting next service: last_service_date={last_service_date}, "
f"usage_hours={usage_hours}, days_per_week={days_per_week}, equipment_type={equipment_type}"
)
# Usage intensity score
usage_score = usage_hours * days_per_week
# Base interval in days
base_interval = 180 # Standard maintenance interval (6 months)
# Adjust interval based on usage
if usage_score >= 120:
interval = 90 # High usage: 3 months
elif usage_score >= 80:
interval = 120 # Medium-high usage: 4 months
elif usage_score >= 40:
interval = 150 # Medium usage: 5 months
else:
interval = 180 # Low usage: 6 months
# Equipment type adjustments (days to add/subtract from interval)
equipment_adjustments = {
"Analyzer": -15, # More sensitive, service sooner
"Centrifuge": -15, # More sensitive, service sooner
"Incubator": 15, # More stable, service later
"Generic": 0 # No adjustment for generic equipment
}
# Apply equipment type adjustment
adjustment = equipment_adjustments.get(equipment_type, 0)
if adjustment == 0 and equipment_type not in equipment_adjustments:
logger.warning(f"Unknown equipment type '{equipment_type}', using default adjustment (0 days)")
interval += adjustment
# Ensure interval is positive
if interval <= 0:
raise ValueError(f"Calculated interval ({interval} days) is invalid; check equipment type adjustments")
# Calculate next service date
next_date = last_service_date + datetime.timedelta(days=interval) # Updated to use datetime.timedelta
logger.info(f"Predicted next service date: {next_date}")
return next_date
except Exception as e:
logger.error(f"Error in predict_next_service: {e}")
raise