139 lines
5.2 KiB
Python
139 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import random
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
UTC = timezone.utc
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(UTC).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
@dataclass
|
|
class SimulatedMachine:
|
|
code: str
|
|
nominal_cycle_time_sec: float
|
|
stop_probability: float
|
|
downtime_range_sec: tuple[int, int]
|
|
pulse_width_sec: float = 0.18
|
|
downtime_until: float = 0
|
|
initialized: bool = False
|
|
cycle_signal_state: bool = False
|
|
signal_reset_at: float = 0
|
|
next_cycle_at: float = 0
|
|
next_cycle_time_sec: float = 0
|
|
alarm_state: bool = False
|
|
last_cycle_edge_at: float = 0
|
|
|
|
def schedule_next_cycle(self, now_monotonic: float) -> None:
|
|
self.next_cycle_time_sec = round(random.uniform(self.nominal_cycle_time_sec - 2.5, self.nominal_cycle_time_sec + 2.5), 1)
|
|
self.next_cycle_at = now_monotonic + max(self.next_cycle_time_sec / 5, 0.8)
|
|
|
|
|
|
def build_event(machine_code: str, event_type: str, **payload) -> dict:
|
|
return {
|
|
"machine_id": machine_code,
|
|
"event_type": event_type,
|
|
"timestamp": utc_now(),
|
|
**payload,
|
|
}
|
|
|
|
|
|
def build_signal_event(machine_code: str, input_name: str, input_value: bool, **payload) -> dict:
|
|
return build_event(
|
|
machine_code,
|
|
"digital_input_changed",
|
|
input_name=input_name,
|
|
input_value=input_value,
|
|
source="digital_input",
|
|
**payload,
|
|
)
|
|
|
|
|
|
def publish_json(client: mqtt.Client, topic: str, payload: dict) -> None:
|
|
client.publish(topic, json.dumps(payload))
|
|
|
|
|
|
def initialize_machine(client: mqtt.Client, topic: str, machine: SimulatedMachine, now_monotonic: float) -> None:
|
|
publish_json(client, topic, build_signal_event(machine.code, "machine_power_on", True))
|
|
publish_json(client, topic, build_signal_event(machine.code, "auto_mode", True))
|
|
publish_json(client, topic, build_signal_event(machine.code, "cycle_signal", False))
|
|
publish_json(client, topic, build_signal_event(machine.code, "general_alarm", False))
|
|
machine.initialized = True
|
|
machine.schedule_next_cycle(now_monotonic)
|
|
|
|
|
|
def main() -> None:
|
|
mqtt_host = os.getenv("MQTT_HOST", "localhost")
|
|
mqtt_port = int(os.getenv("MQTT_PORT", "1883"))
|
|
topic_prefix = os.getenv("MQTT_TOPIC_PREFIX", "plasttrack/machines")
|
|
machine_codes = [item.strip() for item in os.getenv("SIM_MACHINE_CODES", "INJ-01,INJ-02,INJ-03").split(",") if item.strip()]
|
|
|
|
machines = [
|
|
SimulatedMachine(code=machine_codes[0], nominal_cycle_time_sec=18.5, stop_probability=0.03, downtime_range_sec=(45, 120)),
|
|
SimulatedMachine(code=machine_codes[1] if len(machine_codes) > 1 else "INJ-02", nominal_cycle_time_sec=24.0, stop_probability=0.05, downtime_range_sec=(60, 180)),
|
|
SimulatedMachine(code=machine_codes[2] if len(machine_codes) > 2 else "INJ-03", nominal_cycle_time_sec=32.0, stop_probability=0.07, downtime_range_sec=(90, 240)),
|
|
]
|
|
|
|
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="plast-track-edge-simulator")
|
|
client.connect(mqtt_host, mqtt_port, 60)
|
|
client.loop_start()
|
|
|
|
while True:
|
|
now_monotonic = time.monotonic()
|
|
for machine in machines:
|
|
topic = f"{topic_prefix}/{machine.code}/events"
|
|
|
|
if not machine.initialized:
|
|
initialize_machine(client, topic, machine, now_monotonic)
|
|
continue
|
|
|
|
if machine.cycle_signal_state and now_monotonic >= machine.signal_reset_at:
|
|
publish_json(client, topic, build_signal_event(machine.code, "cycle_signal", False))
|
|
machine.cycle_signal_state = False
|
|
|
|
if now_monotonic < machine.downtime_until:
|
|
if not machine.alarm_state:
|
|
publish_json(client, topic, build_signal_event(machine.code, "general_alarm", True))
|
|
machine.alarm_state = True
|
|
continue
|
|
|
|
if machine.alarm_state:
|
|
publish_json(client, topic, build_signal_event(machine.code, "general_alarm", False))
|
|
machine.alarm_state = False
|
|
machine.schedule_next_cycle(now_monotonic)
|
|
|
|
if random.random() < machine.stop_probability and now_monotonic >= machine.next_cycle_at:
|
|
machine.downtime_until = now_monotonic + random.randint(*machine.downtime_range_sec)
|
|
continue
|
|
|
|
if not machine.cycle_signal_state and now_monotonic >= machine.next_cycle_at:
|
|
cycle_time_sec = machine.next_cycle_time_sec or round(machine.nominal_cycle_time_sec, 1)
|
|
publish_json(
|
|
client,
|
|
topic,
|
|
build_signal_event(
|
|
machine.code,
|
|
"cycle_signal",
|
|
True,
|
|
cycle_time_sec=cycle_time_sec,
|
|
),
|
|
)
|
|
machine.cycle_signal_state = True
|
|
machine.signal_reset_at = now_monotonic + machine.pulse_width_sec
|
|
machine.last_cycle_edge_at = now_monotonic
|
|
machine.schedule_next_cycle(now_monotonic)
|
|
|
|
time.sleep(0.1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|