203 lines
7.5 KiB
Python
203 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime, time, timedelta, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..models import Cycle, Downtime, ProductionOrder, Scrap
|
|
|
|
UTC = timezone.utc
|
|
|
|
|
|
@dataclass
|
|
class OeeTotals:
|
|
planned_time_sec: float = 0
|
|
downtime_sec: float = 0
|
|
operating_time_sec: float = 0
|
|
total_cycles: int = 0
|
|
theoretical_cycle_time_sec: float = 0
|
|
performance_time_sec: float = 0
|
|
total_produced_qty: int = 0
|
|
good_qty: int = 0
|
|
scrap_qty: int = 0
|
|
availability: float = 0
|
|
performance: float = 0
|
|
quality: float = 0
|
|
oee: float = 0
|
|
|
|
def as_dict(self) -> dict[str, float | int]:
|
|
return {
|
|
"planned_time_sec": round(self.planned_time_sec, 2),
|
|
"downtime_sec": round(self.downtime_sec, 2),
|
|
"operating_time_sec": round(self.operating_time_sec, 2),
|
|
"total_cycles": self.total_cycles,
|
|
"theoretical_cycle_time_sec": round(self.theoretical_cycle_time_sec, 2),
|
|
"total_produced_qty": self.total_produced_qty,
|
|
"good_qty": self.good_qty,
|
|
"scrap_qty": self.scrap_qty,
|
|
"availability": round(self.availability, 4),
|
|
"performance": round(self.performance, 4),
|
|
"quality": round(self.quality, 4),
|
|
"oee": round(self.oee, 4),
|
|
}
|
|
|
|
|
|
def calculate_oee_metrics(
|
|
*,
|
|
planned_time_sec: float,
|
|
downtime_sec: float,
|
|
total_cycles: int,
|
|
performance_time_sec: float,
|
|
theoretical_cycle_time_sec: float,
|
|
total_produced_qty: int,
|
|
scrap_qty: int,
|
|
) -> OeeTotals:
|
|
operating_time_sec = max(planned_time_sec - downtime_sec, 0)
|
|
good_qty = max(total_produced_qty - scrap_qty, 0)
|
|
|
|
availability = operating_time_sec / planned_time_sec if planned_time_sec > 0 else 0
|
|
performance = performance_time_sec / operating_time_sec if operating_time_sec > 0 else 0
|
|
performance = min(max(performance, 0), 1)
|
|
quality = good_qty / total_produced_qty if total_produced_qty > 0 else 0
|
|
oee = availability * performance * quality
|
|
|
|
return OeeTotals(
|
|
planned_time_sec=planned_time_sec,
|
|
downtime_sec=downtime_sec,
|
|
operating_time_sec=operating_time_sec,
|
|
total_cycles=total_cycles,
|
|
theoretical_cycle_time_sec=theoretical_cycle_time_sec,
|
|
performance_time_sec=performance_time_sec,
|
|
total_produced_qty=total_produced_qty,
|
|
good_qty=good_qty,
|
|
scrap_qty=scrap_qty,
|
|
availability=availability,
|
|
performance=performance,
|
|
quality=quality,
|
|
oee=oee,
|
|
)
|
|
|
|
|
|
def overlap_seconds(window_start: datetime, window_end: datetime, start: datetime, end: datetime | None) -> float:
|
|
actual_end = end or window_end
|
|
overlap_start = max(window_start, start)
|
|
overlap_end = min(window_end, actual_end)
|
|
if overlap_end <= overlap_start:
|
|
return 0
|
|
return (overlap_end - overlap_start).total_seconds()
|
|
|
|
|
|
def compute_machine_oee(session: Session, machine_id: int, window_start: datetime, window_end: datetime) -> dict:
|
|
current_time = datetime.now(UTC)
|
|
effective_window_end = min(window_end, current_time)
|
|
if effective_window_end <= window_start:
|
|
effective_window_end = window_end
|
|
|
|
order_rows = session.execute(
|
|
select(ProductionOrder).where(
|
|
ProductionOrder.machine_id == machine_id,
|
|
ProductionOrder.started_at.is_not(None),
|
|
ProductionOrder.started_at < effective_window_end,
|
|
func.coalesce(ProductionOrder.ended_at, effective_window_end) > window_start,
|
|
)
|
|
).scalars()
|
|
|
|
planned_time_sec = 0.0
|
|
for order in order_rows:
|
|
effective_order_end = min(order.ended_at, effective_window_end) if order.ended_at else effective_window_end
|
|
planned_time_sec += overlap_seconds(window_start, effective_window_end, order.started_at, effective_order_end)
|
|
|
|
cycle_rows = session.execute(
|
|
select(
|
|
Cycle.production_order_id,
|
|
func.count(Cycle.id),
|
|
func.coalesce(func.sum(Cycle.produced_qty), 0),
|
|
).where(
|
|
Cycle.machine_id == machine_id,
|
|
Cycle.timestamp >= window_start,
|
|
Cycle.timestamp <= effective_window_end,
|
|
Cycle.is_valid.is_(True),
|
|
).group_by(Cycle.production_order_id)
|
|
).all()
|
|
|
|
total_cycles = sum(int(row[1]) for row in cycle_rows)
|
|
total_produced_qty = sum(int(row[2]) for row in cycle_rows)
|
|
|
|
performance_time_sec = 0.0
|
|
weighted_theoretical = 0.0
|
|
counted_cycles = 0
|
|
for production_order_id, cycle_count, _ in cycle_rows:
|
|
if production_order_id is None:
|
|
continue
|
|
order = session.get(ProductionOrder, production_order_id)
|
|
if order is None:
|
|
continue
|
|
performance_time_sec += order.theoretical_cycle_time_sec * int(cycle_count)
|
|
weighted_theoretical += order.theoretical_cycle_time_sec * int(cycle_count)
|
|
counted_cycles += int(cycle_count)
|
|
|
|
theoretical_cycle_time_sec = weighted_theoretical / counted_cycles if counted_cycles > 0 else 0
|
|
|
|
downtime_rows = session.execute(
|
|
select(Downtime).where(
|
|
Downtime.machine_id == machine_id,
|
|
Downtime.start_time < effective_window_end,
|
|
func.coalesce(Downtime.end_time, effective_window_end) > window_start,
|
|
)
|
|
).scalars()
|
|
downtime_sec = sum(
|
|
overlap_seconds(window_start, effective_window_end, row.start_time, min(row.end_time, effective_window_end) if row.end_time else effective_window_end)
|
|
for row in downtime_rows
|
|
)
|
|
|
|
scrap_qty = session.execute(
|
|
select(func.coalesce(func.sum(Scrap.quantity), 0)).where(
|
|
Scrap.machine_id == machine_id,
|
|
Scrap.timestamp >= window_start,
|
|
Scrap.timestamp <= effective_window_end,
|
|
)
|
|
).scalar_one()
|
|
|
|
if planned_time_sec == 0 and (total_cycles > 0 or downtime_sec > 0):
|
|
planned_time_sec = (effective_window_end - window_start).total_seconds()
|
|
|
|
totals = calculate_oee_metrics(
|
|
planned_time_sec=planned_time_sec,
|
|
downtime_sec=downtime_sec,
|
|
total_cycles=total_cycles,
|
|
performance_time_sec=performance_time_sec,
|
|
theoretical_cycle_time_sec=theoretical_cycle_time_sec,
|
|
total_produced_qty=total_produced_qty,
|
|
scrap_qty=int(scrap_qty),
|
|
)
|
|
return {
|
|
"machine_id": machine_id,
|
|
"from": window_start.isoformat(),
|
|
"to": effective_window_end.isoformat(),
|
|
**totals.as_dict(),
|
|
}
|
|
|
|
|
|
def compute_daily_oee(session: Session, target_date: date, machine_ids: list[int]) -> dict:
|
|
window_start = datetime.combine(target_date, time.min, tzinfo=UTC)
|
|
window_end = window_start + timedelta(days=1)
|
|
machine_metrics = [compute_machine_oee(session, machine_id, window_start, window_end) for machine_id in machine_ids]
|
|
|
|
overall = calculate_oee_metrics(
|
|
planned_time_sec=sum(item["planned_time_sec"] for item in machine_metrics),
|
|
downtime_sec=sum(item["downtime_sec"] for item in machine_metrics),
|
|
total_cycles=sum(item["total_cycles"] for item in machine_metrics),
|
|
performance_time_sec=sum(item["theoretical_cycle_time_sec"] * item["total_cycles"] for item in machine_metrics),
|
|
theoretical_cycle_time_sec=0,
|
|
total_produced_qty=sum(item["total_produced_qty"] for item in machine_metrics),
|
|
scrap_qty=sum(item["scrap_qty"] for item in machine_metrics),
|
|
)
|
|
|
|
return {
|
|
"date": target_date.isoformat(),
|
|
"machines": machine_metrics,
|
|
"overall": overall.as_dict(),
|
|
}
|