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.operational_problem import build_operational_problem
from fhops.model.milp.driver import solve_operational_milp
from fhops.evaluation import compute_kpis

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

(fhops.optimization.mip.solve_mip() is a deprecated wrapper around the same operational MILP since FHOPS 1.0.1.)

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

Remove part or all of a machine-shift’s production to simulate downtime.

Parameters:

config (DowntimeEventConfig) – Probability, duration distribution (hours), max_concurrent and role filter.

Notes

Eligible rows are assignments with assigned > 0 whose machine role passes target_machine_roles. Sampling order (deterministic for a given RNG state):

  1. Days are visited in ascending order.

  2. Selection on each day: with max_concurrent set, rng.choice draws min(max_concurrent, n) of the day’s n eligible rows without replacement (probability only acts as an on/off switch); otherwise one rng.random() per eligible row (in frame order) selects the row when <= probability.

  3. For each selected row (in selection order) one rng.normal(mean_duration_hours, std_duration_hours) draw gives the duration d, clipped to [0, shift_hours].

shift_hours follows deterministic playback (shift_hours_resolver()): the matching timeline.shifts definition, else the machine’s daily_hours divided by its number of shifts that day. With 0 < d < shift_hours the row’s production is multiplied by 1 - d / shift_hours; with d == shift_hours (or unknown shift hours) the shift is lost entirely (assigned = 0, production = 0); d == 0 leaves the row untouched. Affected rows get _downtime = 1, _downtime_hours = d and _downtime_lost (the proposed volume removed, m³) so playback summaries report the sampled downtime hours and compute_kpis() the lost volume.

Row updates are positional: the caller’s index (which may contain duplicate labels) is ignored and restored on the returned copy.

apply(context, assignments, base_production)[source]

Return a copy of assignments with downtime applied (see class notes).

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:

  • probability (float, default=0.15) – Probability in [0, 1] that an eligible machine-shift assignment is hit by downtime. Ignored for selection when max_concurrent is set (see below); 0 disables the event.

  • mean_duration_hours (float, default=4.0) – Mean of the Normal distribution used to sample the downtime duration (hours, >= 0).

  • std_duration_hours (float, default=1.5) – Standard deviation of the downtime duration (hours, >= 0). 0 makes every downtime last exactly mean_duration_hours (clipped to the shift length).

  • max_concurrent (int | None, default=None) – When set, exactly min(max_concurrent, n) of the n eligible assignments on each day are hit, chosen uniformly without replacement; when None each assignment is hit independently with probability. Must be positive when provided.

  • target_machine_roles (list[str] | None, default=None) – Restrict downtime to machines whose role is in this list (None targets all).

  • enabled (bool)

  • seed_offset (int)

Notes

The sampled duration d is clipped to [0, shift_hours] and removes the fraction d / shift_hours of the assignment’s production. A full-shift loss (d == shift_hours) cancels the assignment. See fhops.evaluation.playback.stochastic.DowntimeEvent.

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:

  • probability (float, default=0.1) – Probability in [0, 1] that a shock starts at a landing on a given calendar day of the scenario horizon (0 disables the event).

  • capacity_multiplier_range (tuple[float, float], default=(0.4, 0.8)) – Ascending (low, high) bounds (both > 0) of the uniform distribution from which each shock’s production multiplier is drawn.

  • duration_days (int, default=1) – Number of consecutive calendar days each shock lasts, starting on its start day (>= 1).

  • target_landing_ids (list[str] | None, default=None) – Restrict shocks to these landings (None uses every scenario landing).

  • enabled (bool)

  • seed_offset (int)

Notes

Every assignment on a block served by a shocked landing has its production scaled by the shock multiplier on each affected day; overlapping shocks use the minimum multiplier. See fhops.evaluation.playback.stochastic.LandingShockEvent.

Shocks start independently for each landing and calendar day, so the expected fraction of landing-days under a shock is 1 - (1 - probability) ** duration_days (days d < duration_days at the start of the horizon use d instead of duration_days). The defaults (probability=0.1, duration_days=1) shock 10 % of landing-days, with a mean multiplier of (low + high) / 2 = 0.6 on those days.

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 multi-day shocks.

Parameters:

config (LandingShockConfig) – Start probability, multiplier range, duration (days) and optional landing filter.

Notes

Sampling order (deterministic for a given RNG state): landings are visited in the order of target_landing_ids (or scenario.landings); for each landing, every calendar day 1..num_days of the scenario horizon is visited in ascending order and one rng.random() draw starts a shock when <= probability. A started shock immediately draws its multiplier with rng.uniform(low, high) and covers duration_days consecutive days starting on its start day. When shocks overlap, the minimum multiplier applies on each day.

Every still-assigned row whose block is served by a shocked landing on an affected day has its current production multiplied by the shock multiplier (so effects compose with earlier events) and records _landing_multiplier. Row updates are positional (the caller’s index is ignored and restored on the returned copy).

The expected fraction of landing-days under a shock is 1 - (1 - probability) ** duration_days (ignoring the horizon start, where fewer start days can cover a day); see LandingShockConfig.

apply(context, assignments, base_production)[source]

Return a copy of assignments with landing shocks applied (see class notes).

Parameters:
Return type:

DataFrame

sample_multipliers(context)[source]

Sample landing shocks and return the per-(landing, day) multiplier map.

Parameters:

context (SamplingContext) – Sample context providing the scenario and RNG (consumed in the documented order).

Returns:

Mapping (landing_id, day) -> multiplier for every day covered by at least one shock (minimum multiplier across overlapping shocks). Days without a shock are absent.

Return type:

dict[tuple[str, int], float]

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, downtime_hours=None)[source]

Bases: object

Atomic shift-level record produced by deterministic playback.

Notes

hours_worked is the productive time of the machine-shift (hours); production_units is the delivered volume (m³) after sequencing caps. downtime flags machine-shifts hit by stochastic downtime and downtime_hours stores the sampled hours lost (None when unknown, in which case summaries fall back to hours_worked).

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)

  • downtime_hours (float | None)

blackout_hit: bool
block_id: str | None
day: int
downtime: bool
downtime_hours: float | None
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.

Notes

delivered_total is the volume (m³) delivered by terminal roles and remaining_work_total the volume (m³) of Block.work_required still undelivered. run_playback() always sets both from the sequencing tracker; for an empty plan delivered_total == 0 and remaining_work_total == sum(Block.work_required) (the field defaults only apply to hand-built instances).

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, default=True) – Whether run_stochastic_playback() applies the event when it builds the default event list.

  • seed_offset (int, default=0) – Deprecated (FHOPS 1.0.1) and ignored: every event of sample i draws from the shared generator numpy.random.default_rng(base_seed + i) in the order downtime → weather → landing shocks (FHOPS 1.0.0 also ignored this field). Setting a non-zero value emits a DeprecationWarning; 0 (e.g. a round trip of a dumped default config) does not.

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

Scale production on days hit by sampled weather spells.

Parameters:

config (WeatherEventConfig) – Day probability, severity levels, spell length and optional shift filter.

Notes

Sampling order: for each distinct assignment day in ascending order, one rng.random() draw starts a spell when <= day_probability; a started spell draws its severity level with rng.integers. Each spell covers impact_window_days consecutive days; overlapping spells keep the maximum severity. Affected, still-assigned rows have their current production multiplied by 1 - severity (so effects compose with earlier events) and record _weather_severity and _weather_lost (the proposed volume removed, m³). correlated_days is deprecated and ignored. Row updates are positional (the caller’s index is ignored and restored on the returned copy).

apply(context, assignments, base_production)[source]

Return a copy of assignments with weather impacts applied (see class notes).

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:

  • day_probability (float, default=0.2) – Probability in [0, 1] that a weather spell starts on a given assignment day.

  • severity_levels (dict[str, float]) – Mapping label -> severity; a spell picks one level uniformly at random and scales affected production by 1 - severity (severity in [0, 1]).

  • correlated_days (bool, default=True) – Deprecated (FHOPS 1.0.1) and ignored. Explicitly setting a non-default value (False) emits a DeprecationWarning; the default True (e.g. a round trip of a dumped config or an older synthetic metadata.yaml) does not. Use impact_window_days to model multi-day spells.

  • impact_window_days (int, default=1) – Number of consecutive calendar days covered by each spell (>= 1); overlapping spells keep the highest severity per day.

  • affected_shifts (list[str] | None, default=None) – Restrict weather impacts to these shift IDs (None affects every shift).

  • enabled (bool)

  • seed_offset (int)

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 (fhops.scenario.contract.Problem) – Problem wrapping the scenario being replayed.

  • assignments (pandas.DataFrame) – Rows with machine_id, block_id, day, shift_id (optional only for single-shift scenarios, default "S1"; see normalise_shift_ids()), assigned (rows with assigned <= 0 are skipped) and production (volume, m³; when missing the production rate capped by remaining work is used). The frame index is ignored. Stochastic events may add the private columns _downtime (flag), _downtime_hours (sampled downtime hours), _downtime_lost (proposed volume removed by downtime, m³), _weather_severity and _weather_lost (proposed volume removed by weather, m³).

Returns:

Iterator over chronologically ordered records. The iterator exposes a sequencing_tracker attribute holding the SequencingTracker used to cap production.

Return type:

Iterator[PlaybackRecord]

Raises:

ValueError – When required columns are missing, or when the scenario has several shifts per day and a replayed row has no shift_id (normalise_shift_ids()).

Notes

Rows flagged by downtime keep their record even when the whole shift is lost (assigned == 0): such records carry production_units = 0, hours_worked = 0 and downtime_hours equal to the lost shift hours, and they bypass the sequencing tracker and mobilisation costing (the machine did not work). Partially lost shifts report hours_worked = shift_hours - downtime_hours. shift_hours comes from shift_hours_resolver(). When the private _downtime_lost / _weather_lost columns are present, their values are copied to metadata["downtime_production_lost"] / metadata["weather_production_lost"] (m³) for record-level loss KPIs.

A row without a production value in a slot that a scenario lock (ScheduleLock) pins to the row’s block is replayed as a locked slot (#158): it proposes min(rate, remaining) capped by ScheduleLock.production when set, produces what the staged input allows, and idles without a sequencing violation when the input or a head-start/truckload threshold is missing (as the operational MILP and the heuristics treat locked slots).

When Scenario.initial_state is set, the sequencing tracker starts from the carried-in staged inventory, role remaining volumes, and role shift counts, and each machine’s mobilisation tracking starts at its last_block_id so the first move to a different block is charged (matching the operational MILP and the heuristics).

An empty plan (assignments is None, has no rows, or has no row with assigned > 0) yields no records but still exposes a fresh sequencing_tracker, so callers see the full Block.work_required as remaining and nothing delivered. Empty frames are accepted without the required-column check.

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

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:

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.

Return type:

KPIResult

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 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 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.

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 (fhops.scenario.contract.Problem) – Problem wrapping the scenario being replayed (including any initial_state).

  • assignments (pandas.DataFrame) – Assignment rows accepted by assignments_to_records(). May be empty (no rows, no columns) or contain only assigned <= 0 rows.

  • config (PlaybackConfig | None, default=None) – Playback options; None uses PlaybackConfig defaults.

  • sample_id (int, default=0) – Identifier stamped on the summaries (stochastic playback sample index).

Returns:

Records, shift/day summaries, the tracker’s sequencing_debug snapshot, delivered_total and remaining_work_total (m³).

Return type:

PlaybackResult

Notes

An empty plan yields no records and no shift summaries; day summaries still list the available hours of every scenario day (utilisation 0). delivered_total is 0 and remaining_work_total equals sum(Block.work_required) (FHOPS ≤ 1.0.0 reported 0 remaining for an empty frame, which made a plan with no assignments look complete).

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

Run stochastic playback over multiple samples.

Parameters:

  • problem (fhops.scenario.contract.Problem) – Problem wrapping the scenario being replayed.

  • assignments (pandas.DataFrame) – Deterministic assignments (machine_id, block_id, day, optional shift_id, assigned, production).

  • sampling_config (SamplingConfig) – Number of samples, base_seed and per-event configuration.

  • events (Iterable[PlaybackEvent] | None, default=None) – Custom events applied in the given order; None builds the enabled default events in the order downtime → weather → landing shocks.

Returns:

base_result (deterministic playback of assignments) and one PlaybackSample per sample.

Return type:

EnsembleResult

Notes

Sample i seeds numpy.random.default_rng(sampling_config.base_seed + i); every event consumes that generator in turn, following the sampling order documented on DowntimeEvent, WeatherEvent and LandingShockEvent. Each sample starts from the deterministic baseline production and events compose multiplicatively.

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]

Notes

downtime_hours sums each downtime record’s sampled downtime_hours (falling back to hours_worked when the record carries no sampled value) and downtime_events counts those records.

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 (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:

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.

Return type:

KPIResult

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 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 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.