agora/contoso_motors/src/mqtt-simulator/data-emulator.py [128:199]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def get_shift(current_date_time): for shift in production_shifts: if current_date_time.hour >= shift['utc_start_hour'] and current_date_time.hour <= shift['utc_end_hour']: return shift['shift'] # produce this randomly with random delay between 1 to 3 hours def simulate_equipment_maintenance(current_time): return { "equipment_id": "WLD-001", # Pick random equipment "maintenance_date": str(current_time.date), "start_time": "10:21", # Produce random "end_time": "12:30", # Add random duration "type": "routine_check", "notes": "All systems operational, no issues found." } # Produce equipment telemetry every 30 seconds or a minute def simulate_equipment_telemetry(current_time): # Convert time string current_time = str(current_time) equipment_telemetry = [ # Assembly Robot { "date_time": current_time, "equipment_id": "EQ-001", "type": "assembly_robot", "status": random.choice(["operational", "maintenance_required"]), # Instead of random, use periodic maintenance "operational_time_hours": random.uniform(10, 12), "cycles_completed": random.randint(3400, 3500), "efficiency": random.uniform(93, 97), "maintenance_alert": random.choice(["none", "scheduled_check", "urgent_maintenance_required"]), # Instead of random, use periodic maintenance alerts "last_maintenance": "2024-02-20", "next_scheduled_maintenance": "2024-04-01" }, # Conveyor Belt { "date_time": current_time, "equipment_id": "EQ-002", "type": "conveyor_belt", "status": random.choice(["operational", "maintenance_required"]), "operational_time_hours": random.uniform(10, 12), "distance_covered_meters": random.randint(10000, 12000), "efficiency": random.uniform(98, 99), "maintenance_alert": random.choice(["none", "scheduled_check"]), "last_maintenance": "2024-03-15", "next_scheduled_maintenance": "2024-03-30" }, # Paint Station { "date_time": current_time, "equipment_id": "EQ-003", "type": "paint_station", "status": random.choice(["operational", "maintenance_required"]), "operational_time_hours": random.uniform(7, 9), "units_processed": random.randint(400, 500), "efficiency": random.uniform(90, 93), "maintenance_alert": random.choice(["none", "urgent_maintenance_required"]), "last_maintenance": "2024-02-25", "next_scheduled_maintenance": "Overdue" }, # Welding-Assembly Robot { "date_time": current_time, "equipment_id": "EQ-004", "status": random.choice(["operational", "maintenance_required"]), "operational_time_hours": random.uniform(10, 12), "cycles_completed": random.randint(3400, 3500), "efficiency": random.uniform(94, 96), "maintenance_alert": random.choice(["none", "scheduled_check"]), "last_maintenance": "2024-03-20", "next_scheduled_maintenance": "2024-04-01", - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - agora/contoso_motors/src/mqtt-simulator/mqtt_simulator.py [274:345]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - def get_shift(current_date_time): for shift in production_shifts: if current_date_time.hour >= shift['utc_start_hour'] and current_date_time.hour <= shift['utc_end_hour']: return shift['shift'] # produce this randomly with random delay between 1 to 3 hours def simulate_equipment_maintenance(current_time): return { "equipment_id": "WLD-001", # Pick random equipment "maintenance_date": str(current_time.date), "start_time": "10:21", # Produce random "end_time": "12:30", # Add random duration "type": "routine_check", "notes": "All systems operational, no issues found." } # Produce equipment telemetry every 30 seconds or a minute def simulate_equipment_telemetry(current_time): # Convert time string current_time = str(current_time) equipment_telemetry = [ # Assembly Robot { "date_time": current_time, "equipment_id": "EQ-001", "type": "assembly_robot", "status": random.choice(["operational", "maintenance_required"]), # Instead of random, use periodic maintenance "operational_time_hours": random.uniform(10, 12), "cycles_completed": random.randint(3400, 3500), "efficiency": random.uniform(93, 97), "maintenance_alert": random.choice(["none", "scheduled_check", "urgent_maintenance_required"]), # Instead of random, use periodic maintenance alerts "last_maintenance": "2024-02-20", "next_scheduled_maintenance": "2024-04-01" }, # Conveyor Belt { "date_time": current_time, "equipment_id": "EQ-002", "type": "conveyor_belt", "status": random.choice(["operational", "maintenance_required"]), "operational_time_hours": random.uniform(10, 12), "distance_covered_meters": random.randint(10000, 12000), "efficiency": random.uniform(98, 99), "maintenance_alert": random.choice(["none", "scheduled_check"]), "last_maintenance": "2024-03-15", "next_scheduled_maintenance": "2024-03-30" }, # Paint Station { "date_time": current_time, "equipment_id": "EQ-003", "type": "paint_station", "status": random.choice(["operational", "maintenance_required"]), "operational_time_hours": random.uniform(7, 9), "units_processed": random.randint(400, 500), "efficiency": random.uniform(90, 93), "maintenance_alert": random.choice(["none", "urgent_maintenance_required"]), "last_maintenance": "2024-02-25", "next_scheduled_maintenance": "Overdue" }, # Welding-Assembly Robot { "date_time": current_time, "equipment_id": "EQ-004", "status": random.choice(["operational", "maintenance_required"]), "operational_time_hours": random.uniform(10, 12), "cycles_completed": random.randint(3400, 3500), "efficiency": random.uniform(94, 96), "maintenance_alert": random.choice(["none", "scheduled_check"]), "last_maintenance": "2024-03-20", "next_scheduled_maintenance": "2024-04-01", - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -