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=14,
    subproblem_days=7,
    lock_days=7,
    solver="sa",
    sa_iters=200,
    sa_seed=123,
)
print(result.metadata, len(result.locked_assignments))

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.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[source]

Bases: RuntimeError

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

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)[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)

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.

Type:

float | None

runtime_s

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

Type:

float | None

warnings

Optional warnings surfaced by the solver hook (e.g., termination condition).

Type:

list[str] | None

horizon_days: int
iteration_index: int
lock_days: int
locked_assignments: int
objective: float | None = None
runtime_s: float | None = None
start_day: int
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.

Type:

pandas.DataFrame | None

baseline_kpis

KPI totals computed from baseline_assignments when provided.

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 a baseline DataFrame or lock list is not supplied.

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 and assignment counts 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; empty list when none are present.

Type:

list[str] | None

iteration_summaries: list[RollingIterationSummary]
locked_assignments: list[ScheduleLock]
metadata: dict[str, object]
warnings: list[str] | None = None
class fhops.planning.SolverOutput(assignments, objective=None, runtime_s=None, warnings=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

assignments: Sequence[ScheduleLock]
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.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, delta fields remain None.

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.

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

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.

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

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

fhops.planning.get_solver_hook(name, *, sa_iters=500, sa_seed=42, mip_solver='auto', mip_time_limit=300, mip_solver_options=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".

  • 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}).

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

Returns:

Columns include machine_id, block_id, day, and optionally the metadata keys.

Return type:

pandas.DataFrame

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. Shift identifiers default to "S1" downstream when omitted.

fhops.planning.run_rolling_horizon(config, solver, *, max_iterations=None, solver_name=None)[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, 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.

Parameters:

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

  • solver (IterableSolver) – Callable that accepts a sliced scenario, the iteration plan, and the current locked assignments (rebased to the sub-horizon) and returns an iterable of ScheduleLock entries for that subproblem, plus optional metadata.

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

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

Returns:

Locked assignments in base-scenario coordinates plus per-iteration summaries.

Return type:

RollingPlanResult

Raises:

RollingInfeasibleError – If a sub-horizon is empty or violates basic feasibility checks before solving.

fhops.planning.slice_scenario_for_window(base, window, locked_assignments=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, filters calendars/locks outside the window, and clamps block availability to the sub-horizon. It also trims mobilisation distances to the surviving block set to satisfy scenario validation.

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 locked assignments to inject; only those falling inside the window are retained and day-rebased.

Returns:

A deep-copied scenario with num_days == window.horizon_days and calendars/locks rebased to start at day 1.

Return type:

Scenario

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

  • 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 (useful for smoke tests).

Returns:

Locked assignments, per-iteration summaries, and metadata describing the run.

Return type:

RollingPlanResult

Raises:

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

  • RollingInfeasibleError – If the solver name is unsupported or a subproblem fails basic feasibility checks.

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:

Dictionary containing iteration records, total locked assignments, and warnings. Suitable for emitting as JSON/CSV telemetry in CLI helpers.

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]