fhops.evaluation Package

The evaluation layer turns solver assignments into KPI-rich reports. It houses deterministic playback, stochastic extensions, and KPI calculators. Use it to:

  • Convert solver outputs into shift/day summaries (fhops.evaluation.playback.run_playback()).

  • Compute KPI bundles for CLI reports or notebooks (fhops.evaluation.metrics.kpis.compute_kpis()).

  • Export CSV/Parquet/Markdown summaries for documentation or telemetry.

Example:

from fhops.scenario.io import load_scenario
from fhops.scenario.contract import Problem
from fhops.optimization.mip import solve_mip
from fhops.evaluation import compute_kpis

scenario = load_scenario("examples/tiny7/scenario.yaml")
problem = Problem.from_scenario(scenario)
mip_res = solve_mip(problem, time_limit=60)
kpis = compute_kpis(problem, mip_res["assignments"])
print(kpis["total_production"], kpis["mobilisation_cost"])

Evaluation layer (playback, metrics, reporting).

class fhops.evaluation.DaySummary(day, available_hours=0.0, total_hours=0.0, production_units=0.0, mobilisation_cost=0.0, completed_blocks=0, idle_hours=None, blackout_conflicts=0, sequencing_violations=0, utilisation_ratio=None, sample_id=0, downtime_hours=0.0, downtime_events=0, weather_severity_total=0.0)[source]

Bases: object

Aggregated metrics per day across machines.

Parameters:

  • day (int)

  • available_hours (float)

  • total_hours (float)

  • production_units (float)

  • mobilisation_cost (float)

  • completed_blocks (int)

  • idle_hours (float | None)

  • blackout_conflicts (int)

  • sequencing_violations (int)

  • utilisation_ratio (float | None)

  • sample_id (int)

  • downtime_hours (float)

  • downtime_events (int)

  • weather_severity_total (float)

available_hours: float
blackout_conflicts: int
completed_blocks: int
day: int
downtime_events: int
downtime_hours: float
idle_hours: float | None
mobilisation_cost: float
production_units: float
sample_id: int
sequencing_violations: int
total_hours: float
utilisation_ratio: float | None
weather_severity_total: float
class fhops.evaluation.DowntimeEvent(config)[source]

Bases: object

Randomly remove assignments to simulate downtime.

apply(context, assignments, base_production)[source]
Parameters:
Return type:

DataFrame

class fhops.evaluation.DowntimeEventConfig(*, enabled=True, seed_offset=0, probability=0.15, mean_duration_hours=4.0, std_duration_hours=1.5, max_concurrent=None, target_machine_roles=None)[source]

Bases: SamplingEventConfig

Describes downtime sampling parameters for machines.

Parameters:

  • enabled (bool)

  • seed_offset (int)

  • probability (float)

  • mean_duration_hours (float)

  • std_duration_hours (float)

  • max_concurrent (int | None)

  • target_machine_roles (list[str] | None)

max_concurrent: int | None
mean_duration_hours: float
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

probability: float
std_duration_hours: float
target_machine_roles: list[str] | None
class fhops.evaluation.EnsembleResult(base_result, samples)[source]

Bases: object

Aggregate of the baseline deterministic playback plus stochastic samples.

Parameters:
base_result: PlaybackResult
samples: list[PlaybackSample]
class fhops.evaluation.KPIResult(totals=<factory>, shift_calendar=None, day_calendar=None, sequencing_debug=None)[source]

Bases: Mapping[str, float | int | str]

Structured KPI bundle with optional shift/day calendar attachments.

Parameters:
DAY_COLUMNS: ClassVar[tuple[str, ...]] = ('day', 'sample_id', 'production_units', 'total_hours', 'idle_hours', 'mobilisation_cost', 'completed_blocks', 'blackout_conflicts', 'sequencing_violations', 'available_hours', 'utilisation_ratio', 'downtime_hours', 'downtime_events', 'weather_severity_total')
SHIFT_COLUMNS: ClassVar[tuple[str, ...]] = ('day', 'shift_id', 'machine_id', 'machine_role', 'sample_id', 'production_units', 'total_hours', 'idle_hours', 'mobilisation_cost', 'sequencing_violations', 'blackout_conflicts', 'available_hours', 'utilisation_ratio', 'downtime_hours', 'downtime_events', 'weather_severity_total')
day_calendar: DataFrame | None
get(k[, d]) → D[k] if k in D, else d.  d defaults to None.[source]
Parameters:
Return type:

Any

sequencing_debug: dict[str, object] | None
shift_calendar: DataFrame | None
to_dict()[source]

Return the scalar KPI totals as a plain dictionary.

Return type:

dict[str, float | int | str]

totals: dict[str, float | int | str]
with_calendars(*, shift_calendar=None, day_calendar=None)[source]

Return a copy with the provided calendars attached.

Parameters:

  • shift_calendar (DataFrame | None)

  • day_calendar (DataFrame | None)

Return type:

KPIResult

class fhops.evaluation.LandingShockConfig(*, enabled=True, seed_offset=0, probability=0.1, capacity_multiplier_range=(0.4, 0.8), duration_days=1, target_landing_ids=None)[source]

Bases: SamplingEventConfig

Parameterises landing congestion shocks reducing capacity.

Parameters:
capacity_multiplier_range: tuple[float, float]
duration_days: int
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

probability: float
target_landing_ids: list[str] | None
class fhops.evaluation.LandingShockEvent(config)[source]

Bases: object

Reduce landing throughput via random shocks.

apply(context, assignments, base_production)[source]
Parameters:
Return type:

DataFrame

class fhops.evaluation.PlaybackConfig(respect_blackouts=True, infer_missing_shifts=True, include_idle_records=False)[source]

Bases: object

Tuning options for deterministic playback execution.

Parameters:

  • respect_blackouts (bool)

  • infer_missing_shifts (bool)

  • include_idle_records (bool)

include_idle_records: bool
infer_missing_shifts: bool
respect_blackouts: bool
class fhops.evaluation.PlaybackEvent(*args, **kwargs)[source]

Bases: Protocol

Stochastic event modifying assignments before playback summarisation.

apply(context, assignments, base_production)[source]
Parameters:
Return type:

DataFrame

class fhops.evaluation.PlaybackRecord(day, shift_id, machine_id, block_id, hours_worked=None, production_units=None, mobilisation_cost=None, blackout_hit=False, landing_id=None, machine_role=None, downtime=False, weather_severity=None, metadata=<factory>, sample_id=0)[source]

Bases: object

Atomic shift-level record produced by deterministic playback.

Parameters:

  • day (int)

  • shift_id (str)

  • machine_id (str)

  • block_id (str | None)

  • hours_worked (float | None)

  • production_units (float | None)

  • mobilisation_cost (float | None)

  • blackout_hit (bool)

  • landing_id (str | None)

  • machine_role (str | None)

  • downtime (bool)

  • weather_severity (float | None)

  • metadata (dict[str, object])

  • sample_id (int)

blackout_hit: bool
block_id: str | None
day: int
downtime: bool
hours_worked: float | None
landing_id: str | None
machine_id: str
machine_role: str | None
metadata: dict[str, object]
mobilisation_cost: float | None
production_units: float | None
sample_id: int
shift_id: str
weather_severity: float | None
class fhops.evaluation.PlaybackResult(records, shift_summaries, day_summaries, config, sequencing_debug=None, sample_id=0, delivered_total=0.0, remaining_work_total=0.0)[source]

Bases: object

Container grouping playback outputs.

Parameters:
config: PlaybackConfig
day_summaries: Sequence[DaySummary]
delivered_total: float
records: Sequence[PlaybackRecord]
remaining_work_total: float
sample_id: int
sequencing_debug: dict[str, object] | None
shift_summaries: Sequence[ShiftSummary]
class fhops.evaluation.PlaybackSample(sample_id, result)[source]

Bases: object

Container pairing a stochastic sample ID with its PlaybackResult.

Parameters:
result: PlaybackResult
sample_id: int
class fhops.evaluation.SamplingConfig(*, samples=10, base_seed=123, downtime=<factory>, weather=<factory>, landing=<factory>)[source]

Bases: BaseModel

Top-level configuration for stochastic playback ensembles.

Parameters:
base_seed: int
downtime: DowntimeEventConfig
landing: LandingShockConfig
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

samples: int
weather: WeatherEventConfig
class fhops.evaluation.SamplingContext(problem, sample_id, rng, config)[source]

Bases: object

Per-sample execution context.

Parameters:
config: SamplingConfig
problem: Problem
rng: Generator
sample_id: int
class fhops.evaluation.SamplingEventConfig(*, enabled=True, seed_offset=0)[source]

Bases: BaseModel

Base configuration shared by all stochastic events.

Parameters:
enabled: bool
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

seed_offset: int
class fhops.evaluation.ShiftSummary(day, shift_id, machine_id, machine_role=None, available_hours=0.0, total_hours=0.0, production_units=0.0, mobilisation_cost=0.0, idle_hours=None, blackout_conflicts=0, sequencing_violations=0, utilisation_ratio=None, sample_id=0, downtime_hours=0.0, downtime_events=0, weather_severity_total=0.0)[source]

Bases: object

Aggregated metrics per machine/shift.

Parameters:

  • day (int)

  • shift_id (str)

  • machine_id (str)

  • machine_role (str | None)

  • available_hours (float)

  • total_hours (float)

  • production_units (float)

  • mobilisation_cost (float)

  • idle_hours (float | None)

  • blackout_conflicts (int)

  • sequencing_violations (int)

  • utilisation_ratio (float | None)

  • sample_id (int)

  • downtime_hours (float)

  • downtime_events (int)

  • weather_severity_total (float)

available_hours: float
blackout_conflicts: int
day: int
downtime_events: int
downtime_hours: float
idle_hours: float | None
machine_id: str
machine_role: str | None
mobilisation_cost: float
production_units: float
sample_id: int
sequencing_violations: int
shift_id: str
total_hours: float
utilisation_ratio: float | None
weather_severity_total: float
class fhops.evaluation.WeatherEvent(config)[source]

Bases: object

Adjust production based on weather severity.

apply(context, assignments, base_production)[source]
Parameters:
Return type:

DataFrame

class fhops.evaluation.WeatherEventConfig(*, enabled=True, seed_offset=0, day_probability=0.2, severity_levels=<factory>, correlated_days=True, impact_window_days=1, affected_shifts=None)[source]

Bases: SamplingEventConfig

Captures stochastic weather impacts affecting production rates.

Parameters:
affected_shifts: list[str] | None
correlated_days: bool
day_probability: float
impact_window_days: int
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

severity_levels: dict[str, float]
fhops.evaluation.assignments_to_records(problem, assignments)[source]

Convert solver assignments dataframe into playback records.

Parameters:

  • problem (Problem)

  • assignments (DataFrame)

Return type:

Iterator[PlaybackRecord]

fhops.evaluation.compute_kpis(pb, assignments)[source]

Compute production, mobilisation, utilisation, and sequencing KPIs from assignments.

Parameters:

  • pb (Problem)

  • assignments (DataFrame)

Return type:

KPIResult

fhops.evaluation.compute_makespan_metrics(problem, shift_df, *, fallback_days=None, fallback_shift_keys=None)[source]

Compute makespan metrics (latest productive day/shift) given playback summaries.

Parameters:
Return type:

dict[str, Any]

fhops.evaluation.compute_utilisation_metrics(shift_df, day_df=None)[source]

Compute utilisation KPI metrics from shift/day playback DataFrames.

Parameters:

  • shift_df (DataFrame)

  • day_df (DataFrame | None)

Return type:

dict[str, Any]

fhops.evaluation.day_dataframe(result)[source]

Return the day-level playback summaries as a DataFrame.

Parameters:

result (PlaybackResult)

Return type:

DataFrame

fhops.evaluation.day_dataframe_from_ensemble(ensemble, *, include_base=False)[source]

Concatenate day-level summaries from an ensemble.

Parameters:
Return type:

DataFrame

fhops.evaluation.export_playback(shift_df, day_df, *, shift_csv=None, day_csv=None, shift_parquet=None, day_parquet=None, summary_md=None)[source]

Write playback summaries to the requested outputs.

Parameters:
Returns:

Scalar metrics from playback_summary_metrics() (samples, total production, etc.).

Return type:

dict

fhops.evaluation.machine_utilisation_summary(shift_df)[source]

Aggregate shift summaries into per-machine utilisation metrics.

Parameters:

shift_df (DataFrame) – DataFrame produced by shift_dataframe() or *_from_ensemble; must include total_hours and available_hours columns.

Returns:

One row per (sample_id, machine_id) with total/available hours, utilisation ratio, production, mobilisation cost, and constraint violation counts.

Return type:

pandas.DataFrame

fhops.evaluation.playback_summary_metrics(shift_df, day_df)[source]

Compute scalar summary metrics (production, hours, mobilisation, utilisation).

Parameters:

  • shift_df (DataFrame)

  • day_df (DataFrame)

Return type:

dict[str, Any]

fhops.evaluation.render_markdown_summary(shift_df, day_df, metrics=None)[source]

Return a Markdown-formatted playback summary.

Parameters:

  • shift_df (DataFrame)

  • day_df (DataFrame)

  • metrics (dict[str, Any] | None)

Return type:

str

fhops.evaluation.run_playback(problem, assignments, *, config=None, sample_id=0)[source]

Convert solver assignments into playback records and aggregated summaries.

Parameters:

  • problem (Problem)

  • assignments (pd.DataFrame)

  • config (PlaybackConfig | None)

  • sample_id (int)

Return type:

PlaybackResult

fhops.evaluation.run_stochastic_playback(problem, assignments, *, sampling_config, events=None)[source]

Run stochastic playback over multiple samples.

Parameters:
Return type:

EnsembleResult

fhops.evaluation.schedule_to_records(problem, schedule)[source]

Convert a heuristic Schedule plan into playback records.

Parameters:

  • problem (Problem)

  • schedule (object)

Return type:

Iterator[PlaybackRecord]

fhops.evaluation.shift_dataframe(result)[source]

Return the shift-level playback summaries as a DataFrame.

Parameters:

result (PlaybackResult) – PlaybackResult returned by fhops.evaluation.playback.core.run_playback().

Return type:

DataFrame

fhops.evaluation.shift_dataframe_from_ensemble(ensemble, *, include_base=False)[source]

Concatenate shift summaries from a stochastic ensemble.

Parameters:

  • ensemble (EnsembleResult) – Result of fhops.evaluation.playback.stochastic.run_stochastic_playback().

  • include_base (bool) – When True, include the base deterministic result in addition to samples.

Return type:

DataFrame

fhops.evaluation.summarise_days(shift_summaries, availability_map, completed_by_day, sample_id=0)[source]

Aggregate playback results to day-level summaries.

Parameters:

  • shift_summaries (Iterable[ShiftSummary]) – Iterable produced by summarise_shifts().

  • availability_map (dict[tuple[int, str, str], float]) – Same map passed to summarise_shifts; used to recover day-level availability totals.

  • completed_by_day (dict[int, set[str]]) – Mapping day -> set(block_id) indicating which blocks completed on each day (consumed for KPI reporting).

  • sample_id (int) – Propagated through stochastic playback.

Return type:

Iterator[DaySummary]

fhops.evaluation.summarise_shifts(records, availability_map, *, include_idle=False, sample_id=0)[source]

Aggregate playback records to machine/shift summaries.

Parameters:

  • records (Iterable[PlaybackRecord]) – Iterator of PlaybackRecord instances produced by assignments_to_records.

  • availability_map (dict[tuple[int, str, str], float]) – Mapping (day, shift_id, machine_id) -> available_hours computed via _compute_shift_availability; used to seed summaries and compute utilisation ratios.

  • include_idle (bool) – When True, emit summaries for every availability entry even if no work occurred (resulting in total_hours = 0 but preserving the availability baseline).

  • sample_id (int) – Identifier propagated through stochastic playback so downstream aggregations can group rows.

Return type:

Iterator[ShiftSummary]

KPI helpers for FHOPS schedules.

class fhops.evaluation.metrics.kpis.KPIResult(totals=<factory>, shift_calendar=None, day_calendar=None, sequencing_debug=None)[source]

Bases: Mapping[str, float | int | str]

Structured KPI bundle with optional shift/day calendar attachments.

Parameters:
DAY_COLUMNS: ClassVar[tuple[str, ...]] = ('day', 'sample_id', 'production_units', 'total_hours', 'idle_hours', 'mobilisation_cost', 'completed_blocks', 'blackout_conflicts', 'sequencing_violations', 'available_hours', 'utilisation_ratio', 'downtime_hours', 'downtime_events', 'weather_severity_total')
SHIFT_COLUMNS: ClassVar[tuple[str, ...]] = ('day', 'shift_id', 'machine_id', 'machine_role', 'sample_id', 'production_units', 'total_hours', 'idle_hours', 'mobilisation_cost', 'sequencing_violations', 'blackout_conflicts', 'available_hours', 'utilisation_ratio', 'downtime_hours', 'downtime_events', 'weather_severity_total')
day_calendar: DataFrame | None
get(k[, d]) → D[k] if k in D, else d.  d defaults to None.[source]
Parameters:
Return type:

Any

sequencing_debug: dict[str, object] | None
shift_calendar: DataFrame | None
to_dict()[source]

Return the scalar KPI totals as a plain dictionary.

Return type:

dict[str, float | int | str]

totals: dict[str, float | int | str]
with_calendars(*, shift_calendar=None, day_calendar=None)[source]

Return a copy with the provided calendars attached.

Parameters:

  • shift_calendar (DataFrame | None)

  • day_calendar (DataFrame | None)

Return type:

KPIResult

fhops.evaluation.metrics.kpis.compute_kpis(pb, assignments)[source]

Compute production, mobilisation, utilisation, and sequencing KPIs from assignments.

Parameters:

  • pb (Problem)

  • assignments (DataFrame)

Return type:

KPIResult