fhops.planning Package

Rolling-horizon replanning utilities that are shared between the CLI and Python callers. Use these helpers to generate iteration plans, slice scenarios into sub-horizons, and execute rolling solves with either the simulated annealing or MILP hooks.

Typical usage:

from fhops.planning import (
    comparison_dataframe,
    compute_rolling_kpis,
    evaluate_rolling_plan,
    solve_rolling_plan,
)
from fhops.scenario.io import load_scenario
import pandas as pd

scenario = load_scenario("examples/tiny7/scenario.yaml")
result = solve_rolling_plan(
    scenario,
    master_days=7,  # tiny7 has a 7-day horizon
    subproblem_days=4,
    lock_days=3,
    solver="sa",
    sa_iters=200,
    sa_seed=123,
)
print(result.metadata, len(result.locked_assignments))

# Full-horizon baseline, e.g. from
# fhops solve-heur examples/tiny7/scenario.yaml --out tmp/tiny7_full_horizon.csv
baseline_df = pd.read_csv("tmp/tiny7_full_horizon.csv")
comparison = compute_rolling_kpis(
    scenario,
    result,
    baseline_assignments=baseline_df,
)
print(comparison.delta_totals.get("total_production_delta"))

# Or use the reporting wrapper that preserves extra metadata and deltas.
comparison = evaluate_rolling_plan(
    result,
    scenario,
    baseline_assignments=baseline_df,
    baseline_label="full_sa",
)
print(comparison.deltas.get("total_production_delta"))

# Build a plotting-friendly frame for MASc experiments.
plot_df = comparison_dataframe(comparison, metrics=["total_production", "mobilisation_cost"])

Solver options (MILP)

Pass solver-specific options (threads, gap targets, log files) via mip_solver_options on the library helper, or set environment variables for backends like Gurobi:

result = solve_rolling_plan(
    scenario,
    master_days=42,
    subproblem_days=21,
    lock_days=7,
    solver="mip",
    mip_solver="gurobi",
    mip_time_limit=600,
    mip_solver_options={"Threads": 64, "LogFile": "med42.log"},
)

mip_solver_options is also accepted by fhops.planning.get_solver_hook() for direct hook construction.

Assignments exported by fhops plan rolling (--out-assignments) can be fed directly into fhops.planning.compute_rolling_kpis() alongside a monolithic baseline DataFrame when you want KPI deltas without re-running the solver in Python.

Planning utilities beyond single-horizon solves.

This package houses helpers that assemble multi-stage planning workflows (e.g., rolling-horizon replanning). Modules here should provide both library-friendly entry points and CLI wiring so the same orchestration logic can be reused by automation scripts and user-facing commands.

class fhops.planning.MILPSolver(solver='auto', time_limit=300, solver_options=None, earliness=True, earliness_time_limit=None)[source]

Bases: object

Rolling-horizon solver hook using the operational MILP.

Parameters:

  • solver (str) – Pyomo backend to invoke (e.g., "highs", "gurobi"). "auto" (default) uses the driver’s automatic selection, Gurobi when available else HiGHS (resolve_operational_mip_solver()); the resolved name is stored in solver and reported as mip_solver in the run metadata.

  • time_limit (int) – Solve time limit in seconds for each subproblem.

  • solver_options (Mapping[str, object] | None) – Optional solver-specific options forwarded to Pyomo (e.g., {"Threads": 64} for Gurobi).

  • earliness (bool) – Break ties of the window objective in favour of early production (default True; FHOPS 1.0.1, #141): the driver’s lexicographic earliness stage (fhops.model.milp.driver.solve_operational_milp(), earliness=True) runs after the base solve, so the window’s base objective is unchanged (up to a relative 10⁻⁶) while work is not deferred past the lock span. Not applied to windows whose lock span covers the whole window (e.g. the final window, or sub == lock), where timing inside the window does not affect what is locked; a single-window run therefore stays a direct solve.

  • earliness_time_limit (int | None) – Time limit (seconds) of the earliness stage; None (default) uses time_limit, so a window may take up to twice time_limit.

Notes

The hook merges the locked_assignments it receives with scenario.locked_assignments (the operational MILP enforces them as equality constraints), keeps only rows with assigned = 1, preserves shift_id and the planned production (so carry-forward and compute_rolling_kpis() replay the MILP plan, not the full production rate), and records the wall-clock runtime of model build + solve in SolverOutput.runtime_s. Solver status and termination condition are reported as warnings (solver_status=…, termination_condition=…), followed by the driver’s solver_error (as solver_error=<text>) and each entry of its warnings list (e.g. locks the MILP pinned to zero or ignored), without duplicates; run_rolling_horizon() copies them into the iteration warnings.

A window without a solution returns SolverOutput(has_solution=False, assignments=[], objective=None): the driver reported no solution (has_solution false, or objective is None: time limit before an incumbent, infeasible model) or a solver failure (outcome="error", e.g. an unknown or unlicensed solver; the driver reports these instead of raising since #139), whose text is copied to SolverOutput.error. run_rolling_horizon() then applies its no-solution policy. Exceptions raised while building the window problem or model (FHOPS defects, invalid data) are not caught (FHOPS 1.0.1, #141; before, every exception became a no-solution window).

The window objective (SolverOutput.objective) is always the base objective; the earliness stage is reported as an earliness=<status> value=<E> ... warning.

available()[source]

Return True when SolverFactory(self.solver) reports the backend as available.

Return type:

bool

name = 'mip'
class fhops.planning.RollingCarryState(through_day, remaining_work, initial_state=None)[source]

Bases: object

Boundary state carried into a rolling-horizon window (FHOPS 1.0.1, #92).

Produced by carry_forward_state() from a deterministic replay of the stitched locked plan against the base scenario, and consumed by slice_scenario_for_window().

Parameters:

  • through_day (int)

  • remaining_work (dict[str, float])

  • initial_state (ScenarioInitialState | None)

through_day

Last base-scenario day (inclusive) covered by the replay; the next window starts at through_day + 1.

Type:

int

remaining_work

block_id -> m³ terminal volume still to deliver after through_day (the sequencing tracker’s remaining_work; values <= 1e-6 are reported as 0.0). Every base block is present; finished blocks map to 0.0. The slicer uses it as the window’s Block.work_required.

Type:

dict[str, float]

initial_state

fhops.scenario.contract.ScenarioInitialState for the next window, or None when every entry is at its default. Per explicit-system block it lists role_remaining (capped at the block’s remaining terminal volume; omitted when equal to it, the contract default), staged_inventory (omitted when zero, and for terminal roles, whose output is delivered volume rather than staged input) and role_shift_counts (omitted when zero); per machine it lists last_block_id (the block of the machine’s last locked assignment, ordered by (day, shift_id), falling back to the base initial_state).

Type:

fhops.scenario.contract.models.ScenarioInitialState | None

initial_state: ScenarioInitialState | None = None
remaining_work: dict[str, float]
through_day: int
class fhops.planning.RollingHorizonConfig(scenario, master_days, subproblem_days, lock_days, start_day=1)[source]

Bases: object

Configuration for a rolling-horizon planning run.

Parameters:

  • scenario (fhops.scenario.contract.models.Scenario) – Validated scenario to slice into rolling subproblems.

  • master_days (int) – Total number of days to cover with locked-in plans (e.g., 84 or 112).

  • subproblem_days (int) – Length of each optimisation window. Must be >= lock_days.

  • lock_days (int) – Number of days to freeze after each solve. Typically smaller than subproblem_days.

  • start_day (int) – One-indexed day in the base scenario where the rolling window begins.

lock_days: int
master_days: int
scenario: Scenario
start_day: int = 1
subproblem_days: int
exception fhops.planning.RollingInfeasibleError(message, *, iteration_index=None, partial_result=None)[source]

Bases: RuntimeError

Raised when a sub-horizon is infeasible or when solver configuration is invalid.

Parameters:

  • message (str) – Human-readable reason (names the iteration when raised inside the rolling loop).

  • iteration_index (int | None) – Zero-based iteration that failed, or None when the error is not tied to one.

  • partial_result (RollingPlanResult | None) – RollingPlanResult with the assignments locked and iterations recorded before the failure (including the failing iteration’s record when it was solved), or None. Set by run_rolling_horizon() so callers (e.g. the CLI) can still export partial outputs.

Return type:

None

class fhops.planning.RollingIterationPlan(iteration_index, start_day, horizon_days, lock_days)[source]

Bases: object

Metadata for a single rolling-horizon iteration.

Parameters:

  • iteration_index (int)

  • start_day (int)

  • horizon_days (int)

  • lock_days (int)

iteration_index

Zero-based iteration counter.

Type:

int

start_day

One-indexed start day in the base scenario.

Type:

int

horizon_days

Sub-horizon length (days) solved in this iteration.

Type:

int

lock_days

Days to freeze after the solve before advancing the window.

Type:

int

property end_day: int

Inclusive end day of the subproblem (in base-scenario coordinates).

horizon_days: int
iteration_index: int
lock_days: int
start_day: int
class fhops.planning.RollingIterationSummary(iteration_index, start_day, horizon_days, lock_days, locked_assignments, objective=None, runtime_s=None, warnings=None, remaining_work_start=None, status='solved', has_solution=True, planned_delivered=None, locked_delivered=None, empty=False, locked_production=None, error=None)[source]

Bases: object

Outcome summary for a single rolling-horizon iteration.

Parameters:

  • iteration_index (int)

  • start_day (int)

  • horizon_days (int)

  • lock_days (int)

  • locked_assignments (int)

  • objective (float | None)

  • runtime_s (float | None)

  • warnings (list[str] | None)

  • remaining_work_start (float | None)

  • status (str)

  • has_solution (bool)

  • planned_delivered (float | None)

  • locked_delivered (float | None)

  • empty (bool)

  • locked_production (float | None)

  • error (str | None)

iteration_index

Zero-based iteration counter.

Type:

int

start_day

One-indexed day in the base scenario where the sub-horizon begins.

Type:

int

horizon_days

Number of days solved in this iteration (sub-horizon length).

Type:

int

lock_days

Number of leading days frozen into the master plan after this solve.

Type:

int

locked_assignments

Count of fhops.scenario.contract.models.ScheduleLock entries injected into the master plan from this iteration (base-scenario coordinates).

Type:

int

objective

Solver objective value (units match the chosen solver) or None when not reported (and always None for windows without a solution). Window objectives include the leftover penalty on the carried remaining volume, so they are not comparable across windows or with FHOPS 1.0.0 per-window objectives; score runs on the stitched plan instead.

Type:

float | None

runtime_s

Wall-clock runtime in seconds, if the solver provides it.

Type:

float | None

warnings

Warnings surfaced by the solver hook (e.g., termination condition) plus the orchestrator’s no-solution / skipped / empty-window notices; None when there are none.

Type:

list[str] | None

remaining_work_start

Terminal volume (m³) still to deliver across all base blocks at the start of this window, from the carried-forward state (the base work_required total for the first window). None when not computed (e.g. summaries built by hand).

Type:

float | None

status

"solved" (the hook returned a plan, possibly empty), "no_solution" (the hook reported no solution: time limit before an incumbent, infeasible window, solver error) or "skipped" (nothing to plan: no blocks, no production rates, or no available shift slots; the solver is not called). Windows that are not "solved" lock no assignments.

Type:

str

has_solution

False only for status == "no_solution".

Type:

bool

planned_delivered

Terminal volume (m³) the window’s whole plan delivers when replayed on the window scenario (deterministic playback rule); 0.0 for windows that are not "solved". None when not computed.

Type:

float | None

locked_delivered

Terminal volume (m³) the locked span of this window delivers in the stitched-plan replay (remaining_work_start minus the remaining volume after the lock span). None when not computed.

Type:

float | None

empty

True when the window had remaining work (> 1e-6 m³ over its blocks) but its locked span produces nothing (no locks, or only locks without production: locked_production <= 1e-6) or its plan delivers nothing (e.g. an all-zero time-limited MILP incumbent, a plan that defers all work beyond the lock span, a no-solution or skipped window). A lock span whose blocks only open later in the window is flagged too.

Type:

bool

locked_production

Machine production (m³, every role) of the window plan’s locked span in the window replay (FHOPS 1.0.1, #141); 0.0 for windows that are not "solved". None when not computed.

Type:

float | None

error

Solver failure text of a "no_solution" window whose solver failed (SolverOutput.error), else None (#141).

Type:

str | None

empty: bool = False
error: str | None = None
has_solution: bool = True
horizon_days: int
iteration_index: int
lock_days: int
locked_assignments: int
locked_delivered: float | None = None
locked_production: float | None = None
objective: float | None = None
planned_delivered: float | None = None
remaining_work_start: float | None = None
runtime_s: float | None = None
start_day: int
status: str = 'solved'
warnings: list[str] | None = None
class fhops.planning.RollingKPIComparison(rolling_assignments, rolling_kpis, baseline_assignments=None, baseline_kpis=None, delta_totals=None)[source]

Bases: object

Comparison payload capturing rolling vs. baseline KPI metrics.

Parameters:

  • rolling_assignments (DataFrame)

  • rolling_kpis (KPIResult)

  • baseline_assignments (DataFrame | None)

  • baseline_kpis (KPIResult | None)

  • delta_totals (dict[str, float] | None)

rolling_assignments

DataFrame version of RollingPlanResult.locked_assignments suitable for playback/KPI runs.

Type:

pandas.DataFrame

rolling_kpis

KPI totals computed from the rolling plan assignments.

Type:

fhops.evaluation.metrics.kpis.KPIResult

baseline_assignments

Optional baseline schedule (full-horizon heuristic/MIP output) for comparison. An empty frame (no rows) when an empty baseline was supplied.

Type:

pandas.DataFrame | None

baseline_kpis

KPI totals computed from baseline_assignments when provided (zero-delivery KPIs for an empty baseline).

Type:

fhops.evaluation.metrics.kpis.KPIResult | None

delta_totals

Numeric difference rolling - baseline for KPI keys present in both payloads.

Type:

dict[str, float] | None

baseline_assignments: DataFrame | None = None
baseline_kpis: KPIResult | None = None
delta_totals: dict[str, float] | None = None
rolling_assignments: DataFrame
rolling_kpis: KPIResult
class fhops.planning.RollingPlanComparison(rolling_kpis, baseline_kpis, deltas, metadata)[source]

Bases: object

Bundle containing KPI totals for rolling vs. baseline plans.

Parameters:
rolling_kpis

KPI bundle computed from the locked assignments emitted by the rolling planner. Contains scalar totals plus cached shift/day calendars via fhops.evaluation.metrics.kpis.KPIResult.

Type:

KPIResult

baseline_kpis

KPI bundle computed from a full-horizon baseline such as a monolithic MILP solve. None when no baseline is supplied; an empty baseline yields zero-delivery KPIs.

Type:

KPIResult | None

deltas

Numeric KPI differences keyed by <metric>_delta and <metric>_pct_delta (when a non-zero baseline exists). Only numeric KPI entries are compared.

Type:

dict[str, float]

metadata

Copy of fhops.planning.rolling.RollingPlanResult.metadata with the additional baseline_label, assignment counts, and the empty_windows / no_solution_windows iteration counts (#117) so telemetry exports can capture context.

Type:

dict[str, object]

baseline_kpis: KPIResult | None
deltas: dict[str, float]
metadata: dict[str, object]
rolling_kpis: KPIResult
class fhops.planning.RollingPlanResult(locked_assignments, iteration_summaries, metadata, warnings=None)[source]

Bases: object

Aggregated result for a rolling-horizon run.

Parameters:

  • locked_assignments (list[ScheduleLock])

  • iteration_summaries (list[RollingIterationSummary])

  • metadata (dict[str, object])

  • warnings (list[str] | None)

locked_assignments

Locked fhops.scenario.contract.models.ScheduleLock entries rebased to the base scenario (one-indexed days).

Type:

list[fhops.scenario.contract.models.ScheduleLock]

iteration_summaries

Per-iteration summaries (objective, runtime, warnings, lock span, horizon span).

Type:

list[fhops.planning.rolling.RollingIterationSummary]

metadata

Descriptive metadata such as scenario name, master/sub/lock horizon lengths, start day, and solver identifier. Keys are JSON-serialisable so telemetry exporters can persist them.

Type:

dict[str, object]

warnings

Optional warnings accumulated across the rolling run (slicer notices plus the no-solution / skipped / empty-window notices); empty list when none are present.

Type:

list[str] | None

property empty_windows: list[int]

Indices of iterations flagged as empty (RollingIterationSummary.empty).

iteration_summaries: list[RollingIterationSummary]
locked_assignments: list[ScheduleLock]
metadata: dict[str, object]
property no_solution_windows: list[int]

Indices of iterations whose solver returned no solution.

warnings: list[str] | None = None
class fhops.planning.SolverOutput(assignments, objective=None, runtime_s=None, warnings=None, has_solution=True, error=None)[source]

Bases: object

Return type for rolling-horizon solver hooks.

Parameters:
assignments

Sequence of fhops.scenario.contract.models.ScheduleLock entries in sub-horizon coordinates (day 1 maps to the iteration start). Only the first RollingIterationPlan.lock_days will be frozen by the orchestrator.

Type:

collections.abc.Sequence[fhops.scenario.contract.models.ScheduleLock]

objective

Objective value reported by the solver or None when unavailable.

Type:

float | None

runtime_s

Wall-clock runtime in seconds, when provided by the solver.

Type:

float | None

warnings

Optional warnings emitted by the solver (solver status, termination condition, etc.).

Type:

list[str] | None

has_solution

False when the solver produced no solution (time limit before an incumbent, infeasible window, solver error). The orchestrator then ignores assignments and applies the no-solution policy (see run_rolling_horizon()). Defaults to True.

Type:

bool

error

Solver failure text (e.g. unknown or unlicensed solver) for a window without a solution; None otherwise (FHOPS 1.0.1, #141). Copied to RollingIterationSummary.error.

Type:

str | None

assignments: Sequence[ScheduleLock]
error: str | None = None
has_solution: bool = True
objective: float | None = None
runtime_s: float | None = None
warnings: list[str] | None = None
class fhops.planning.TacticalOperationalScenario(*, name, planning_level='tactical_operational', schema_version='0.1.0', parent=None, overlay_id=None, source_hash=None, economics=<factory>, periods, products, planning_units, harvest_system_options, fleet_capacity=<factory>, facilities, facility_demand=<factory>, initial_inventory=<factory>, transport_arcs=<factory>, external_supply=<factory>, roads=<factory>, road_dependencies=<factory>, block_road_access=<factory>, silviculture_transitions=<factory>, fleet_options=<factory>)[source]

Bases: BaseModel

Validated aggregate planning contract for TOPM-inspired FHOPS scenarios.

The schema uses long-form, dimension-flexible tables. Cross-validation checks all period, product, block, system-option, facility, transport, inventory, and external-supply references.

Parameters:
block_road_access: list[BlockRoadAccess]
dimension_summary()[source]

Return model-dimension counts used by CLI validation and telemetry.

Return type:

dict[str, int]

economics: Economics
external_supply: list[ExternalSupply]
facilities: list[Facility]
facility_demand: list[FacilityDemand]
fleet_capacity: list[FleetCapacity]
fleet_options: list[FleetOption]
harvest_system_options: list[HarvestSystemOption]
initial_inventory: list[InitialInventory]
model_config = {'extra': 'forbid'}

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

name: str
overlay_id: str | None
parent: str | None
periods: list[PlanningPeriod]
planning_level: str
planning_units: list[PlanningUnit]
products: list[Product]
road_dependencies: list[RoadDependency]
roads: list[RoadProject]
schema_version: str
silviculture_transitions: list[SilvicultureTransition]
source_hash: str | None
to_dict()[source]

Return a JSON-compatible round-trip representation of the scenario.

Return type:

dict[str, Any]

transport_arcs: list[TransportArc]
fhops.planning.carry_forward_state(base, locked_assignments, *, through_day)[source]

Replay a stitched locked plan on the base scenario and return the carried-forward state.

Parameters:

  • base (Scenario) – Base (unsliced) scenario. Its initial_state (when set) seeds the replay, so carried state composes with user-supplied state. Not mutated.

  • locked_assignments (Sequence[ScheduleLock]) – Stitched locked plan in base-scenario coordinates (e.g. RollingPlanResult.locked_assignments). Entries with day > through_day are ignored. shift_id and production are honoured. A lock without shift_id is a day-level lock and is replayed in every shift slot of its day in which the machine is available (operational MILP semantics, #125; FHOPS < 1.0.1 replayed it as "S1" on multi-shift days).

  • through_day (int) – Last base day (inclusive) to replay. Must be >= 0.

Returns:

Remaining terminal volume per block plus the per-block/per-machine initial state for a window starting at through_day + 1.

Return type:

RollingCarryState

Raises:

ValueError – If a day-level lock carries a planned production and its day has more than one available shift slot for the machine (the split is ambiguous).

Notes

Production follows the deterministic playback rule: each assignment proposes its lock’s planned production when set (operational-MILP rolling runs store the MILP prod value) and min(rate, block remaining) otherwise (SA runs, user locks); the shared fhops.evaluation.sequencing.SequencingTracker caps it by the role’s remaining output and the staged upstream inventory. The carried state is therefore exactly the state compute_rolling_kpis() / fhops.evaluation.compute_kpis() report for the same locked plan, and for MILP runs it is the state the window MILP planned. Role-keyed state is only emitted for blocks with an explicit harvest_system_id and for roles of that system (matching fhops.scenario.contract.validate_initial_state()).

fhops.planning.comparison_dataframe(comparison, *, metrics=None)[source]

Return a tidy DataFrame of rolling vs. baseline KPIs for plotting.

Parameters:

  • comparison (RollingPlanComparison) – KPI bundle returned by evaluate_rolling_plan(). Must contain deltas for each metric of interest when a baseline is supplied.

  • metrics (Sequence[str] | None) – Optional subset of KPI metric names to include. Defaults to all keys present in comparison.rolling_kpis.

Returns:

DataFrame with columns metric, rolling, baseline, delta, pct_delta, and baseline_label. Rows contain one metric per KPI, suitable for plotting or Markdown table generation.

Return type:

pandas.DataFrame

fhops.planning.compute_rolling_kpis(scenario, result, *, baseline_assignments=None)[source]

Compute KPI totals for a rolling plan and compare them to an optional baseline.

Parameters:

  • scenario (Scenario | Problem) – Scenario or fhops.scenario.contract.Problem describing the planning horizon. The helper converts it to a Problem for KPI evaluation when necessary.

  • result (RollingPlanResult | DataFrame | Sequence[ScheduleLock]) – Rolling plan output returned by solve_rolling_plan(), or a DataFrame/ScheduleLock sequence of locked assignments exported by the CLI (columns: machine_id, block_id, day and optional shift_id).

  • baseline_assignments (DataFrame | Sequence[ScheduleLock] | None) – Optional baseline schedule (full-horizon MILP/SA run) supplied as a Pandas DataFrame or sequence of fhops.scenario.contract.models.ScheduleLock rows. Required columns mirror the rolling assignments (machine_id, block_id, day and optional shift_id). When omitted (None), delta fields remain None. An empty DataFrame or empty sequence is evaluated as a plan with no assignments (see Notes).

Returns:

Bundle containing rolling_assignments (DataFrame), rolling_kpis and baseline_kpis (when provided), and delta_totals containing <metric>_delta and <metric>_pct_delta entries for numeric KPIs.

Return type:

RollingKPIComparison

Raises:

  • ValueError – If the rolling plan does not contain any locked assignments (a RollingPlanResult without locks, an empty DataFrame, or an empty ScheduleLock sequence).

  • TypeError – If baseline_assignments is not a DataFrame or sequence of ScheduleLock entries.

Notes

Day-level locks (shift_id None) in a RollingPlanResult or ScheduleLock sequence are expanded to every shift slot of their day in which the machine is available (carry_forward_state() semantics, #125). DataFrame rows are passed to playback as given (rows without shift_id are rejected on multi-shift scenarios).

Empty plans never look complete (#108):

  • An empty rolling plan raises ValueError (unchanged from v1.0.0) so a failed rolling run cannot be scored silently. To score it as a zero-delivery plan call fhops.evaluation.compute_kpis() with an empty frame directly.

  • An explicitly supplied empty baseline is scored as a zero-delivery plan: baseline_kpis["total_production"] == 0 and remaining_work_total equals the scenario’s total work_required. <metric>_pct_delta entries are omitted where the baseline value is zero. (v1.0.0 silently dropped an empty baseline and returned baseline_kpis=None.)

  • Rows with assigned <= 0 are kept in rolling_assignments but deliver nothing, so a DataFrame whose rows are all unassigned is scored as a zero-delivery plan.

Examples

>>> from fhops.planning import solve_rolling_plan, compute_rolling_kpis
>>> scenario = load_scenario("examples/tiny7/scenario.yaml")
>>> rolling = solve_rolling_plan(scenario, master_days=7, subproblem_days=7, lock_days=7)
>>> comparison = compute_rolling_kpis(scenario, rolling)
>>> comparison.rolling_kpis["total_production"]
5000.0  # example value
fhops.planning.evaluate_rolling_plan(result, scenario, *, baseline_assignments=None, baseline_label='single_horizon')[source]

Compute KPI and playback deltas between a rolling plan and a baseline plan.

Parameters:

  • result (RollingPlanResult) – Locked plan and iteration summaries produced by fhops.planning.solve_rolling_plan() or fhops.planning.run_rolling_horizon().

  • scenario (Scenario | Problem) – Scenario or operational problem used to evaluate the plan (typically the same scenario used to generate result).

  • baseline_assignments (DataFrame | Sequence[ScheduleLock] | None) – Optional baseline schedule to compare against. Accepts either a schedule DataFrame with machine_id, block_id, day (and optional shift_id) columns or a sequence of fhops.scenario.contract.models.ScheduleLock entries. Use a monolithic MILP/SA schedule to quantify rolling suboptimality; pass None to skip baseline deltas. An empty DataFrame/sequence (e.g. a baseline solve that found no solution) is scored as a zero-delivery plan rather than dropped.

  • baseline_label (str) – Label describing the baseline schedule (e.g., "full_mip_600s"). This value is threaded into the comparison metadata so telemetry exports remain traceable.

Returns:

KPI totals for the rolling run, baseline KPIs when provided, per-metric deltas, and merged metadata that includes assignment counts and the baseline_label.

Return type:

RollingPlanComparison

Raises:

  • ValueError – If result contains no locked assignments.

  • TypeError – If baseline_assignments is not a DataFrame or sequence of ScheduleLock items.

Notes

Empty-plan semantics follow fhops.planning.compute_rolling_kpis().

fhops.planning.get_solver_hook(name, *, sa_iters=500, sa_seed=42, mip_solver='auto', mip_time_limit=300, mip_solver_options=None, mip_earliness=True, mip_earliness_time_limit=None)[source]

Resolve a solver hook by name.

Parameters:

  • name (str) – Solver identifier ("sa", "mip"/"milp", or "stub").

  • sa_iters (int) – Number of iterations to run when name == "sa".

  • sa_seed (int) – Random seed passed to the SA hook for deterministic runs.

  • mip_solver (str) – Pyomo MILP driver to invoke when name is "mip" or "milp"; "auto" resolves to "highs".

  • mip_time_limit (int) – Solve time limit in seconds for the MILP hook.

  • mip_solver_options (Mapping[str, object] | None) – Optional solver-specific parameters forwarded to the MILP backend (e.g., {"Threads": 64}).

  • mip_earliness (bool) – Earliness tie-break in MILP windows (default True; see MILPSolver).

  • mip_earliness_time_limit (int | None) – Time limit of the earliness stage (None: mip_time_limit).

Returns:

Callable that consumes a sliced scenario and iteration plan.

Return type:

IterableSolver

Raises:

RollingInfeasibleError – If an unsupported solver name is supplied.

fhops.planning.load_tactical_operational_scenario(yaml_path)[source]

Load and validate a TOPM-inspired tactical–operational scenario.

The loader accepts either inline YAML sections or a data: mapping from section names to CSV files. topm-mini-style specifications may use fixture_id/specification_version; those aliases are normalized to name and schema_version before Pydantic validation.

Parameters:

yaml_path (str | Path)

Return type:

TacticalOperationalScenario

fhops.planning.rolling_assignments_dataframe(result, *, include_metadata=False, scenario=None)[source]

Return locked assignments as a DataFrame for playback/KPI workflows.

Parameters:

  • result (RollingPlanResult) – Rolling plan output produced by run_rolling_horizon() or solve_rolling_plan().

  • include_metadata (bool) – When True append run metadata (scenario, horizons, solver label) to each row so exports remain self-describing.

  • scenario (Scenario | Problem | None) – Optional base scenario (or Problem). When given, day-level locks (shift_id None) are expanded to one row per shift slot of the day in which the machine is available, as in carry_forward_state(). Plans returned by run_rolling_horizon() are already shift-level; this matters for hand-built results.

Returns:

Columns machine_id, block_id, day, shift_id, assigned (always 1), production (only when locks carry planned production, e.g. MILP rolling runs; m³), and optionally the metadata keys, sorted by day, shift_id, machine_id, block_id. shift_id is None for day-level locks when scenario is omitted.

Return type:

pandas.DataFrame

Raises:

ValueError – With scenario: a day-level lock with a planned production covers several shift slots.

Notes

The resulting frame can be passed directly to fhops.evaluation.compute_kpis() or fhops.evaluation.playback.run_playback() to evaluate the rolling plan in the same way as a monolithic solve. Playback rejects rows without shift_id on multi-shift scenarios (and treats them as "S1" otherwise), so pass scenario for plans with day-level locks. Evaluating the frame against the base scenario reproduces exactly the state carried between windows (see carry_forward_state()).

fhops.planning.run_rolling_horizon(config, solver, *, max_iterations=None, solver_name=None, fail_on_empty_window=False, fail_on_no_solution=False)[source]

Execute the rolling-horizon loop with a user-supplied solver hook.

The solver hook is responsible for producing assignments for each subproblem. This orchestrator handles window planning, scenario slicing, lock rebasing, state carry-forward, and aggregation of locked decisions. Only the first lock_days of each iteration are frozen; the remainder of the sub-horizon is discarded when rolling forward.

Before every iteration after the first, the stitched locked plan is replayed on the base scenario through day start_day - 1 (carry_forward_state()), and the window is sliced with the resulting remaining volumes and initial_state (slice_scenario_for_window()). The first window uses the base work_required and initial_state unchanged, so a single-window run (subproblem_days == lock_days == master_days) is a direct solve.

Parameters:

  • config (RollingHorizonConfig) – Rolling-horizon configuration describing master/sub/lock horizons.

  • solver (IterableSolver) – Callable that accepts a sliced scenario, the iteration plan, and the window’s locks (locked_assignments=: the base scenario’s user locks that fall in the window, rebased to window days; identical to scenario.locked_assignments) and returns a SolverOutput whose assignments are window-day ScheduleLock entries. Hooks should set ScheduleLock.shift_id; an assignment without it is a day-level lock and is expanded to one lock per shift slot of that day in which the machine is available (MILP semantics, #125); a day-level assignment with a planned production on a day with several such slots raises ValueError. Hooks report a failed solve with SolverOutput(has_solution=False).

  • max_iterations (int | None) – Optional guard to cap the number of iterations (useful for smoke tests); None runs every window. Must be >= 1 when given.

  • solver_name (str | None) – Optional solver label to persist into RollingPlanResult.metadata.

  • fail_on_empty_window (bool) – When True, raise RollingInfeasibleError at the first window whose solver returned no solution or that was solved to an empty plan (RollingIterationSummary.empty: its lock span produces nothing or its plan delivers nothing while its blocks hold work), instead of recording it and continuing (FHOPS 1.0.1, #141; before, only windows without a solution raised). Skipped windows never raise. Default False.

  • fail_on_no_solution (bool) – When True, raise RollingInfeasibleError at the first window whose solver returned no solution (the pre-#141 meaning of fail_on_empty_window). Default False.

Returns:

Locked assignments in base-scenario coordinates (shift_id preserved; day-level hook output expanded to shift-level locks) plus per-iteration summaries (status, has_solution, planned_delivered, locked_delivered, empty). warnings lists slicer notices (e.g. a carried last_block_id that falls outside a window) and the no-solution / skipped / empty-window notices.

Return type:

RollingPlanResult

Raises:

RollingInfeasibleError – If a user lock lies outside its block’s earliest_start/latest_finish window (checked up front for every lock, independent of the window settings), or, with fail_on_empty_window=True / fail_on_no_solution=True, at the first window that triggers it. The exception’s partial_result holds the run up to the failure. Any other exception raised inside the loop (e.g. by a custom hook) propagates unchanged with a best-effort partial_result attribute attached.

Notes

No-solution policy (default): when the hook returns has_solution=False (the MILP hook does so for a time limit before any incumbent, an infeasible window, or a solver exception), the iteration is recorded with status="no_solution", objective=None and a warning; no assignments are locked for its lock span (machines idle; user locks in that span are not applied either), so the carried state is unchanged across the span, and the next window is solved from it. Windows with nothing to plan (no blocks, no production rates, or no available shift slots, e.g. days outside a partial shift_calendar) are not passed to the solver and are recorded with status="skipped". A window is flagged empty (warning, counted in summarize_plan()["empty_windows"]) when its blocks still hold work but it locked no assignments or its plan delivers no volume. Policy warnings are also emitted as UserWarning.

fhops.planning.slice_scenario_for_window(base, window, locked_assignments=None, *, carry_state=None)[source]

Return a horizon-trimmed scenario for the given iteration window.

The slice rebases day indices so the window start maps to day 1: machine and shift calendars, timeline blackouts, block windows, and locks outside the window are dropped and the rest are shifted. Blocks whose [earliest_start, latest_finish] window does not overlap the sub-horizon are dropped together with their production rates and mobilisation distances. When the base scenario has a shift_calendar but none of its entries fall inside the window, the window receives explicit available=0 entries so it has no shift slots (machines unavailable), exactly like the same days in a direct solve; an empty calendar would otherwise fall back to the timeline shifts or S1 and invent capacity (FHOPS 1.0.0 behaviour, fixed in #117).

Parameters:

  • base (Scenario) – Original scenario to slice. This object is not mutated.

  • window (RollingIterationPlan) – Iteration window describing the start day and sub-horizon length.

  • locked_assignments (Sequence[ScheduleLock] | None) – Optional extra locks (base-scenario coordinates) merged with base.locked_assignments (FHOPS <= 1.0.0 replaced the scenario’s locks instead). Exact duplicates are collapsed; conflicting locks are rejected by Scenario validation. Only locks inside the window are retained (day-rebased, shift_id preserved).

  • carry_state (RollingCarryState | None) – Optional RollingCarryState (see carry_forward_state()). When given, each block’s work_required is set to carry_state.remaining_work (finished blocks stay with 0) and initial_state is set to carry_state.initial_state. When omitted, the base work_required and initial_state are kept (window 0 / direct-solve behaviour).

Returns:

A re-validated deep copy with num_days == window.horizon_days. Initial-state entries for dropped blocks are removed, as are last_block_id values that point at dropped blocks (a UserWarning is emitted for the latter because the boundary move is then free in the window solve).

Return type:

Scenario

Raises:

  • RollingInfeasibleError – If a lock inside the window references a block that is outside the window. This only happens for locks outside their block’s earliest_start/latest_finish window, which run_rolling_horizon() rejects up front.

  • ValueError – (Pydantic ValidationError) if the merged locks conflict.

fhops.planning.solve_rolling_plan(scenario, *, master_days, subproblem_days, lock_days, solver='sa', sa_iters=500, sa_seed=42, mip_solver='auto', mip_time_limit=300, mip_solver_options=None, max_iterations=None, fail_on_empty_window=False, fail_on_no_solution=False, mip_earliness=True, mip_earliness_time_limit=None)[source]

Library-facing helper to execute a rolling-horizon plan.

Parameters:

  • scenario (Scenario) – Validated scenario to slice into rolling subproblems.

  • master_days (int) – Total number of days to lock in across the rolling run. Must satisfy master_days + start_day - 1 <= scenario.num_days.

  • subproblem_days (int) – Number of days solved per iteration (≥ lock_days).

  • lock_days (int) – Number of leading days to freeze after each iteration (≤ subproblem_days).

  • solver (str) – Solver hook to use ("sa", "mip", "milp", or "stub").

  • sa_iters (int) – Simulated annealing iteration budget when solver == "sa".

  • sa_seed (int) – Random seed for SA runs to keep results deterministic across iterations.

  • mip_solver (str) – Pyomo MILP driver name when solver is MILP-backed; "auto" uses the driver’s automatic selection (Gurobi when available, otherwise HiGHS).

  • mip_time_limit (int) – Time limit in seconds for each MILP subproblem solve.

  • mip_solver_options (Mapping[str, object] | None) – Optional solver-specific parameters forwarded to the MILP backend (e.g., {"Threads": 64}).

  • max_iterations (int | None) – Optional guard to cap the number of iterations (>= 1; useful for smoke tests).

  • fail_on_empty_window (bool) – Raise RollingInfeasibleError at the first window without a solution or solved to an empty plan instead of recording it and continuing (default False; see run_rolling_horizon()).

  • fail_on_no_solution (bool) – Raise RollingInfeasibleError at the first window without a solution only (default False).

  • mip_earliness (bool) – Earliness tie-break in MILP windows (default True; see MILPSolver).

  • mip_earliness_time_limit (int | None) – Time limit of the earliness stage in seconds (None: mip_time_limit).

Returns:

Locked assignments, per-iteration summaries (with status / empty flags), and metadata describing the run.

Return type:

RollingPlanResult

Raises:

  • ValueError – If horizon parameters violate basic bounds (e.g., master horizon exceeds scenario length) or max_iterations < 1.

  • RollingInfeasibleError – If the solver name is unsupported, a user lock lies outside its block’s time window, or a window triggers fail_on_empty_window / fail_on_no_solution.

fhops.planning.summarize_plan(result)[source]

Return a JSON-serialisable summary of a rolling-horizon run.

Parameters:

result (RollingPlanResult) – Rolling plan result with locked assignments and per-iteration summaries.

Returns:

JSON-serialisable mapping with keys:

iterations

One record per iteration: iteration_index, start_day, horizon_days, lock_days, locked_assignments, objective, runtime_s, warnings, remaining_work_start, status, has_solution, planned_delivered, locked_delivered, locked_production, empty and error (see RollingIterationSummary).

total_locked_assignments

Number of locked assignments in the stitched plan.

empty_windows / no_solution_windows / skipped_windows

Counts of iterations flagged empty, without a solution, and skipped.

empty_window_indices / no_solution_window_indices

The corresponding iteration indices.

metadata / warnings

Run metadata and accumulated warnings.

Return type:

dict

fhops.planning.tactical_scenario_dimensions(scenario)[source]

Return model dimension counts for CLI output and telemetry.

Parameters:

scenario (TacticalOperationalScenario)

Return type:

dict[str, int]

fhops.planning.tactical_scenario_to_dict(scenario)[source]

Serialize a tactical–operational scenario to a JSON-compatible dictionary.

Parameters:

scenario (TacticalOperationalScenario)

Return type:

dict[str, Any]