Source code for fhops.evaluation.metrics.kpis

"""KPI helpers for FHOPS schedules."""

from __future__ import annotations

import json
from collections import Counter, defaultdict
from collections.abc import Iterable, Iterator, Mapping
from dataclasses import dataclass, field
from typing import Any, ClassVar

import pandas as pd

from fhops.evaluation.playback import PlaybackConfig, run_playback
from fhops.evaluation.playback.aggregates import (
    DAY_SUMMARY_COLUMNS,
    SHIFT_SUMMARY_COLUMNS,
    day_dataframe,
    shift_dataframe,
)
from fhops.evaluation.playback.core import PlaybackRecord, shift_hours_resolver
from fhops.evaluation.sequencing import build_role_priority
from fhops.optimization.operational_problem import build_operational_problem
from fhops.scenario.contract import Problem

from .aggregates import compute_makespan_metrics, compute_utilisation_metrics

__all__ = ["KPIResult", "compute_kpis"]


[docs] @dataclass(slots=True) class KPIResult(Mapping[str, float | int | str]): """Structured KPI bundle with optional shift/day calendar attachments.""" totals: dict[str, float | int | str] = field(default_factory=dict) shift_calendar: pd.DataFrame | None = None day_calendar: pd.DataFrame | None = None sequencing_debug: dict[str, object] | None = None SHIFT_COLUMNS: ClassVar[tuple[str, ...]] = tuple(SHIFT_SUMMARY_COLUMNS) DAY_COLUMNS: ClassVar[tuple[str, ...]] = tuple(DAY_SUMMARY_COLUMNS) def __post_init__(self) -> None: if self.shift_calendar is not None: missing = set(self.SHIFT_COLUMNS) - set(self.shift_calendar.columns) if missing: raise ValueError(f"Shift calendar missing columns: {sorted(missing)}") self.shift_calendar = self.shift_calendar.reindex(columns=self.SHIFT_COLUMNS).copy() if self.day_calendar is not None: missing = set(self.DAY_COLUMNS) - set(self.day_calendar.columns) if missing: raise ValueError(f"Day calendar missing columns: {sorted(missing)}") self.day_calendar = self.day_calendar.reindex(columns=self.DAY_COLUMNS).copy() # Mapping interface ------------------------------------------------- def __getitem__(self, key: str) -> float | int | str: return self.totals[key] def __iter__(self) -> Iterator[str]: return iter(self.totals) def __len__(self) -> int: return len(self.totals)
[docs] def get(self, key: str, default: Any = None) -> Any: return self.totals.get(key, default)
# Convenience helpers -----------------------------------------------
[docs] def to_dict(self) -> dict[str, float | int | str]: """Return the scalar KPI totals as a plain dictionary.""" return dict(self.totals)
[docs] def with_calendars( self, *, shift_calendar: pd.DataFrame | None = None, day_calendar: pd.DataFrame | None = None, ) -> KPIResult: """Return a copy with the provided calendars attached.""" return KPIResult( totals=self.to_dict(), shift_calendar=shift_calendar if shift_calendar is not None else self.shift_calendar, day_calendar=day_calendar if day_calendar is not None else self.day_calendar, )
@dataclass(slots=True) class _EventLosses: downtime_volume: float = 0.0 weather_volume: float = 0.0 weather_hours: float = 0.0 def _event_losses(pb: Problem, records: Iterable[PlaybackRecord]) -> _EventLosses: """Sum record-level downtime/weather losses (see :func:`compute_kpis` Notes).""" scenario = pb.scenario rates = { (rate.machine_id, rate.block_id): float(rate.rate) for rate in scenario.production_rates } hours_for = shift_hours_resolver(scenario) losses = _EventLosses() for record in records: shift_hours, _source = hours_for(record.machine_id, record.shift_id, record.day) if record.downtime: recorded = record.metadata.get("downtime_production_lost") if isinstance(recorded, int | float): losses.downtime_volume += float(recorded) elif record.downtime_hours and shift_hours and record.block_id is not None: fraction = min(float(record.downtime_hours) / float(shift_hours), 1.0) losses.downtime_volume += fraction * rates.get( (record.machine_id, record.block_id), 0.0 ) severity = float(record.weather_severity or 0.0) if severity > 0: if shift_hours: losses.weather_hours += severity * float(shift_hours) recorded = record.metadata.get("weather_production_lost") if isinstance(recorded, int | float): losses.weather_volume += float(recorded) elif severity < 1.0 and record.production_units: losses.weather_volume += ( float(record.production_units) * severity / (1.0 - severity) ) return losses
[docs] def compute_kpis(pb: Problem, assignments: pd.DataFrame) -> KPIResult: """Compute production, mobilisation, utilisation, and sequencing KPIs from assignments. Parameters ---------- pb : fhops.scenario.contract.Problem Problem wrapping the scenario whose blocks define ``work_required`` (m³). assignments : pandas.DataFrame Solver assignments (``machine_id``, ``block_id``, ``day`` and optional ``shift_id``, ``assigned``, ``production``) replayed through deterministic playback. May be empty (e.g. a solver run that found no solution); see Notes. Returns ------- KPIResult Scalar KPI totals. Volume KPIs are in m³, the units of ``Block.work_required``: ``total_production`` is the volume delivered by each block's terminal role and ``remaining_work_total`` / ``staged_production`` the volume still to deliver. Notes ----- ``total_production`` is always the playback ``delivered_total``; ``remaining_work_total`` is ``sum(Block.work_required) - total_production`` (tracked per block by playback). Both respect the scenario as given, so for a rolling-horizon window with carried-forward ``work_required`` the total is the reduced volume. Empty plans (no rows, or no row with ``assigned > 0``) and partial plans are evaluated like any other plan, which gives for an empty plan: * ``total_production = 0``, ``remaining_work_total = staged_production = sum(Block.work_required)``, ``completed_blocks = 0``; * ``makespan_day = 0`` and ``makespan_shift = "N/A"``; * ``utilisation_ratio_mean_day`` / ``utilisation_ratio_weighted_day`` are ``0`` (every available scenario day is idle) and the shift/machine/role utilisation keys are absent (there are no worked machine-shifts); * ``sequencing_violation_*`` counts are ``0``, ``sequencing_violation_breakdown`` is ``"none"`` and ``sequencing_clean_blocks`` counts every harvest-system block (no block has a violation, including blocks the plan never touches); * the mobilisation, downtime, and weather keys are absent (they are only emitted when non-zero). FHOPS ≤ 1.0.0 reported ``total_production = sum(Block.work_required)`` and ``remaining_work_total = 0`` for an empty assignment frame (#108). Downtime and weather loss KPIs (emitted only for frames carrying stochastic event columns, e.g. a sample produced by :func:`~fhops.evaluation.playback.run_stochastic_playback`) are record-level sums (FHOPS 1.0.1, #116): * ``downtime_production_loss_est`` — Σ over downtime records of the proposed volume (m³) the downtime event removed from that record (``_downtime_lost``: the full proposed volume for a cancelled shift, ``volume × d / shift_hours`` for a partial loss). Records without that column use ``rate(machine, block) × min(downtime_hours / shift_hours, 1)``. * ``weather_production_loss_est`` — Σ over weather-hit records of the proposed volume (m³) removed by weather (``_weather_lost`` = ``volume × severity``); without that column, ``production_units × severity / (1 − severity)``. * ``weather_hours_est`` — Σ over weather-hit records of ``severity × shift_hours``. ``shift_hours`` follows :func:`~fhops.evaluation.playback.core.shift_hours_resolver`. The losses are measured on the proposed production of each machine-shift, before sequencing caps, so they include upstream (non-terminal) volume; they are not the change in ``total_production``, which also reflects knock-on sequencing effects. FHOPS 1.0.0 estimated both as hours × (delivered terminal volume / recorded hours of every role), which does not match the volume the events actually removed. """ playback_result = run_playback(pb, assignments, config=PlaybackConfig()) shift_df = shift_dataframe(playback_result) day_df = day_dataframe(playback_result) delivered_total = float(playback_result.delivered_total) remaining_work_total = float(playback_result.remaining_work_total) sc = pb.scenario total_required = sum(block.work_required for block in sc.blocks) mobilisation_cost = 0.0 mobilisation_by_machine: dict[str, float] = defaultdict(float) mobilisation_by_landing: dict[str, float] = defaultdict(float) completed_blocks: set[str] = set() seq_violation_events = 0 seq_violation_blocks: set[str] = set() seq_violation_days: set[tuple[str, int]] = set() seq_reason_counts: Counter[str] = Counter() for record in playback_result.records: if record.mobilisation_cost: cost = float(record.mobilisation_cost) mobilisation_cost += cost mobilisation_by_machine[record.machine_id] += cost landing_id = record.metadata.get("landing_id") or record.landing_id if landing_id is not None: mobilisation_by_landing[str(landing_id)] += cost if record.metadata.get("block_completed") and record.block_id: completed_blocks.add(record.block_id) violation = record.metadata.get("sequencing_violation") if violation and record.block_id: seq_violation_events += 1 seq_violation_blocks.add(record.block_id) seq_violation_days.add((record.block_id, record.day)) seq_reason_counts[str(violation)] += 1 result: dict[str, float | int | str] = { "total_production": delivered_total, "completed_blocks": float(len(completed_blocks)), } staged_volume = remaining_work_total residual = float(total_required) - delivered_total if residual >= 0 and abs(staged_volume - residual) <= 1e-6: staged_volume = residual if abs(staged_volume) <= 1e-6: staged_volume = 0.0 result["staged_production"] = staged_volume result["remaining_work_total"] = staged_volume # ``total_required - remaining`` is only a float-tidy form of the delivered volume; never # let it replace ``delivered_total`` when the two disagree (FHOPS <= 1.0.0 did, so an empty # plan reported the full scenario volume as delivered, #108). adjusted_total = float(total_required) - staged_volume tolerance = max(1e-6, 1e-9 * float(total_required)) if adjusted_total >= 0 and abs(adjusted_total - delivered_total) <= tolerance: result["total_production"] = adjusted_total if mobilisation_cost > 0: result["mobilisation_cost"] = mobilisation_cost result["mobilisation_cost_by_machine"] = json.dumps( {machine: round(cost, 3) for machine, cost in sorted(mobilisation_by_machine.items())} ) result["mobilisation_cost_by_landing"] = json.dumps( {landing: round(cost, 3) for landing, cost in sorted(mobilisation_by_landing.items())} ) system_blocks = {block.id for block in sc.blocks if block.harvest_system_id} if system_blocks: result["sequencing_violation_count"] = seq_violation_events result["sequencing_violation_blocks"] = len(seq_violation_blocks) result["sequencing_violation_days"] = len(seq_violation_days) clean_blocks = max(len(system_blocks - seq_violation_blocks), 0) result["sequencing_clean_blocks"] = clean_blocks result["sequencing_violation_breakdown"] = ( ", ".join(f"{reason}={count}" for reason, count in sorted(seq_reason_counts.items())) if seq_reason_counts else "none" ) non_default_repairs = [ machine.id for machine in sc.machines if getattr(machine, "repair_usage_hours", None) not in (None, 10000) ] if non_default_repairs: result["repair_usage_alert"] = ", ".join(sorted(non_default_repairs)) ctx = build_operational_problem(pb) utilisation_metrics = compute_utilisation_metrics(shift_df, day_df) role_metric = utilisation_metrics.get("utilisation_ratio_by_role") if role_metric: role_priority = build_role_priority(ctx) ordered_payload = json.loads(role_metric) ordered_items = sorted( ordered_payload.items(), key=lambda item: (role_priority.get(item[0], 999), item[0]), ) utilisation_metrics["utilisation_ratio_by_role"] = json.dumps( {role: value for role, value in ordered_items} ) result.update(utilisation_metrics) if not shift_df.empty and "production_units" in shift_df.columns: productive_rows = shift_df[shift_df["production_units"] > 0] days_with_work = set(int(day) for day in productive_rows["day"].tolist()) shift_keys_with_work = { (int(row["day"]), str(row["shift_id"])) for _, row in productive_rows[["day", "shift_id"]].iterrows() } else: days_with_work = set() shift_keys_with_work = set() makespan_metrics = compute_makespan_metrics( pb, shift_df, fallback_days=days_with_work, fallback_shift_keys=shift_keys_with_work, ) result.update(makespan_metrics) loss = _event_losses(pb, playback_result.records) if "downtime_hours" in shift_df.columns: total_downtime_hours = float(shift_df["downtime_hours"].sum()) if total_downtime_hours > 0: result["downtime_hours_total"] = total_downtime_hours result["downtime_production_loss_est"] = loss.downtime_volume downtime_by_machine = shift_df.groupby("machine_id", dropna=False)["downtime_hours"].sum() downtime_by_machine = downtime_by_machine[downtime_by_machine > 0] if not downtime_by_machine.empty: result["downtime_hours_by_machine"] = json.dumps( { machine: round(float(hours), 3) for machine, hours in sorted(downtime_by_machine.items()) } ) downtime_events_series = shift_df.get("downtime_events") total_downtime_events = ( int(float(downtime_events_series.sum())) if downtime_events_series is not None else 0 ) if total_downtime_events > 0: result["downtime_event_count"] = total_downtime_events if "weather_severity_total" in shift_df.columns: total_weather_severity = float(shift_df["weather_severity_total"].sum()) if total_weather_severity > 0: result["weather_severity_total"] = total_weather_severity result["weather_hours_est"] = loss.weather_hours result["weather_production_loss_est"] = loss.weather_volume weather_by_machine = shift_df.groupby("machine_id", dropna=False)[ "weather_severity_total" ].sum() weather_by_machine = weather_by_machine[weather_by_machine > 0] if not weather_by_machine.empty: result["weather_severity_by_machine"] = json.dumps( { machine: round(float(value), 3) for machine, value in sorted(weather_by_machine.items()) } ) return KPIResult( totals=result, shift_calendar=shift_df, day_calendar=day_df, sequencing_debug=playback_result.sequencing_debug, )