fhops.optimization Package

The optimisation stack converts fhops.scenario.contract.Problem objects into either Pyomo MIP models or heuristic schedules. Use these modules when:

  • Building or inspecting the Pyomo model (objective weights, mobilisation constraints, sequencing).

  • Running the HiGHS/Gurobi solver via fhops.optimization.mip.highs_driver.solve_mip().

  • Running simulated annealing / ILS / Tabu heuristics from fhops.optimization.heuristics.

Typical usage:

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

pb = Problem.from_scenario(load_scenario("examples/tiny7/scenario.yaml"))
result = solve_mip(pb, time_limit=300)
assignments = result["assignments"]
print(result["objective"], len(assignments))

Optimization layer (MIP builders, heuristics, constraints).

fhops.optimization.build_model(pb)[source]

Build the core FHOPS MIP model.

Parameters:

pb (Problem) – fhops.scenario.contract.Problem produced by Problem.from_scenario. The helper must include the full shift list (pb.shifts) so the model can build the (machine, block, day, shift) assignment tensor.

Returns:

Fully constructed model containing decision variables for assignments/production, optional transition binaries (when mobilisation or transition penalties are enabled), and the objective/constraints described in the FHOPS roadmap.

Return type:

pyomo.ConcreteModel

Notes

The builder purposely mirrors the documented objective weights:

  • ObjectiveWeights.production – coefficient for total production.

  • ObjectiveWeights.mobilisation – coefficient for mobilisation spend derived from transition binaries.

  • ObjectiveWeights.transitions – optional penalty for the number of transitions itself.

  • ObjectiveWeights.landing_surplus – enables soft landing-capacity overages using surplus variables.

Any change to this function should be reflected in docs/howto/thesis_eval.rst and the MIP section of the API docs.

fhops.optimization.solve_mip(pb, time_limit=60, driver='auto', debug=False)[source]

Build and solve the FHOPS MIP.

Parameters:

  • pb (Problem)

  • time_limit (int)

  • driver (str)

  • debug (bool)

Return type:

Mapping[str, object]

Pyomo builder for FHOPS MIP.

fhops.optimization.mip.builder.build_model(pb)[source]

Build the core FHOPS MIP model.

Parameters:

pb (Problem) – fhops.scenario.contract.Problem produced by Problem.from_scenario. The helper must include the full shift list (pb.shifts) so the model can build the (machine, block, day, shift) assignment tensor.

Returns:

Fully constructed model containing decision variables for assignments/production, optional transition binaries (when mobilisation or transition penalties are enabled), and the objective/constraints described in the FHOPS roadmap.

Return type:

pyomo.ConcreteModel

Notes

The builder purposely mirrors the documented objective weights:

  • ObjectiveWeights.production – coefficient for total production.

  • ObjectiveWeights.mobilisation – coefficient for mobilisation spend derived from transition binaries.

  • ObjectiveWeights.transitions – optional penalty for the number of transitions itself.

  • ObjectiveWeights.landing_surplus – enables soft landing-capacity overages using surplus variables.

Any change to this function should be reflected in docs/howto/thesis_eval.rst and the MIP section of the API docs.

MIP solver driver plumbing (HiGHS by default, optional Gurobi).

fhops.optimization.mip.highs_driver.solve_mip(pb, time_limit=60, driver='auto', debug=False)[source]

Build and solve the FHOPS MIP.

Parameters:

  • pb (Problem)

  • time_limit (int)

  • driver (str)

  • debug (bool)

Return type:

Mapping[str, object]

Simulated annealing heuristic for FHOPS.

class fhops.optimization.heuristics.sa.Schedule(plan, matrix=<factory>, mobilisation_cache=<factory>, dirty_machines=<factory>, block_remaining_cache=None, role_remaining_cache=None, dirty_blocks=<factory>, block_slots=<factory>, dirty_slots=<factory>, slot_production=<factory>, watch_stats=None)[source]

Bases: object

Machine assignment plan storing both dict and array views.

Parameters:
block_remaining_cache: dict[str, float] | None
block_slots: dict[str, list[tuple[int, str]]]
dirty_blocks: set[str]
dirty_machines: set[str]
dirty_slots: set[tuple[str, int, str]]
matrix: dict[str, list[str | None]]
mobilisation_cache: dict[str, MobilisationStats]
plan: dict[str, dict[tuple[int, str], str | None]]
role_remaining_cache: dict[tuple[str, str], float] | None
slot_production: dict[tuple[str, int, str], SlotProduction]
watch_stats: dict[str, Any] | None
fhops.optimization.heuristics.sa.solve_sa(pb, iters=2000, seed=42, operators=None, operator_weights=None, batch_size=None, max_workers=None, cooling_rate=0.999, restart_interval=None, telemetry_log=None, telemetry_context=None, watch_sink=None, watch_interval=None, watch_metadata=None, watch_debug=False, use_local_repairs=False, objective_weight_overrides=None, milp_objective=None)[source]

Solve the scheduling problem with simulated annealing.

Parameters:

  • pb (fhops.scenario.contract.Problem) – Parsed scenario context describing machines, blocks, and shifts.

  • iters (int, default=2000) – Number of annealing iterations. Higher values increase runtime and solution quality.

  • seed (int, default=42) – RNG seed used for deterministic runs.

  • operators (list[str] | None) – Optional list of operator names to enable (default: all registered operators).

  • operator_weights (dict[str, float] | None) – Optional weight overrides for operators (values <= 0 disable an operator).

  • batch_size (int | None) – When set, sample up to batch_size neighbour candidates per iteration. None or <= 1 keeps the sequential single-candidate behaviour.

  • max_workers (int | None) – Maximum worker threads for evaluating batched neighbours. None/<=1 keeps sequential scoring.

  • cooling_rate (float, default=0.999) – Multiplicative cooling factor applied each iteration (0 < rate < 1). Larger values cool more slowly.

  • restart_interval (int | None, optional) – Number of consecutive non-accepting iterations before restarting from the greedy seed. None auto-scales to max(1000, iters / 5).

  • telemetry_log (str | pathlib.Path | None) – Optional telemetry JSONL path. When provided, solver progress and final metrics are logged.

  • telemetry_context (dict[str, Any] | None) – Additional context merged into telemetry records (scenario metadata, tuner info, etc.).

  • watch_sink (SnapshotSink | None, optional) – Optional callback that receives fhops.telemetry.watch.Snapshot updates for live dashboards. When omitted, no live progress is emitted.

  • watch_interval (int | None, optional) – Iteration interval between snapshot emissions. Defaults to max(1, iters / 200) when a sink is provided.

  • watch_metadata (dict[str, str] | None) – Additional metadata (e.g., scenario/solver labels) attached to each snapshot.

  • watch_debug (bool, default=False) – When True capture sequencing debug stats for watch snapshots (adds overhead).

  • use_local_repairs (bool, default=False) – When True repairs only the slots touched by a candidate before scoring. The final schedule is always re-scored with a full repair before reporting.

  • objective_weight_overrides (dict[str, float] | None, optional) – Override scenario objective weights (keys: production, mobilisation, transitions, landing_surplus). None keeps scenario defaults, but Tiny7/Small21 scenarios auto-apply a reduced mobilisation weight to encourage exploration.

  • milp_objective (float | None, optional) – Reference MILP objective used for gap reporting (best - MILP) in watch/telemetry output. None skips gap metrics.

Returns:

Dictionary with the following keys:

objective (float)

Best objective value achieved during the run (higher is better).

assignments (pandas.DataFrame)

Assignment matrix with columns machine_id, block_id, day, shift_id, assigned.

meta (dict[str, Any])

Telemetry payload including operators weights, optional operators_stats, and bookkeeping such as proposals or telemetry_run_id.

Return type:

dict