296 lines
11 KiB
Python
296 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models import Cycle, Downtime, Machine, MachineEvent, ProductionOrder
|
|
from ..schemas import MachineEventIn
|
|
|
|
|
|
REGALAGE_REASON_CODES = {"reglage", "changement_moule"}
|
|
SIGNAL_EVENT_TYPE = "digital_input_changed"
|
|
POWER_INPUT_NAME = "machine_power_on"
|
|
CYCLE_INPUT_NAME = "cycle_signal"
|
|
UTC = timezone.utc
|
|
|
|
|
|
def _ensure_utc(value: datetime) -> datetime:
|
|
return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
|
|
|
|
|
|
def parse_machine_event(payload: str | bytes) -> MachineEventIn:
|
|
raw = payload.decode("utf-8") if isinstance(payload, bytes) else payload
|
|
data = json.loads(raw)
|
|
return MachineEventIn.model_validate(data)
|
|
|
|
|
|
def normalize_signal_value(value: bool | int | str | None) -> bool | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, int):
|
|
return value != 0
|
|
normalized = str(value).strip().lower()
|
|
if normalized in {"1", "true", "on", "high"}:
|
|
return True
|
|
if normalized in {"0", "false", "off", "low"}:
|
|
return False
|
|
return None
|
|
|
|
|
|
def _get_or_create_machine(session: Session, machine_code: str) -> Machine:
|
|
machine = session.execute(select(Machine).where(Machine.code == machine_code)).scalar_one_or_none()
|
|
if machine is None:
|
|
machine = Machine(code=machine_code, name=machine_code, status="hors_planning")
|
|
session.add(machine)
|
|
session.flush()
|
|
return machine
|
|
|
|
|
|
def get_running_order(session: Session, machine_id: int) -> ProductionOrder | None:
|
|
return session.execute(
|
|
select(ProductionOrder).where(
|
|
ProductionOrder.machine_id == machine_id,
|
|
ProductionOrder.status == "running",
|
|
).order_by(ProductionOrder.started_at.desc())
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def get_open_downtime(session: Session, machine_id: int) -> Downtime | None:
|
|
return session.execute(
|
|
select(Downtime).where(
|
|
Downtime.machine_id == machine_id,
|
|
Downtime.end_time.is_(None),
|
|
).order_by(Downtime.start_time.desc())
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def get_last_input_event(session: Session, machine_id: int, input_name: str) -> MachineEvent | None:
|
|
rows = session.execute(
|
|
select(MachineEvent).where(
|
|
MachineEvent.machine_id == machine_id,
|
|
MachineEvent.event_type == SIGNAL_EVENT_TYPE,
|
|
).order_by(MachineEvent.timestamp.desc()).limit(50)
|
|
).scalars()
|
|
for row in rows:
|
|
if row.payload.get("input_name") == input_name:
|
|
return row
|
|
return None
|
|
|
|
|
|
def derive_machine_status(machine: Machine, open_downtime: Downtime | None, running_order: ProductionOrder | None) -> str:
|
|
if not machine.powered_on or running_order is None:
|
|
return "hors_planning"
|
|
if open_downtime is None:
|
|
return "production"
|
|
if open_downtime.reason_code in REGALAGE_REASON_CODES:
|
|
return "reglage"
|
|
if open_downtime.reason_code:
|
|
return "arret_qualifie"
|
|
return "arret_non_qualifie"
|
|
|
|
|
|
def open_downtime_if_needed(session: Session, machine: Machine, event_time: datetime, auto_detected: bool = True) -> Downtime | None:
|
|
running_order = get_running_order(session, machine.id)
|
|
if running_order is None or not machine.powered_on:
|
|
machine.status = "hors_planning"
|
|
machine.status_since = event_time
|
|
return None
|
|
|
|
open_downtime = get_open_downtime(session, machine.id)
|
|
if open_downtime is not None:
|
|
return open_downtime
|
|
|
|
downtime = Downtime(
|
|
machine_id=machine.id,
|
|
production_order_id=running_order.id,
|
|
start_time=event_time,
|
|
auto_detected=auto_detected,
|
|
)
|
|
session.add(downtime)
|
|
machine.status = "arret_non_qualifie"
|
|
machine.status_since = event_time
|
|
session.flush()
|
|
return downtime
|
|
|
|
|
|
def close_open_downtime(session: Session, machine: Machine, event_time: datetime) -> Downtime | None:
|
|
downtime = get_open_downtime(session, machine.id)
|
|
if downtime is None:
|
|
return None
|
|
downtime.end_time = event_time
|
|
downtime.duration_sec = max(int((event_time - downtime.start_time).total_seconds()), 0)
|
|
machine.status_since = event_time
|
|
session.flush()
|
|
return downtime
|
|
|
|
|
|
def _is_matching_edge(previous_value: bool | None, current_value: bool | None, configured_edge: str) -> bool:
|
|
if previous_value is None or current_value is None:
|
|
return False
|
|
if configured_edge == "falling":
|
|
return previous_value is True and current_value is False
|
|
return previous_value is False and current_value is True
|
|
|
|
|
|
def _persist_cycle(
|
|
session: Session,
|
|
machine: Machine,
|
|
running_order: ProductionOrder | None,
|
|
event_time: datetime,
|
|
cycle_time_sec: float,
|
|
) -> None:
|
|
cavities = running_order.cavities if running_order else 1
|
|
is_valid = machine.min_cycle_time_sec <= cycle_time_sec <= machine.max_cycle_time_sec
|
|
session.add(
|
|
Cycle(
|
|
machine_id=machine.id,
|
|
production_order_id=running_order.id if running_order else None,
|
|
timestamp=event_time,
|
|
cycle_time_sec=cycle_time_sec,
|
|
cavities=cavities,
|
|
produced_qty=cavities if is_valid else 0,
|
|
is_valid=is_valid,
|
|
)
|
|
)
|
|
if cycle_time_sec >= machine.min_cycle_time_sec:
|
|
machine.last_cycle_at = event_time
|
|
machine.status_since = event_time
|
|
|
|
|
|
def _handle_cycle_signal_event(
|
|
session: Session,
|
|
machine: Machine,
|
|
running_order: ProductionOrder | None,
|
|
incoming_event: MachineEventIn,
|
|
event_time: datetime,
|
|
previous_signal_event: MachineEvent | None,
|
|
) -> None:
|
|
input_value = normalize_signal_value(incoming_event.input_value)
|
|
previous_value = None
|
|
previous_signal_time = None
|
|
if previous_signal_event is not None:
|
|
previous_value = normalize_signal_value(previous_signal_event.payload.get("input_value"))
|
|
previous_signal_time = _ensure_utc(previous_signal_event.timestamp)
|
|
|
|
if input_value is None or not _is_matching_edge(previous_value, input_value, machine.cycle_signal_edge):
|
|
return
|
|
|
|
if previous_signal_time is not None:
|
|
elapsed_ms = (event_time - previous_signal_time).total_seconds() * 1000
|
|
if elapsed_ms < machine.debounce_ms:
|
|
return
|
|
|
|
close_open_downtime(session, machine, event_time)
|
|
|
|
if machine.last_cycle_at is None:
|
|
fallback_cycle_time = incoming_event.cycle_time_sec
|
|
if fallback_cycle_time and fallback_cycle_time >= machine.min_cycle_time_sec:
|
|
_persist_cycle(session, machine, running_order, event_time, float(fallback_cycle_time))
|
|
else:
|
|
machine.last_cycle_at = event_time
|
|
machine.status_since = event_time
|
|
return
|
|
|
|
cycle_time_sec = (event_time - _ensure_utc(machine.last_cycle_at)).total_seconds()
|
|
_persist_cycle(session, machine, running_order, event_time, cycle_time_sec)
|
|
|
|
|
|
def handle_machine_event(session: Session, incoming_event: MachineEventIn) -> dict:
|
|
event_time = _ensure_utc(incoming_event.timestamp)
|
|
machine = _get_or_create_machine(session, incoming_event.machine_id)
|
|
previous_signal_event = None
|
|
should_refresh = False
|
|
if incoming_event.event_type == SIGNAL_EVENT_TYPE and incoming_event.input_name:
|
|
previous_signal_event = get_last_input_event(session, machine.id, incoming_event.input_name)
|
|
|
|
session.add(
|
|
MachineEvent(
|
|
machine_id=machine.id,
|
|
event_type=incoming_event.event_type,
|
|
timestamp=event_time,
|
|
payload=incoming_event.model_dump(mode="json"),
|
|
)
|
|
)
|
|
|
|
running_order = get_running_order(session, machine.id)
|
|
|
|
if incoming_event.event_type == "machine_power_on":
|
|
machine.powered_on = True
|
|
should_refresh = True
|
|
elif incoming_event.event_type == "machine_power_off":
|
|
machine.powered_on = False
|
|
close_open_downtime(session, machine, event_time)
|
|
should_refresh = True
|
|
elif incoming_event.event_type == "machine_stopped":
|
|
open_downtime_if_needed(session, machine, event_time, auto_detected=True)
|
|
should_refresh = True
|
|
elif incoming_event.event_type == "cycle_completed":
|
|
close_open_downtime(session, machine, event_time)
|
|
cycle_time_sec = float(incoming_event.cycle_time_sec or 0)
|
|
_persist_cycle(session, machine, running_order, event_time, cycle_time_sec)
|
|
should_refresh = True
|
|
elif incoming_event.event_type == SIGNAL_EVENT_TYPE:
|
|
input_name = incoming_event.input_name
|
|
input_value = normalize_signal_value(incoming_event.input_value)
|
|
if input_name == POWER_INPUT_NAME and input_value is not None:
|
|
machine.powered_on = input_value
|
|
if not input_value:
|
|
close_open_downtime(session, machine, event_time)
|
|
should_refresh = True
|
|
elif input_name == CYCLE_INPUT_NAME:
|
|
previous_cycle_at = machine.last_cycle_at
|
|
_handle_cycle_signal_event(session, machine, running_order, incoming_event, event_time, previous_signal_event)
|
|
should_refresh = machine.last_cycle_at != previous_cycle_at
|
|
elif input_name == "general_alarm" and input_value:
|
|
should_refresh = True
|
|
|
|
machine.status = derive_machine_status(machine, get_open_downtime(session, machine.id), running_order)
|
|
session.commit()
|
|
|
|
return {
|
|
"machine_code": machine.code,
|
|
"machine_id": machine.id,
|
|
"event_type": incoming_event.event_type,
|
|
"timestamp": event_time.isoformat(),
|
|
"status": machine.status,
|
|
"should_refresh": should_refresh,
|
|
}
|
|
|
|
|
|
def detect_missing_cycles(session: Session, now: datetime | None = None) -> list[dict]:
|
|
current_time = _ensure_utc(now) if now else datetime.now(UTC)
|
|
machines = session.execute(select(Machine).where(Machine.is_active.is_(True))).scalars().all()
|
|
changes: list[dict] = []
|
|
|
|
for machine in machines:
|
|
running_order = get_running_order(session, machine.id)
|
|
if running_order is None or not machine.powered_on or machine.last_cycle_at is None:
|
|
continue
|
|
|
|
delay_sec = max(machine.stop_detection_delay_sec, 1)
|
|
last_cycle_at = _ensure_utc(machine.last_cycle_at)
|
|
elapsed = (current_time - last_cycle_at).total_seconds()
|
|
if elapsed < delay_sec:
|
|
continue
|
|
|
|
downtime = get_open_downtime(session, machine.id)
|
|
if downtime is not None:
|
|
continue
|
|
|
|
downtime_start = last_cycle_at + timedelta(seconds=delay_sec)
|
|
open_downtime_if_needed(session, machine, downtime_start, auto_detected=True)
|
|
machine.status = "arret_non_qualifie"
|
|
machine.status_since = current_time
|
|
changes.append({"machine_code": machine.code, "event_type": "machine_stopped", "status": machine.status})
|
|
|
|
if changes:
|
|
session.commit()
|
|
else:
|
|
session.rollback()
|
|
return changes
|