fhops.hauling Package

Pydantic contracts for operational log hauling.

The hauling layer is an optional extension to fhops.scenario.contract.Scenario. It adds the entities that the harvesting and tactical-operational models do not carry: log sorts, contractors, truck types and trucks, depots, sort yards, haul arcs, loader location bindings, and the driver policy from Ghotb (2023).

The contract is deliberately a data layer. It validates identifiers, units, ranges, and cross references; it does not build or solve an optimisation model. The Phase I flow model, Phase II scheduling model, and Phase III simulated-annealing solver live in sibling modules.

References

Ghotb, S. (2023). Optimization of Log Logistics at the Operational Level Considering Sorting Decisions and Synchronization Requirements. PhD thesis, University of British Columbia. <https://hdl.handle.net/2429/85563>

class fhops.hauling.contract.Barge(*, id, contractor_id, barge_type_id, enabled=True, notes=None)[source]

Bases: BaseModel

An individual barge or vessel owned by a contractor.

Parameters:

  • id (str)

  • contractor_id (str)

  • barge_type_id (str)

  • enabled (bool)

  • notes (str | None)

id

Unique barge identifier. Phase II uses it as the vessel label.

Type:

str

contractor_id

Contractor that owns and dispatches the barge.

Type:

str

barge_type_id

Barge type that determines capacity, cost, and load compatibility.

Type:

str

enabled

Set False to keep the barge in the contract but exclude it from a solve.

Type:

bool

notes

Optional free-form note surfaced in reports.

Type:

str | None

barge_type_id: str
contractor_id: str
enabled: bool
id: str
model_config = {}

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

notes: str | None
class fhops.hauling.contract.BargeType(*, id, capacity_m3, fixed_cost=0.0, variable_cost_per_minute=0.0, overtime_cost_per_minute=0.0, load_minutes=0.0, unload_minutes=0.0, can_carry_sorted=True, can_carry_unsorted=True)[source]

Bases: BaseModel

A homogeneous barge/vessel type used on a water haul leg.

Parameters:

  • id (str)

  • capacity_m3 (float)

  • fixed_cost (float)

  • variable_cost_per_minute (float)

  • overtime_cost_per_minute (float)

  • load_minutes (float)

  • unload_minutes (float)

  • can_carry_sorted (bool)

  • can_carry_unsorted (bool)

id

Unique barge-type identifier.

Type:

str

capacity_m3

Cargo capacity in cubic metres per barge trip. Must be greater than zero.

Type:

float

fixed_cost

Fixed cost of dispatching a barge for the day, in scenario currency. Non-negative. Phase II uses it; Phase I charges only variable, handling, and trip costs.

Type:

float

variable_cost_per_minute

Variable cost per minute of barge working time. Non-negative.

Type:

float

overtime_cost_per_minute

Overtime cost per barge minute. Non-negative.

Type:

float

load_minutes

Default loading time at the origin port, in minutes. Non-negative.

Type:

float

unload_minutes

Default unloading time at the destination port, in minutes. Non-negative.

Type:

float

can_carry_sorted

Whether the barge may carry roadside-sorted loads. Defaults to True.

Type:

bool

can_carry_unsorted

Whether the barge may carry unsorted loads. Defaults to True.

Type:

bool

can_carry_sorted: bool
can_carry_unsorted: bool
capacity_m3: float
fixed_cost: float
id: str
load_minutes: float
model_config = {}

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

overtime_cost_per_minute: float
unload_minutes: float
variable_cost_per_minute: float
class fhops.hauling.contract.ContractCompatibility(*, contractor_id, block_id=None, truck_type_id=None, log_sort_id=None, allowed=True)[source]

Bases: BaseModel

Explicit compatibility override for a contractor, block, truck type, or log sort.

The first release derives most compatibility from ownership and truck-type flags. This model is reserved for later exceptions and is not required. When allowed is False, the matching combination is refused; when True, it is explicitly allowed. At least one target field must be present.

Parameters:

  • contractor_id (str)

  • block_id (str | None)

  • truck_type_id (str | None)

  • log_sort_id (str | None)

  • allowed (bool)

allowed: bool
block_id: str | None
contractor_id: str
log_sort_id: str | None
model_config = {}

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

truck_type_id: str | None
class fhops.hauling.contract.Contractor(*, id, name=None, preferred_daily_truckloads=None)[source]

Bases: BaseModel

A log-trucking contractor.

Parameters:

  • id (str)

  • name (str | None)

  • preferred_daily_truckloads (float | None)

id

Unique contractor identifier referenced by trucks and block eligibility.

Type:

str

name

Optional human-readable label.

Type:

str | None

preferred_daily_truckloads

Desired number of truckloads per day used by the Phase I social objective (Ghotb’s phi_o). None lets Phase I derive the preferred level from the cost-only solve.

Type:

float | None

id: str
model_config = {}

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

name: str | None
preferred_daily_truckloads: float | None
class fhops.hauling.contract.ContractorBlock(*, contractor_id, block_id)[source]

Bases: BaseModel

Eligibility of a contractor to serve a block.

When the contract contains no contractor_blocks entries, every contractor may serve every block. When it contains any entry, the list is treated as an allowlist: only the listed contractor/block pairs are eligible. This mirrors Ghotb’s alpha_bo parameter.

Parameters:

  • contractor_id (str)

  • block_id (str)

block_id: str
contractor_id: str
model_config = {}

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

class fhops.hauling.contract.Depot(*, id, name=None)[source]

Bases: BaseModel

A central depot where trucks start and end their daily routes.

Parameters:
id

Unique depot identifier. It is also the location key used by DepotLeg.

Type:

str

name

Optional human-readable label.

Type:

str | None

id: str
model_config = {}

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

name: str | None
class fhops.hauling.contract.DepotLeg(*, depot_id, location_id, truck_type_id, travel_minutes)[source]

Bases: BaseModel

Travel time between a depot and a pickup or delivery location.

Parameters:

  • depot_id (str)

  • location_id (str)

  • truck_type_id (str)

  • travel_minutes (float)

depot_id

Depot identifier.

Type:

str

location_id

Block or sort-yard identifier. The contract validates it against the scenario’s blocks and the hauling contract’s yards.

Type:

str

truck_type_id

Truck type for which the travel time applies.

Type:

str

travel_minutes

One-way travel time in minutes. Must be greater than zero.

Type:

float

depot_id: str
location_id: str
model_config = {}

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

travel_minutes: float
truck_type_id: str
class fhops.hauling.contract.DriverPolicy(*, regular_minutes=510.0, max_overtime_minutes=150.0, overtime_violation_penalty_per_minute=0.0)[source]

Bases: BaseModel

Regular-time, overtime, and overtime-violation policy for truck drivers.

Parameters:

  • regular_minutes (float)

  • max_overtime_minutes (float)

  • overtime_violation_penalty_per_minute (float)

regular_minutes

Regular working minutes per truck per day. The thesis case uses 510 (8.5 h).

Type:

float

max_overtime_minutes

Maximum overtime that may be assigned before a violation penalty applies. The thesis case uses 150 (2.5 h).

Type:

float

overtime_violation_penalty_per_minute

Penalty per minute of overtime above max_overtime_minutes. Phase III may use this as a soft penalty; Phase II treats the limit as a hard constraint.

Type:

float

max_overtime_minutes: float
model_config = {}

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

overtime_violation_penalty_per_minute: float
regular_minutes: float
class fhops.hauling.contract.HaulArc(*, block_id, yard_id, truck_type_id, travel_minutes, distance_km=None, return_travel_minutes=None, mode=TransportMode.ROAD)[source]

Bases: BaseModel

A feasible one-way road haul leg from a block to a sort yard for one truck type.

Water legs are not represented as HaulArc records. Declare a WaterArc, Port pair, BargeType, and MarineRoute instead. The mode field is explicit so a report can distinguish road arcs without inferring the mode from identifiers.

Parameters:

  • block_id (str)

  • yard_id (str)

  • truck_type_id (str)

  • travel_minutes (float)

  • distance_km (float | None)

  • return_travel_minutes (float | None)

  • mode (TransportMode)

block_id

Origin block identifier.

Type:

str

yard_id

Destination sort-yard identifier.

Type:

str

truck_type_id

Truck type that can traverse the arc.

Type:

str

travel_minutes

Loaded travel time in minutes, including any arc-specific delay but excluding loading and unloading. Must be greater than zero.

Type:

float

distance_km

Optional distance for reporting and cost checks. Non-negative.

Type:

float | None

return_travel_minutes

Optional empty return time from the yard to the block. When None, Phase II uses travel_minutes for the empty leg as well.

Type:

float | None

mode

Transport mode. Only road is accepted here; water and rail legs use their dedicated models.

Type:

TransportMode

block_id: str
distance_km: float | None
empty_travel_minutes()[source]

Return the empty return time from the yard to the block.

Return type:

float

mode: TransportMode
model_config = {}

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

return_travel_minutes: float | None
travel_minutes: float
truck_type_id: str
yard_id: str
class fhops.hauling.contract.HaulingConfig(*, planning_horizon_days=20, start_day=1, log_sorts, contractors, truck_types, trucks, depots, sort_yards, depot_legs=<factory>, loader_locations=<factory>, supply, demand=<factory>, haul_arcs, barge_types=<factory>, barges=<factory>, ports=<factory>, water_arcs=<factory>, port_legs=<factory>, marine_routes=<factory>, contractor_blocks=<factory>, compatibility=<factory>, driver_policy=<factory>, roadside_inventory_penalty_per_m3=0.0, port_inventory_penalty_per_m3=0.0, loader_travel_cost_per_km=0.0, default_loader_service_minutes=20.0)[source]

Bases: BaseModel

Optional hauling: section of a FHOPS scenario.

The config carries Ghotb’s operational log-hauling entities. It is validated against the parent scenario so every block, machine, and horizon reference is real. It does not require the tactical-operational contract; product_id and facility_id links are optional.

Parameters:

  • planning_horizon_days (int)

  • start_day (int)

  • log_sorts (list[LogSort])

  • contractors (list[Contractor])

  • truck_types (list[TruckType])

  • trucks (list[Truck])

  • depots (list[Depot])

  • sort_yards (list[SortYard])

  • depot_legs (list[DepotLeg])

  • loader_locations (list[LoaderLocation])

  • supply (list[HaulingSupply])

  • demand (list[HaulingDemand])

  • haul_arcs (list[HaulArc])

  • barge_types (list[BargeType])

  • barges (list[Barge])

  • ports (list[Port])

  • water_arcs (list[WaterArc])

  • port_legs (list[PortLeg])

  • marine_routes (list[MarineRoute])

  • contractor_blocks (list[ContractorBlock])

  • compatibility (list[ContractCompatibility])

  • driver_policy (DriverPolicy)

  • roadside_inventory_penalty_per_m3 (float)

  • port_inventory_penalty_per_m3 (float)

  • loader_travel_cost_per_km (float)

  • default_loader_service_minutes (float)

planning_horizon_days

Number of working days in the Phase I horizon. Defaults to 20 (four weeks).

Type:

int

start_day

One-indexed day of the parent scenario where the hauling horizon starts.

Type:

int

log_sorts / contractors / truck_types / trucks / depots / sort_yards

Core entity lists.

depot_legs

Optional travel times from each depot to blocks and yards, by truck type.

Type:

list[DepotLeg]

loader_locations

Bindings from FHOPS Machine records to pickup or delivery locations.

Type:

list[LoaderLocation]

supply / demand / haul_arcs

The Phase I flow inputs.

haul_arcs

The Phase I road flow inputs.

Type:

list[HaulArc]

barge_types / barges

Marine equipment. Barges are first-class vessels, never aliases of truck types.

ports / water_arcs / port_legs / marine_routes

The marine path: terminal nodes, sailing legs, road drayage legs, and complete block-to-yard chains with a water segment.

contractor_blocks

Optional contractor-to-block allowlist. When empty, all contractors may serve all blocks.

Type:

list[ContractorBlock]

compatibility

Optional explicit exceptions. Reserved for later phases.

Type:

list[ContractCompatibility]

driver_policy

Regular and overtime rules for Phase II and Phase III.

Type:

DriverPolicy

roadside_inventory_penalty_per_m3

Penalty per cubic metre of sorted or unsorted volume left at a block at the end of the Phase I horizon (Ghotb’s P). Non-negative.

Type:

float

port_inventory_penalty_per_m3

Penalty per cubic metre of log volume left at a port at the end of the Phase I horizon. Non-negative. Zero lets the model store volume at a terminal for free.

Type:

float

loader_travel_cost_per_km

Loader travel cost per kilometre (Ghotb’s DC_l). The thesis case uses stationary loaders, so this defaults to zero. Non-negative.

Type:

float

default_loader_service_minutes

Fallback loading/unloading service time when LoaderLocation.service_minutes is omitted. Must be greater than zero.

Type:

float

barge_ids()[source]

Return the set of barge identifiers.

Return type:

set[str]

barge_type(barge_type_id)[source]

Return a barge type by id.

Raises:

KeyError – When barge_type_id is not defined.

Parameters:

barge_type_id (str)

Return type:

BargeType

barge_type_ids()[source]

Return the set of barge-type identifiers.

Return type:

set[str]

barge_types: list[BargeType]
barges: list[Barge]
barges_for(contractor_id, barge_type_id)[source]

Return enabled barges of one contractor and type.

Parameters:

  • contractor_id (str)

  • barge_type_id (str)

Return type:

list[Barge]

compatibility: list[ContractCompatibility]
contractor_blocks: list[ContractorBlock]
contractor_can_serve_block(contractor_id, block_id)[source]

Return whether a contractor may serve a block under the allowlist rule.

Parameters:

  • contractor_id (str)

  • block_id (str)

Return type:

bool

contractor_ids()[source]

Return the set of contractor identifiers.

Return type:

set[str]

contractors: list[Contractor]
default_loader_service_minutes: float
demand: list[HaulingDemand]
depot_ids()[source]

Return the set of depot identifiers.

Return type:

set[str]

depot_legs: list[DepotLeg]
depots: list[Depot]
driver_policy: DriverPolicy
haul_arcs: list[HaulArc]
loader_locations: list[LoaderLocation]
loader_travel_cost_per_km: float
log_sort_ids()[source]

Return the set of log-sort identifiers.

Return type:

set[str]

log_sorts: list[LogSort]
marine_route(route_id)[source]

Return a marine route by id.

Raises:

KeyError – When route_id is not defined.

Parameters:

route_id (str)

Return type:

MarineRoute

marine_route_ids()[source]

Return the set of marine-route identifiers.

Return type:

set[str]

marine_routes: list[MarineRoute]
model_config = {}

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

planning_horizon_days: int
port(port_id)[source]

Return a port by id.

Raises:

KeyError – When port_id is not defined.

Parameters:

port_id (str)

Return type:

Port

port_ids()[source]

Return the set of port identifiers.

Return type:

set[str]

port_inventory_penalty_per_m3: float
port_leg_for(route, side)[source]

Return the drayage leg for one side of a marine route, if configured.

Parameters:

  • route (MarineRoute)

  • side (PortSide)

Return type:

PortLeg | None

port_legs: list[PortLeg]
ports: list[Port]
roadside_inventory_penalty_per_m3: float
sort_yards: list[SortYard]
start_day: int
supply: list[HaulingSupply]
truck_ids()[source]

Return the set of truck identifiers.

Return type:

set[str]

truck_type_ids()[source]

Return the set of truck-type identifiers.

Return type:

set[str]

truck_types: list[TruckType]
trucks: list[Truck]
trucks_for(contractor_id, truck_type_id)[source]

Return enabled trucks of one contractor and type.

Parameters:

  • contractor_id (str)

  • truck_type_id (str)

Return type:

list[Truck]

validate_against_scenario(*, block_ids, machine_ids, num_days)[source]

Validate references to the parent scenario and its horizon.

Parameters:

  • block_ids (set[str]) – Every block identifier defined by the scenario.

  • machine_ids (set[str]) – Every machine identifier defined by the scenario (loaders are machines).

  • num_days (int) – Parent scenario horizon length.

Raises:

ValueError – When a block, loader, or horizon reference is invalid.

Return type:

None

water_arc(water_arc_id)[source]

Return a water arc by id.

Raises:

KeyError – When water_arc_id is not defined.

Parameters:

water_arc_id (str)

Return type:

WaterArc

water_arcs: list[WaterArc]
yard_ids()[source]

Return the set of sort-yard identifiers.

Return type:

set[str]

class fhops.hauling.contract.HaulingDemand(*, yard_id, log_sort_id, day, volume_m3)[source]

Bases: BaseModel

Demand for one log sort at one yard on one day of the hauling horizon.

Parameters:
yard_id

Sort-yard identifier from the hauling contract.

Type:

str

log_sort_id

Log-sort identifier from the hauling contract.

Type:

str

day

One-indexed day relative to the hauling horizon (not the parent scenario horizon).

Type:

int

volume_m3

Required volume in cubic metres. Non-negative.

Type:

float

day: int
log_sort_id: str
model_config = {}

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

volume_m3: float
yard_id: str
class fhops.hauling.contract.HaulingSupply(*, block_id, log_sort_id, sorted_m3=0.0, unsorted_m3=0.0)[source]

Bases: BaseModel

Available sorted and unsorted volume of one log sort at one block.

Parameters:
block_id

Block identifier from Scenario.blocks.

Type:

str

log_sort_id

Log-sort identifier from the hauling contract.

Type:

str

sorted_m3

Roadside-sorted supply in cubic metres (Ghotb’s SW_bs). Non-negative.

Type:

float

unsorted_m3

Unsorted supply in cubic metres (SC_bs). Non-negative.

Type:

float

block_id: str
log_sort_id: str
model_config = {}

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

sorted_m3: float
unsorted_m3: float
class fhops.hauling.contract.LoaderLocation(*, loader_id, location_id, location_kind, service_minutes=None)[source]

Bases: BaseModel

Binding between a FHOPS machine (the loader) and a hauling location.

Parameters:

  • loader_id (str)

  • location_id (str)

  • location_kind (LoaderLocationKind)

  • service_minutes (float | None)

loader_id

Machine identifier from Scenario.machines. The machine is the physical loader.

Type:

str

location_id

Block identifier for a pickup loader or sort-yard identifier for a delivery loader.

Type:

str

location_kind

pickup or delivery; must agree with location_id.

Type:

LoaderLocationKind

service_minutes

Optional loading/unloading service time override. When None, the model uses the value configured by Phase II. Must be greater than zero when supplied.

Type:

float | None

loader_id: str
location_id: str
location_kind: LoaderLocationKind
model_config = {}

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

service_minutes: float | None
class fhops.hauling.contract.LoaderLocationKind(*values)[source]

Bases: StrEnum

Whether a loader serves a pickup, delivery, or port/terminal node.

PORT loaders serve both the road drayage and the barge load/unload activity at a port.

DELIVERY = 'delivery'
PICKUP = 'pickup'
PORT = 'port'
class fhops.hauling.contract.LogSort(*, id, product_id=None, sorted=True, description=None)[source]

Bases: BaseModel

A log assortment handled by the hauling network.

Parameters:

  • id (str)

  • product_id (str | None)

  • sorted (bool)

  • description (str | None)

id

Unique log-sort identifier referenced by supply, demand, and compatibility records.

Type:

str

product_id

Optional link to a tactical-operational Product.product_id. The hauling layer does not require the tactical contract; when the link is present it must be non-empty.

Type:

str | None

sorted

True for a roadside-sorted sort, False for an unsorted sort that is processed at a yard. Phase I carries both sorted and unsorted supply for every sort (Ghotb’s SW and SC parameters); this flag describes the kind of load, not a restriction on supply.

Type:

bool

description

Optional human-readable label.

Type:

str | None

description: str | None
id: str
model_config = {}

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

product_id: str | None
sorted: bool
class fhops.hauling.contract.MarineRoute(*, id, block_id, yard_id, origin_port_id, destination_port_id, water_arc_id, contractor_id, truck_type_id)[source]

Bases: BaseModel

A complete block-to-yard chain with a water leg between two ports.

The route is a composite of a road pickup leg, terminal handling at both ports, a barge sailing leg, and a road delivery leg. Phase I treats it as an alternative to a direct road HaulArc; Phase II schedules the drayage loads and barge trips separately.

Parameters:

  • id (str)

  • block_id (str)

  • yard_id (str)

  • origin_port_id (str)

  • destination_port_id (str)

  • water_arc_id (str)

  • contractor_id (str)

  • truck_type_id (str)

id

Unique marine-route identifier.

Type:

str

block_id

Origin block for the pickup drayage leg.

Type:

str

yard_id

Destination sort yard for the delivery drayage leg.

Type:

str

origin_port_id

Loading port.

Type:

str

destination_port_id

Unloading port.

Type:

str

water_arc_id

Sailing arc. Its barge type provides capacity, cost, and load compatibility.

Type:

str

contractor_id

Contractor responsible for the route. The contractor must own an enabled barge of the water arc’s type and at least one enabled truck of truck_type_id.

Type:

str

truck_type_id

Road truck type used for the pickup and delivery drayage legs.

Type:

str

block_id: str
contractor_id: str
destination_port_id: str
id: str
model_config = {}

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

origin_port_id: str
truck_type_id: str
water_arc_id: str
yard_id: str
class fhops.hauling.contract.Port(*, id, name=None, storage_m3=0.0, initial_inventory_m3=0.0, handling_minutes_per_trip=0.0, handling_cost_per_trip=0.0, daily_handling_m3=None)[source]

Bases: BaseModel

A port or barge terminal at one end of a water leg.

Parameters:

  • id (str)

  • name (str | None)

  • storage_m3 (float)

  • initial_inventory_m3 (float)

  • handling_minutes_per_trip (float)

  • handling_cost_per_trip (float)

  • daily_handling_m3 (float | None)

id

Unique port identifier. It is the location key used by water arcs, port legs, and port loader locations.

Type:

str

name

Optional human-readable label.

Type:

str | None

storage_m3

Total log-storage capacity at the terminal, in cubic metres. Non-negative; zero means no spare storage constraint is modelled.

Type:

float

initial_inventory_m3

Opening inventory for every log sort at the port. Non-negative.

Type:

float

handling_minutes_per_trip

Terminal handling time charged once per barge trip at this port. Non-negative.

Type:

float

handling_cost_per_trip

Terminal handling cost charged once per barge trip at this port. Non-negative.

Type:

float

daily_handling_m3

Optional daily throughput cap in cubic metres. None means uncapped.

Type:

float | None

daily_handling_m3: float | None
handling_cost_per_trip: float
handling_minutes_per_trip: float
id: str
initial_inventory_m3: float
model_config = {}

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

name: str | None
storage_m3: float
class fhops.hauling.contract.PortLeg(*, port_id, location_id, side, truck_type_id, travel_minutes, distance_km=None, return_travel_minutes=None)[source]

Bases: BaseModel

A road drayage leg between a block/yard and a port.

Parameters:

  • port_id (str)

  • location_id (str)

  • side (PortSide)

  • truck_type_id (str)

  • travel_minutes (float)

  • distance_km (float | None)

  • return_travel_minutes (float | None)

port_id

Port identifier.

Type:

str

location_id

Block identifier for an origin-side leg or sort-yard identifier for a destination-side leg.

Type:

str

side

origin for the block-to-port leg, destination for the port-to-yard leg.

Type:

PortSide

truck_type_id

Truck type that performs the drayage.

Type:

str

travel_minutes

Loaded one-way travel time in minutes. Must be greater than zero.

Type:

float

distance_km

Optional road distance. Non-negative.

Type:

float | None

return_travel_minutes

Optional empty return time. When None, the loaded time is used.

Type:

float | None

distance_km: float | None
empty_travel_minutes()[source]

Return the empty return time between the port and the land location.

Return type:

float

location_id: str
model_config = {}

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

port_id: str
return_travel_minutes: float | None
side: PortSide
travel_minutes: float
truck_type_id: str
class fhops.hauling.contract.PortSide(*values)[source]

Bases: StrEnum

Which end of a marine route a road drayage leg serves.

ORIGIN is the block-to-port leg on the loading side; DESTINATION is the port-to-yard leg on the unloading side.

DESTINATION = 'destination'
ORIGIN = 'origin'
class fhops.hauling.contract.SortYard(*, id, facility_id=None, unsorted_storage_m3=0.0, sorted_processing_m3_per_day=0.0, unsorted_processing_m3_per_day=0.0, total_processing_m3_per_day=0.0, initial_unsorted_inventory_m3=0.0, initial_water_inventory_m3=0.0)[source]

Bases: BaseModel

A sort yard or mill that receives logs and may process unsorted volume.

Parameters:

  • id (str)

  • facility_id (str | None)

  • unsorted_storage_m3 (float)

  • sorted_processing_m3_per_day (float)

  • unsorted_processing_m3_per_day (float)

  • total_processing_m3_per_day (float)

  • initial_unsorted_inventory_m3 (float)

  • initial_water_inventory_m3 (float)

id

Unique yard identifier. It is also the delivery location key used by loaders, demand, and haul arcs.

Type:

str

facility_id

Optional link to a tactical-operational Facility.facility_id.

Type:

str | None

unsorted_storage_m3

Maximum unsorted inventory that can be stored at the yard (Ghotb’s SC_y).

Type:

float

sorted_processing_m3_per_day

Daily capacity for sorted logs (CW_y).

Type:

float

unsorted_processing_m3_per_day

Daily capacity for sorting unsorted logs (CC_y).

Type:

float

total_processing_m3_per_day

Daily total processing capacity for sorted and unsorted logs (CY_y). Must be at least the larger of the two component capacities.

Type:

float

initial_unsorted_inventory_m3

Unsorted inventory at the start of the first day. Non-negative.

Type:

float

initial_water_inventory_m3

Processed/water inventory at the start of the first day. Non-negative.

Type:

float

facility_id: str | None
id: str
initial_unsorted_inventory_m3: float
initial_water_inventory_m3: float
model_config = {}

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

sorted_processing_m3_per_day: float
total_processing_m3_per_day: float
unsorted_processing_m3_per_day: float
unsorted_storage_m3: float
class fhops.hauling.contract.TransportMode(*values)[source]

Bases: StrEnum

Transport mode for a haul leg.

Road legs use a TruckType; water legs use a BargeType. Rail is reserved for a later extension and is not accepted by the road or water models yet.

RAIL = 'rail'
ROAD = 'road'
WATER = 'water'
class fhops.hauling.contract.Truck(*, id, contractor_id, truck_type_id, enabled=True, notes=None)[source]

Bases: BaseModel

An individual truck owned by a contractor.

Parameters:

  • id (str)

  • contractor_id (str)

  • truck_type_id (str)

  • enabled (bool)

  • notes (str | None)

id

Unique truck identifier. Phase II uses this as the vehicle label.

Type:

str

contractor_id

Contractor that owns and dispatches the truck.

Type:

str

truck_type_id

Truck type that determines capacity, cost, and load compatibility.

Type:

str

enabled

Set False to keep a truck in the contract but exclude it from a solve.

Type:

bool

notes

Optional free-form note surfaced in reports.

Type:

str | None

contractor_id: str
enabled: bool
id: str
model_config = {}

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

notes: str | None
truck_type_id: str
class fhops.hauling.contract.TruckType(*, id, capacity_m3, fixed_cost=0.0, variable_cost_per_minute=0.0, overtime_cost_per_minute=0.0, can_carry_sorted=True, can_carry_unsorted=True)[source]

Bases: BaseModel

A homogeneous truck type with capacity, costs, and load compatibility.

Parameters:

  • id (str)

  • capacity_m3 (float)

  • fixed_cost (float)

  • variable_cost_per_minute (float)

  • overtime_cost_per_minute (float)

  • can_carry_sorted (bool)

  • can_carry_unsorted (bool)

id

Unique truck-type identifier.

Type:

str

capacity_m3

Payload capacity in cubic metres (Ghotb’s Q_r). Must be greater than zero.

Type:

float

fixed_cost

Fixed operating cost of dispatching a truck of this type for the day, in scenario currency (Ghotb’s FC_k). Non-negative.

Type:

float

variable_cost_per_minute

Variable operating cost per minute of regular working time (VC_k). Non-negative.

Type:

float

overtime_cost_per_minute

Overtime cost per minute (OC_k). Non-negative.

Type:

float

can_carry_sorted

Whether the type may carry roadside-sorted loads (gamma_r).

Type:

bool

can_carry_unsorted

Whether the type may carry unsorted loads (epsilon_r).

Type:

bool

can_carry_sorted: bool
can_carry_unsorted: bool
capacity_m3: float
fixed_cost: float
id: str
model_config = {}

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

overtime_cost_per_minute: float
variable_cost_per_minute: float
class fhops.hauling.contract.WaterArc(*, id, origin_port_id, destination_port_id, barge_type_id, travel_minutes, distance_km=None, return_travel_minutes=None, load_minutes=None, unload_minutes=None)[source]

Bases: BaseModel

A sailing leg between two ports for one barge type.

Parameters:

  • id (str)

  • origin_port_id (str)

  • destination_port_id (str)

  • barge_type_id (str)

  • travel_minutes (float)

  • distance_km (float | None)

  • return_travel_minutes (float | None)

  • load_minutes (float | None)

  • unload_minutes (float | None)

id

Unique water-arc identifier referenced by MarineRoute.

Type:

str

origin_port_id

Loading port identifier.

Type:

str

destination_port_id

Unloading port identifier.

Type:

str

barge_type_id

Barge type that can sail the arc.

Type:

str

travel_minutes

Loaded sailing time in minutes. Must be greater than zero.

Type:

float

distance_km

Optional sailing distance for reporting and cost checks. Non-negative.

Type:

float | None

return_travel_minutes

Optional empty return sailing time. When None, Phase I/II use travel_minutes.

Type:

float | None

load_minutes

Optional loading-time override for this arc. When None, the barge type default is used.

Type:

float | None

unload_minutes

Optional unloading-time override for this arc. When None, the barge type default is used.

Type:

float | None

barge_type_id: str
destination_port_id: str
distance_km: float | None
empty_travel_minutes()[source]

Return the empty return sailing time from the destination to the origin.

Return type:

float

id: str
load_minutes: float | None
model_config = {}

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

origin_port_id: str
return_travel_minutes: float | None
travel_minutes: float
unload_minutes: float | None

Phase I monthly flow model for operational log hauling.

This module implements the first phase of Ghotb’s decomposition: a bi-objective mixed-integer program for daily sorted and unsorted log flows from blocks to sort yards, with yard processing, inventory, contractor compatibility, fleet capacity, and contractor workload balance. The bi-objective model is solved by goal programming.

The formulation is documented in notes/hauling_formulation_audit.md (equations 3-1 to 3-23). The implementation deliberately tightens the thesis in three places:

  • compatibility is enforced by index restriction instead of a big-M;

  • the absolute workload deviation is linearised with two inequalities;

  • goal deviations are non-negative continuous variables (the thesis/paper disagree on which one is integer).

The model is built from a validated fhops.scenario.contract.Scenario with a hauling: section. It does not import or modify the operational or tactical MILPs.

class fhops.hauling.flow.FlowSolveResult(status, objective, cost_objective, workload_objective, goal_deviations=<factory>, goal1=None, goal2=None, preferred_truckloads=<factory>, flows=<factory>, truckloads=<factory>, yard_inventory=<factory>, block_inventory=<factory>, contractor_workload=<factory>, marine_flows=<factory>, barge_trips=<factory>, port_inventory=<factory>, marine_truckloads=<factory>, solver='highs', termination='', runtime_seconds=0.0, gap=None, solver_error=None)[source]

Bases: object

Result payload for a Phase I solve.

Parameters:

  • status (str)

  • objective (float | None)

  • cost_objective (float | None)

  • workload_objective (float | None)

  • goal_deviations (dict[str, float])

  • goal1 (float | None)

  • goal2 (float | None)

  • preferred_truckloads (dict[str, float])

  • flows (DataFrame)

  • truckloads (DataFrame)

  • yard_inventory (DataFrame)

  • block_inventory (DataFrame)

  • contractor_workload (DataFrame)

  • marine_flows (DataFrame)

  • barge_trips (DataFrame)

  • port_inventory (DataFrame)

  • marine_truckloads (DataFrame)

  • solver (str)

  • termination (str)

  • runtime_seconds (float)

  • gap (float | None)

  • solver_error (str | None)

status

optimal, feasible, infeasible, unbounded, or error.

Type:

str

objective

Value of the objective that was actually solved (cost, workload, or the normalized goal objective).

Type:

float | None

cost_objective

Value of Ghotb’s Z1 at the returned solution.

Type:

float | None

workload_objective

Value of Ghotb’s Z2 at the returned solution.

Type:

float | None

goal_deviations

cost and workload positive deviations when a goal-programming solve ran.

Type:

dict[str, float]

goal1 / goal2

Goals used by the goal-programming solve.

preferred_truckloads

Preferred daily truckloads used for the workload objective.

Type:

dict[str, float]

flows / truckloads / yard_inventory / block_inventory / contractor_workload

DataFrames with the documented schemas.

marine_flows / barge_trips / port_inventory / marine_truckloads

Marine-leg volume, barge trip, terminal inventory, and drayage truckload tables. They are empty for road-only scenarios.

solver / termination / runtime_seconds / gap / solver_error

Solver provenance and diagnostics.

barge_trips: DataFrame
block_inventory: DataFrame
contractor_workload: DataFrame
cost_objective: float | None
flows: DataFrame
gap: float | None
goal1: float | None
goal2: float | None
goal_deviations: dict[str, float]
marine_flows: DataFrame
marine_truckloads: DataFrame
objective: float | None
port_inventory: DataFrame
preferred_truckloads: dict[str, float]
runtime_seconds: float
solver: str
solver_error: str | None
status: str
termination: str
truckloads: DataFrame
workload_objective: float | None
yard_inventory: DataFrame
class fhops.hauling.flow.HaulingFlowProblem(scenario_name, start_day, horizon_days, days, blocks, yards, sorts, truck_types, contractors, supply_sorted, supply_unsorted, demand, cycle_minutes, transport_cost, capacity, fleet_count, preferred, contractor_block_allowlist, can_carry_sorted, can_carry_unsorted, yard_storage, yard_sorted_capacity, yard_unsorted_capacity, yard_total_capacity, initial_unsorted, initial_water, regular_minutes, roadside_penalty, default_service_minutes, pickup_service, delivery_service, marine_lanes=<factory>, marine_ports=(), marine_port_storage=<factory>, marine_port_initial=<factory>, marine_port_daily_handling=<factory>, marine_barge_fleet=<factory>, port_inventory_penalty=0.0)[source]

Bases: object

Normalised Phase I data derived from a scenario’s hauling section.

Parameters:
scenario_name

Parent scenario name.

Type:

str

start_day

One-indexed parent-scenario day where the hauling horizon starts.

Type:

int

horizon_days

Number of days in the Phase I horizon.

Type:

int

days

Relative days 1..horizon_days.

Type:

tuple[int, …]

blocks / yards / sorts / truck_types / contractors

Ordered entity identifiers.

supply_sorted / supply_unsorted

(block_id, log_sort_id) -> m3. Zero-supply pairs are omitted.

demand

(yard_id, log_sort_id, relative_day) -> m3. Missing pairs are zero.

Type:

dict[tuple[str, str, int], float]

cycle_minutes

(block_id, yard_id, truck_type_id) -> round-trip plus loading/unloading minutes.

Type:

dict[tuple[str, str, str], float]

transport_cost

(block_id, yard_id, truck_type_id) -> cost per started truckload.

Type:

dict[tuple[str, str, str], float]

capacity

truck_type_id -> m3.

Type:

dict[str, float]

fleet_count

(contractor_id, truck_type_id) -> number of enabled trucks.

Type:

dict[tuple[str, str], int]

preferred

contractor_id -> preferred daily truckloads (None when not supplied).

Type:

dict[str, float | None]

contractor_block_allowlist

Allowed (contractor_id, block_id) pairs. Empty means every contractor may serve every block.

Type:

frozenset[tuple[str, str]]

can_carry_sorted / can_carry_unsorted

Truck-type flags.

yard_storage / yard_sorted_capacity / yard_unsorted_capacity / yard_total_capacity

yard_id -> capacity.

initial_unsorted / initial_water

(yard_id, log_sort_id) -> opening inventory.

regular_minutes

Regular truck minutes per day.

Type:

float

roadside_penalty

Cost per m3 left at a block at the end of the horizon.

Type:

float

default_service_minutes

Fallback loader service time.

Type:

float

pickup_service / delivery_service

location_id -> service minutes.

marine_lanes

route_id -> MarineLane. Empty for a road-only problem.

Type:

dict[str, fhops.hauling.flow.MarineLane]

marine_ports / marine_port_storage / marine_port_initial / marine_port_daily_handling

Port ids and their storage, opening inventory, and optional daily handling caps.

marine_barge_fleet

(contractor_id, barge_type_id) -> number of enabled barges.

Type:

dict[tuple[str, str], int]

port_inventory_penalty

Cost per m3 left at a port at the end of the horizon.

Type:

float

blocks: tuple[str, ...]
can_carry_sorted: dict[str, bool]
can_carry_unsorted: dict[str, bool]
capacity: dict[str, float]
contractor_block_allowlist: frozenset[tuple[str, str]]
contractor_can_serve(contractor_id, block_id)[source]

Return whether a contractor may serve a block under the allowlist rule.

Parameters:

  • contractor_id (str)

  • block_id (str)

Return type:

bool

contractors: tuple[str, ...]
cycle_minutes: dict[tuple[str, str, str], float]
days: tuple[int, ...]
default_service_minutes: float
delivery_service: dict[str, float]
demand: dict[tuple[str, str, int], float]
demand_for(yard_id, log_sort_id, day)[source]

Return demand for one yard/sort/day (zero when absent).

Parameters:
Return type:

float

fleet_count: dict[tuple[str, str], int]
has_marine()[source]

Return whether the problem contains any marine routes.

Return type:

bool

horizon_days: int
initial_unsorted: dict[tuple[str, str], float]
initial_water: dict[tuple[str, str], float]
marine_barge_fleet: dict[tuple[str, str], int]
marine_lanes: dict[str, MarineLane]
marine_port_daily_handling: dict[str, float | None]
marine_port_initial: dict[tuple[str, str], float]
marine_port_storage: dict[str, float]
marine_ports: tuple[str, ...]
marine_routes()[source]

Return marine lanes in route-id order.

Return type:

tuple[MarineLane, …]

pickup_service: dict[str, float]
port_inventory_penalty: float
preferred: dict[str, float | None]
regular_minutes: float
roadside_penalty: float
scenario_name: str
sorts: tuple[str, ...]
start_day: int
supply_sorted: dict[tuple[str, str], float]
supply_unsorted: dict[tuple[str, str], float]
transport_cost: dict[tuple[str, str, str], float]
truck_types: tuple[str, ...]
yard_sorted_capacity: dict[str, float]
yard_storage: dict[str, float]
yard_total_capacity: dict[str, float]
yard_unsorted_capacity: dict[str, float]
yards: tuple[str, ...]
class fhops.hauling.flow.MarineLane(route_id, block_id, yard_id, origin_port_id, destination_port_id, contractor_id, truck_type_id, barge_type_id, water_arc_id, pickup_travel_minutes, pickup_return_minutes, delivery_travel_minutes, delivery_return_minutes, water_travel_minutes, water_return_minutes, load_minutes, unload_minutes, origin_handling_minutes, destination_handling_minutes, origin_handling_cost, destination_handling_cost, barge_capacity, truck_capacity, barge_cycle_minutes, arrival_lag_days, occupancy_days, pickup_cycle_minutes, delivery_cycle_minutes, barge_cost_per_trip, pickup_drayage_cost_per_m3, delivery_drayage_cost_per_m3, drayage_cost_per_m3, can_carry_sorted, can_carry_unsorted)[source]

Bases: object

Normalised Phase I data for one marine route.

A lane bundles the road pickup leg, the water leg, terminal handling at both ports, and the road delivery leg. The Phase I model treats the lane as an alternative supply path from a block to a yard, while still reporting the three physical legs separately.

Parameters:

  • route_id (str)

  • block_id (str)

  • yard_id (str)

  • origin_port_id (str)

  • destination_port_id (str)

  • contractor_id (str)

  • truck_type_id (str)

  • barge_type_id (str)

  • water_arc_id (str)

  • pickup_travel_minutes (float)

  • pickup_return_minutes (float)

  • delivery_travel_minutes (float)

  • delivery_return_minutes (float)

  • water_travel_minutes (float)

  • water_return_minutes (float)

  • load_minutes (float)

  • unload_minutes (float)

  • origin_handling_minutes (float)

  • destination_handling_minutes (float)

  • origin_handling_cost (float)

  • destination_handling_cost (float)

  • barge_capacity (float)

  • truck_capacity (float)

  • barge_cycle_minutes (float)

  • arrival_lag_days (int)

  • occupancy_days (int)

  • pickup_cycle_minutes (float)

  • delivery_cycle_minutes (float)

  • barge_cost_per_trip (float)

  • pickup_drayage_cost_per_m3 (float)

  • delivery_drayage_cost_per_m3 (float)

  • drayage_cost_per_m3 (float)

  • can_carry_sorted (bool)

  • can_carry_unsorted (bool)

route_id

Marine-route identifier.

Type:

str

block_id / yard_id

Pickup block and delivery sort yard.

origin_port_id / destination_port_id

Loading and unloading ports.

contractor_id / truck_type_id

Contractor and road truck type for the drayage legs.

barge_type_id / water_arc_id

Barge equipment and sailing arc.

pickup_travel_minutes / pickup_return_minutes

Loaded and empty road times between the block and the origin port.

delivery_travel_minutes / delivery_return_minutes

Loaded and empty road times between the destination port and the yard.

water_travel_minutes / water_return_minutes

Loaded and empty sailing times.

load_minutes / unload_minutes

Barge loading and unloading service times.

origin_handling_minutes / destination_handling_minutes

Terminal handling time at each port.

origin_handling_cost / destination_handling_cost

Terminal handling cost per barge trip at each port.

barge_capacity / truck_capacity

Cargo capacities in cubic metres.

barge_cycle_minutes

Round-trip barge time including loading, unloading, and terminal handling.

Type:

float

arrival_lag_days

Whole-day lag between a departure and cargo arrival at the destination. 0 means same-day delivery.

Type:

int

occupancy_days

Whole days a barge is occupied by the round trip. Used by the sliding-window fleet constraint.

Type:

int

pickup_cycle_minutes / delivery_cycle_minutes

Round-trip drayage truck time on each side.

barge_cost_per_trip

Variable barge cost plus terminal handling cost for one trip.

Type:

float

pickup_drayage_cost_per_m3 / delivery_drayage_cost_per_m3

Road truck variable cost per cubic metre on each drayage leg.

drayage_cost_per_m3

Sum of the pickup and delivery drayage rates, kept for reporting.

Type:

float

can_carry_sorted / can_carry_unsorted

Barge-type compatibility flags.

arrival_lag_days: int
barge_capacity: float
barge_cost_per_trip: float
barge_cycle_minutes: float
barge_type_id: str
block_id: str
can_carry_sorted: bool
can_carry_unsorted: bool
contractor_id: str
delivery_cycle_minutes: float
delivery_drayage_cost_per_m3: float
delivery_return_minutes: float
delivery_travel_minutes: float
destination_handling_cost: float
destination_handling_minutes: float
destination_port_id: str
drayage_cost_per_m3: float
load_minutes: float
occupancy_days: int
origin_handling_cost: float
origin_handling_minutes: float
origin_port_id: str
pickup_cycle_minutes: float
pickup_drayage_cost_per_m3: float
pickup_return_minutes: float
pickup_travel_minutes: float
route_id: str
truck_capacity: float
truck_type_id: str
unload_minutes: float
water_arc_id: str
water_return_minutes: float
water_travel_minutes: float
yard_id: str
fhops.hauling.flow.build_flow_model(problem, *, objective='cost', preferred=None, goals=None, weights=(1.0, 1.0))[source]

Build the Phase I Pyomo model.

Parameters:

  • problem (HaulingFlowProblem) – Normalised hauling data.

  • objective (str) – "cost" (Ghotb’s Z1), "workload" (Z2), or "goal" (goal programming).

  • preferred (dict[str, float] | None) – Preferred daily truckloads per contractor. None uses 0 for the workload constraint; solve_flow() derives values from a cost-only solve when needed.

  • goals (tuple[float, float] | None) – (Goal1, Goal2) for the goal-programming objective. Required when objective="goal" unless solve_flow() derives them.

  • weights (tuple[float, float]) – (w1, w2) goal weights. Must be non-negative and not both zero when objective="goal".

Returns:

Model with cost_expr, workload_expr, and a single objective component.

Return type:

pyomo.ConcreteModel

Raises:

ValueError – For an unknown objective or invalid goal weights/goals.

fhops.hauling.flow.build_flow_problem(scenario)[source]

Build a HaulingFlowProblem from a validated scenario.

Parameters:

scenario (Scenario) – Scenario with a non-None hauling section.

Returns:

Normalised data ready for build_flow_model() and solve_flow().

Return type:

HaulingFlowProblem

Raises:

ValueError – When the scenario has no hauling section, or a supply/demand/arc combination cannot be represented by the contract.

fhops.hauling.flow.solve_flow(problem, *, objective='cost', solver='highs', time_limit=None, gap=None, weights=(1.0, 1.0), goals=None, preferred=None, tee=False, solver_options=None)[source]

Solve the Phase I flow model.

Parameters:

  • problem (HaulingFlowProblem) – Normalised hauling data.

  • objective (str) – "cost", "workload", or "goal".

  • solver (str) – Pyomo solver name. highs is the default; gurobi is used when installed and licensed.

  • time_limit (float | None) – Optional solver time limit in seconds.

  • gap (float | None) – Optional relative MIP gap target.

  • weights (tuple[float, float]) – Goal weights for objective="goal".

  • goals (tuple[float, float] | None) – (Goal1, Goal2). When omitted for a goal solve, a cost-only solve supplies Goal1 and Goal2 is set to zero; preferred truckloads are derived from the same cost-only plan.

  • preferred (dict[str, float] | None) – Preferred daily truckloads per contractor. When omitted, the contract values are used, or the cost-only plan’s daily averages for a goal solve.

  • tee (bool) – Stream the solver log.

  • solver_options (dict[str, Any] | None) – Additional solver options forwarded verbatim.

Returns:

Result payload with objective values, DataFrames, solver provenance, and diagnostics.

Return type:

FlowSolveResult

fhops.hauling.flow.summarize_flow_result(result)[source]

Return a compact human-readable summary of a Phase I solve.

Parameters:

result (FlowSolveResult) – Result payload from solve_flow().

Returns:

Multi-line summary suitable for a CLI or notebook cell. It reports status, objective values, goal deviations, truckloads, delivered volume, and the ending yard inventory.

Return type:

str

Network and data preparation for Phase II daily truck scheduling.

Phase II takes the integer truckloads from the Phase I flow plan and turns them into individual loads on a single working day. This module builds the normalised ScheduleProblem that the continuous-time MIP in fhops.hauling.schedule solves.

class fhops.hauling.network.ScheduleProblem(scenario_name, day, scenario_day, horizon_minutes, depot_id, loads, trucks, truck_types, contractors, pickup_loaders, delivery_loaders, loader_service, block_to_yard, yard_to_block, depot_to_block, yard_to_depot, regular_minutes, max_overtime_minutes, overtime_violation_penalty_per_minute, marine=None)[source]

Bases: object

Normalised Phase II scheduling data.

Parameters:
scenario_name

Parent scenario name.

Type:

str

day

Relative hauling-horizon day being scheduled.

Type:

int

scenario_day

Parent-scenario day (start_day + day - 1).

Type:

int

horizon_minutes

Daily time horizon, regular_minutes + max_overtime_minutes.

Type:

float

depot_id

The single central depot used for this first scheduling cut.

Type:

str

loads

Individual truckloads.

Type:

tuple[Truckload, …]

trucks

Enabled trucks from the contract.

Type:

tuple[Truck, …]

truck_types / contractors

Contract records keyed by id.

pickup_loaders / delivery_loaders

location_id -> tuple of loader ids.

loader_service

loader_id -> service minutes.

Type:

dict[str, float]

block_to_yard / yard_to_block

(block, yard, truck_type) -> minutes.

depot_to_block / yard_to_depot

(location, truck_type) -> minutes.

regular_minutes / max_overtime_minutes

Driver policy.

overtime_violation_penalty_per_minute

Soft penalty per minute above regular_minutes + max_overtime_minutes used by the SA solver. The exact MIP treats the limit as hard.

Type:

float

marine

Optional marine scheduling data for barge cycles and port loaders. None for road-only days.

Type:

MarineScheduleProblem | None

block_to_yard: dict[tuple[str, str, str], float]
compatible_trucks(load)[source]

Return enabled trucks that can carry a load.

Parameters:

load (Truckload)

Return type:

tuple[Truck, …]

contractors: dict[str, Contractor]
day: int
delivery_loaders: dict[str, tuple[str, ...]]
delivery_loaders_for(load)[source]

Return the delivery loaders at a load’s yard.

Parameters:

load (Truckload)

Return type:

tuple[str, …]

delivery_loads()[source]

Return the loads sorted by yard id and load id.

Return type:

tuple[Truckload, …]

depot_id: str
depot_to_block: dict[tuple[str, str], float]
depot_to_load_minutes(load)[source]

Return the depot-to-pickup travel time.

Parameters:

load (Truckload)

Return type:

float

empty_travel_minutes(previous, following)[source]

Return the empty travel time from previous’s yard to following’s block.

Marine drayage stages are separate physical movements: a truck that delivers to the destination port cannot carry a pickup load from the origin block without a deadhead that the first release does not model. Missing cross-stage pairs receive a synthetic full-day deadhead so the MIP can express the transition as infeasible rather than raising a key error; a schedule that uses one is reported as a validation warning.

Parameters:

  • previous (Truckload)

  • following (Truckload)

Return type:

float

horizon_minutes: float
load_by_id(load_id)[source]

Return a load by id.

Parameters:

load_id (str)

Return type:

Truckload

load_to_depot_minutes(load)[source]

Return the delivery-to-depot travel time.

Parameters:

load (Truckload)

Return type:

float

load_travel_minutes(load)[source]

Return the loaded block-to-yard travel time.

Parameters:

load (Truckload)

Return type:

float

loader_service: dict[str, float]
loads: tuple[Truckload, ...]
marine: MarineScheduleProblem | None
max_overtime_minutes: float
overtime_violation_penalty_per_minute: float
pickup_loaders: dict[str, tuple[str, ...]]
pickup_loaders_for(load)[source]

Return the pickup loaders at a load’s block.

Parameters:

load (Truckload)

Return type:

tuple[str, …]

pickup_loads()[source]

Return the loads sorted by block id and load id.

Return type:

tuple[Truckload, …]

regular_minutes: float
scenario_day: int
scenario_name: str
truck_types: dict[str, TruckType]
trucks: tuple[Truck, ...]
yard_to_block: dict[tuple[str, str, str], float]
yard_to_depot: dict[tuple[str, str], float]
class fhops.hauling.network.Truckload(load_id, day, block_id, yard_id, contractor_id, truck_type_id, sorted, volume_m3, leg='road', marine_route_id=None)[source]

Bases: object

One individual truckload on the scheduling day.

Parameters:

  • load_id (str)

  • day (int)

  • block_id (str)

  • yard_id (str)

  • contractor_id (str)

  • truck_type_id (str)

  • sorted (bool)

  • volume_m3 (float)

  • leg (str)

  • marine_route_id (str | None)

load_id

Unique identifier within the day.

Type:

str

day

Relative hauling-horizon day (one-indexed).

Type:

int

block_id

Pickup block.

Type:

str

yard_id

Delivery sort yard.

Type:

str

contractor_id

Contractor responsible for the load.

Type:

str

truck_type_id

Compatible truck type.

Type:

str

sorted

True for a roadside-sorted load, False for an unsorted load.

Type:

bool

volume_m3

Nominal load volume. Phase II uses it for reporting; the scheduling constraints use the truck type’s capacity and the load count, not the exact partial volume.

Type:

float

leg

road for a direct block-to-yard haul, pickup for the block-to-origin-port drayage leg, or delivery for the destination-port-to-yard drayage leg.

Type:

str

marine_route_id

Marine route identifier for drayage loads; None for road loads.

Type:

str | None

block_id: str
contractor_id: str
day: int
leg: str
load_id: str
marine_route_id: str | None
sorted: bool
truck_type_id: str
volume_m3: float
yard_id: str
fhops.hauling.network.build_schedule_problem(scenario, flow_result, day)[source]

Build a scheduling problem from a Phase I result and a selected day.

Parameters:

  • scenario (Scenario) – Scenario with a hauling section.

  • flow_result (FlowSolveResult) – Result from fhops.hauling.flow.solve_flow().

  • day (int) – Relative hauling-horizon day to schedule.

Returns:

Normalised loads, trucks, loaders, and travel times.

Return type:

ScheduleProblem

Raises:

ValueError – When the day is outside the horizon, the scenario has no hauling section, the day has no truckloads, or a required loader, depot leg, or haul arc is missing.

fhops.hauling.network.build_schedule_problem_from_loads(scenario, loads, day)[source]

Build a scheduling problem from an explicit load list.

This is the test and experimentation entry point. The production path is build_schedule_problem().

Parameters:

  • scenario (Scenario) – Scenario with a hauling section.

  • loads (Sequence[Truckload]) – Individual loads on the selected day.

  • day (int) – Relative hauling-horizon day.

Returns:

Normalised scheduling data.

Return type:

ScheduleProblem

Raises:

ValueError – When the day is invalid, the scenario has no hauling section, the contract has more than one depot, or a required loader/travel leg is missing.

Phase II continuous-time daily truck routing and scheduling.

This module implements the second phase of Ghotb’s decomposition: a mixed-integer program that assigns Phase I truckloads to compatible trucks and contractors, routes the trucks from a central depot, and synchronises trucks with stationary loaders at pickup and delivery. Time is continuous, so the model returns arrival, waiting, service, and return times rather than slot assignments.

The formulation follows Chapter 4 of Ghotb (2023), with the corrections recorded in notes/hauling_formulation_audit.md:

  • each truckload is assigned to exactly one compatible truck;

  • each truck has a start/end route and a sequence of loads, with MTZ subtour elimination;

  • each pickup and delivery load is assigned to one loader at its location;

  • loaders have a total order and no overlapping service;

  • trucks and loaders share the service start time, so waiting is the gap between arrival and start;

  • regular time and overtime are separate variables, so overtime is charged exactly.

The exact MIP is intended for small and medium daily instances. The simulated-annealing solver in P11.4 handles the large case.

class fhops.hauling.schedule.SchedulePlan(routes, pickup_loader, delivery_loader, pickup_order, delivery_order)[source]

Bases: object

A route and loader-order plan for a scheduling problem.

Parameters:
routes

truck_id -> ordered tuple of load ids.

Type:

dict[str, tuple[str, …]]

pickup_loader / delivery_loader

load_id -> loader_id.

pickup_order / delivery_order

loader_id -> ordered tuple of load ids.

delivery_loader: dict[str, str]
delivery_order: dict[str, tuple[str, ...]]
pickup_loader: dict[str, str]
pickup_order: dict[str, tuple[str, ...]]
routes: dict[str, tuple[str, ...]]
class fhops.hauling.schedule.ScheduleSolveResult(status, objective, fixed_cost, variable_cost, overtime_cost, schedule=<factory>, truck_summary=<factory>, loader_summary=<factory>, barge_schedule=<factory>, barge_summary=<factory>, barge_objective=None, barge_fixed_cost=0.0, plan=None, solver='highs', termination='', runtime_seconds=0.0, solver_error=None)[source]

Bases: object

Result payload for a Phase II solve.

Parameters:

  • status (str)

  • objective (float | None)

  • fixed_cost (float | None)

  • variable_cost (float | None)

  • overtime_cost (float | None)

  • schedule (DataFrame)

  • truck_summary (DataFrame)

  • loader_summary (DataFrame)

  • barge_schedule (DataFrame)

  • barge_summary (DataFrame)

  • barge_objective (float | None)

  • barge_fixed_cost (float)

  • plan (SchedulePlan | None)

  • solver (str)

  • termination (str)

  • runtime_seconds (float)

  • solver_error (str | None)

status

optimal, feasible, infeasible, unbounded, or error.

Type:

str

objective / fixed_cost / variable_cost / overtime_cost

Objective components.

schedule / truck_summary / loader_summary

DataFrames with the documented schemas.

barge_schedule / barge_summary

Barge-trip and barge-level tables for a marine day; empty for road-only days.

plan

The route and loader-order plan, when a solution exists.

Type:

fhops.hauling.schedule.SchedulePlan | None

solver / termination / runtime_seconds / solver_error

Solver provenance and diagnostics.

barge_fixed_cost: float
barge_objective: float | None
barge_schedule: DataFrame
barge_summary: DataFrame
fixed_cost: float | None
loader_summary: DataFrame
objective: float | None
overtime_cost: float | None
plan: SchedulePlan | None
runtime_seconds: float
schedule: DataFrame
solver: str
solver_error: str | None
status: str
termination: str
truck_summary: DataFrame
variable_cost: float | None
fhops.hauling.schedule.build_schedule_model(problem)[source]

Build the Phase II continuous-time scheduling MIP.

Parameters:

problem (ScheduleProblem) – Normalised scheduling data from fhops.hauling.network.build_schedule_problem().

Returns:

Minimisation model with a single objective.

Return type:

pyomo.ConcreteModel

Raises:

ValueError – When a load has no compatible truck or no loader.

fhops.hauling.schedule.simulate_plan(problem, plan)[source]

Recompute the earliest schedule for a fixed route and loader order.

The simulator solves the max-plus system defined by the truck routes and loader sequences by fixed-point iteration. It is independent of the MIP’s time variables and is used as a verification oracle for the model.

Parameters:

  • problem (ScheduleProblem) – Scheduling data.

  • plan (SchedulePlan) – Truck routes and loader orders.

Returns:

One row per load with arrival_pickup, pickup_start, pickup_finish, arrival_delivery, delivery_start, delivery_finish, and load_order.

Return type:

pandas.DataFrame

fhops.hauling.schedule.solve_schedule(problem, *, solver='highs', time_limit=None, gap=None, tee=False, solver_options=None)[source]

Solve the Phase II scheduling MIP.

Parameters:

  • problem (ScheduleProblem) – Normalised scheduling data.

  • solver (str) – Pyomo solver name. highs is the default; gurobi is used when installed and licensed.

  • time_limit (float | None) – Optional solver time limit in seconds.

  • gap (float | None) – Optional relative MIP gap target.

  • tee (bool) – Stream the solver log.

  • solver_options (dict[str, Any] | None) – Additional solver options forwarded verbatim.

Returns:

Schedule, truck summary, loader summary, plan, objective components, and solver provenance.

Return type:

ScheduleSolveResult

fhops.hauling.schedule.validate_schedule(problem, schedule)[source]

Check a schedule for compatibility, synchronisation, and overlap violations.

Parameters:

  • problem (ScheduleProblem) – Scheduling data.

  • schedule (DataFrame) – A schedule table, usually ScheduleSolveResult.schedule.

Returns:

Human-readable violations. An empty list means the schedule is feasible.

Return type:

list[str]

Marine barge-cycle scheduling and port synchronization.

The road scheduler in fhops.hauling.schedule routes trucks and loaders. A marine route adds a second equipment class: a barge that loads at an origin port, sails a water arc, unloads at a destination port, and returns empty. This module turns the Phase I barge trips into a deterministic single-day barge schedule with:

  • one row per barge trip;

  • port-loader assignment and non-overlapping terminal service;

  • barge load, sail, unload, and empty-return times;

  • the synchronized road pickup and delivery windows for the drayage legs.

The deterministic scheduler is deliberately small and auditable. It complements the exact truck MIP and the SA solver; it does not replace either. The exact truck MIP schedules the road loads (including the drayage legs when a marine route is present), while this module schedules the water leg. The returned road windows are the handoff between the two schedules and are reported, not hard-linked, in the first release.

class fhops.hauling.marine.BargeSchedulePlan(barge_trips)[source]

Bases: object

Deterministic barge and port-loader assignment.

Parameters:

barge_trips (dict[str, tuple[tuple[str, str, str], ...]])

barge_trips

route_id -> tuple of (barge_id, origin_loader_id, destination_loader_id) in schedule order.

Type:

dict[str, tuple[tuple[str, str, str], …]]

barge_trips: dict[str, tuple[tuple[str, str, str], ...]]
class fhops.hauling.marine.BargeScheduleResult(status, schedule=<factory>, summary=<factory>, plan=None, objective=0.0, fixed_cost=0.0, runtime_seconds=0.0, error=None)[source]

Bases: object

Barge schedule payload attached to a Phase II or Phase III result.

Parameters:

  • status (str)

  • schedule (DataFrame)

  • summary (DataFrame)

  • plan (BargeSchedulePlan | None)

  • objective (float)

  • fixed_cost (float)

  • runtime_seconds (float)

  • error (str | None)

status

feasible when every request was scheduled, empty when there were no requests.

Type:

str

schedule / summary

Trip-level and barge-level DataFrames.

plan

Deterministic assignment plan.

Type:

fhops.hauling.marine.BargeSchedulePlan | None

objective

Total barge fixed, variable, and terminal handling cost.

Type:

float

fixed_cost

Fixed dispatch cost of the barges used.

Type:

float

runtime_seconds

Scheduler wall time.

Type:

float

error

Optional diagnostic.

Type:

str | None

error: str | None
fixed_cost: float
objective: float
plan: BargeSchedulePlan | None
runtime_seconds: float
schedule: DataFrame
status: str
summary: DataFrame
class fhops.hauling.marine.MarineScheduleProblem(scenario_name, day, scenario_day, horizon_minutes, requests, barges, barge_types, ports, port_loaders, loader_service)[source]

Bases: object

Normalised marine scheduling data for one day.

Parameters:
scenario_name / day / scenario_day / horizon_minutes

Run context.

requests

Barge-trip requests from Phase I.

Type:

tuple[fhops.hauling.marine.MarineTripRequest, …]

barges

Enabled barges keyed by id.

Type:

dict[str, fhops.hauling.contract.Barge]

barge_types

Barge types keyed by id.

Type:

dict[str, fhops.hauling.contract.BargeType]

ports

Ports keyed by id.

Type:

dict[str, fhops.hauling.contract.Port]

port_loaders

port_id -> tuple of loader ids. A port loader serves both barge handling and drayage.

Type:

dict[str, tuple[str, …]]

loader_service

loader_id -> service minutes.

Type:

dict[str, float]

barge_types: dict[str, BargeType]
barges: dict[str, Barge]
day: int
horizon_minutes: float
loader_service: dict[str, float]
port_loaders: dict[str, tuple[str, ...]]
ports: dict[str, Port]
requests: tuple[MarineTripRequest, ...]
scenario_day: int
scenario_name: str
class fhops.hauling.marine.MarineTripRequest(route_id, contractor_id, truck_type_id, barge_type_id, origin_port_id, destination_port_id, water_arc_id, volume_m3, trips, load_minutes, unload_minutes, sail_minutes, return_minutes, origin_handling_minutes, destination_handling_minutes, origin_handling_cost, destination_handling_cost, pickup_travel_minutes, delivery_travel_minutes, truck_capacity, barge_variable_cost_per_trip)[source]

Bases: object

One Phase I barge-trip request for a route/day.

Parameters:

  • route_id (str)

  • contractor_id (str)

  • truck_type_id (str)

  • barge_type_id (str)

  • origin_port_id (str)

  • destination_port_id (str)

  • water_arc_id (str)

  • volume_m3 (float)

  • trips (int)

  • load_minutes (float)

  • unload_minutes (float)

  • sail_minutes (float)

  • return_minutes (float)

  • origin_handling_minutes (float)

  • destination_handling_minutes (float)

  • origin_handling_cost (float)

  • destination_handling_cost (float)

  • pickup_travel_minutes (float)

  • delivery_travel_minutes (float)

  • truck_capacity (float)

  • barge_variable_cost_per_trip (float)

route_id

Marine route identifier.

Type:

str

contractor_id / truck_type_id / barge_type_id

Responsible contractor, drayage truck type, and barge type.

origin_port_id / destination_port_id / water_arc_id

Water-leg topology.

volume_m3

Water-leg volume in cubic metres.

Type:

float

trips

Number of barge trips that carry volume_m3.

Type:

int

load_minutes / unload_minutes / sail_minutes / return_minutes

Barge service and sailing times per trip.

origin_handling_minutes / destination_handling_minutes

Terminal handling time per trip at each port.

origin_handling_cost / destination_handling_cost

Terminal handling cost per trip at each port.

pickup_travel_minutes / delivery_travel_minutes

Loaded road drayage travel times used to place the synchronized road windows.

truck_capacity

Drayage truck capacity in cubic metres.

Type:

float

barge_variable_cost_per_trip

Variable barge cost per trip.

Type:

float

barge_type_id: str
barge_variable_cost_per_trip: float
contractor_id: str
delivery_travel_minutes: float
destination_handling_cost: float
destination_handling_minutes: float
destination_port_id: str
load_minutes: float
origin_handling_cost: float
origin_handling_minutes: float
origin_port_id: str
pickup_travel_minutes: float
return_minutes: float
route_id: str
sail_minutes: float
trips: int
truck_capacity: float
truck_type_id: str
unload_minutes: float
volume_m3: float
water_arc_id: str
fhops.hauling.marine.build_marine_schedule_problem(scenario, flow_result, day)[source]

Build the marine scheduling input from a Phase I result.

Parameters:

  • scenario (Scenario) – Scenario with a hauling section.

  • flow_result (FlowSolveResult) – Phase I result. The barge_trips table supplies the trip counts and water volumes.

  • day (int) – Relative hauling-horizon day to schedule.

Returns:

None when the day has no barge trips.

Return type:

MarineScheduleProblem | None

Raises:

ValueError – When a barge trip references an unknown route, a port has no loader, or the contract lacks an enabled barge for the route.

fhops.hauling.marine.schedule_barge_trips(problem)[source]

Schedule barge trips deterministically with port-loader synchronization.

The scheduler assigns each trip to the compatible barge that becomes available first, and to the compatible port loader that becomes available first at each end. It is a deterministic list scheduler, not an optimiser.

Parameters:

problem (MarineScheduleProblem) – Normalised marine scheduling data.

Returns:

One row per barge trip plus a barge-level summary.

Return type:

BargeScheduleResult

fhops.hauling.marine.validate_barge_schedule(problem, schedule)[source]

Validate a barge schedule against its problem.

Parameters:

  • problem (MarineScheduleProblem) – Marine scheduling data.

  • schedule (DataFrame) – Barge schedule DataFrame produced by schedule_barge_trips().

Returns:

Human-readable violations. An empty list means the schedule is valid.

Return type:

list[str]

Phase III simulated annealing for daily log-truck scheduling.

This module implements the solution approach of Ghotb (2023), Chapter 5: a permutation-based representation of truck routes, a shortest-processing-time (SPT) schedule decoder, swap/reversion/ insertion neighbourhoods, and a geometric cooling schedule. It is the large-instance path; the exact MIP in fhops.hauling.schedule remains the oracle for small days.

The decoder is deliberately built on top of fhops.hauling.schedule.simulate_plan(), so the timing logic is shared with the exact-model verification oracle. It derives loader orders from projected arrival times (SPT) and then recomputes the earliest schedule for the fixed routes and loader orders.

class fhops.hauling.heuristic.HaulingSAResult(status, objective, fixed_cost, variable_cost, overtime_cost, violation_cost, violation_minutes, schedule, truck_summary, loader_summary, plan, iterations, accepted, runtime_seconds, seed, history, barge_schedule=<factory>, barge_summary=<factory>, barge_objective=None, barge_fixed_cost=0.0, solver_error=None)[source]

Bases: object

Result payload for a simulated-annealing run.

Parameters:

  • status (str)

  • objective (float | None)

  • fixed_cost (float | None)

  • variable_cost (float | None)

  • overtime_cost (float | None)

  • violation_cost (float | None)

  • violation_minutes (float | None)

  • schedule (DataFrame)

  • truck_summary (DataFrame)

  • loader_summary (DataFrame)

  • plan (SchedulePlan | None)

  • iterations (int)

  • accepted (int)

  • runtime_seconds (float)

  • seed (int)

  • history (DataFrame)

  • barge_schedule (DataFrame)

  • barge_summary (DataFrame)

  • barge_objective (float | None)

  • barge_fixed_cost (float)

  • solver_error (str | None)

status

feasible when a schedule was evaluated; error only for an invalid problem.

Type:

str

objective / fixed_cost / variable_cost / overtime_cost / violation_cost

Best-solution cost components.

violation_minutes

Best-solution overtime violation.

Type:

float | None

schedule / truck_summary / loader_summary / plan

Best-solution artefacts.

barge_schedule / barge_summary / barge_objective

Barge-cycle tables for a marine day; empty for road-only days.

iterations / accepted / runtime_seconds / seed

Search provenance.

history

Per-move objective trace.

Type:

pandas.DataFrame

accepted: int
barge_fixed_cost: float
barge_objective: float | None
barge_schedule: DataFrame
barge_summary: DataFrame
fixed_cost: float | None
history: DataFrame
iterations: int
loader_summary: DataFrame
objective: float | None
overtime_cost: float | None
plan: SchedulePlan | None
runtime_seconds: float
schedule: DataFrame
seed: int
solver_error: str | None
status: str
truck_summary: DataFrame
variable_cost: float | None
violation_cost: float | None
violation_minutes: float | None
class fhops.hauling.heuristic.SAConfig(initial_temperature=300.0, min_temperature_ratio=0.0005, cooling_rate=0.95, iterations_per_temperature=90, max_iterations=None, time_limit_seconds=None, seed=42, operator_weights=<factory>, overtime_violation_penalty_per_minute=None)[source]

Bases: object

Simulated-annealing configuration.

Parameters:

  • initial_temperature (float)

  • min_temperature_ratio (float)

  • cooling_rate (float)

  • iterations_per_temperature (int)

  • max_iterations (int | None)

  • time_limit_seconds (float | None)

  • seed (int)

  • operator_weights (dict[str, float])

  • overtime_violation_penalty_per_minute (float | None)

initial_temperature

Starting temperature. The thesis Taguchi calibration selects 300.

Type:

float

min_temperature_ratio

Stop when temperature <= initial_temperature * min_temperature_ratio. The thesis uses 0.0005.

Type:

float

cooling_rate

Geometric cooling factor. The thesis uses 0.95.

Type:

float

iterations_per_temperature

Moves evaluated at each temperature. The thesis uses 90.

Type:

int

max_iterations

Optional hard cap on evaluated moves. None uses the temperature schedule.

Type:

int | None

time_limit_seconds

Optional wall-clock cap. None disables it.

Type:

float | None

seed

RNG seed for reproducibility.

Type:

int

operator_weights

Relative weights for swap, reversion, and insertion. Must be non-negative and not all zero.

Type:

dict[str, float]

overtime_violation_penalty_per_minute

Penalty per minute above the daily horizon. None uses the scenario’s driver policy.

Type:

float | None

cooling_rate: float
initial_temperature: float
iterations_per_temperature: int
max_iterations: int | None
min_temperature_ratio: float
operator_weights: dict[str, float]
overtime_violation_penalty_per_minute: float | None
seed: int
time_limit_seconds: float | None
class fhops.hauling.heuristic.SolutionEvaluation(objective, fixed_cost, variable_cost, overtime_cost, violation_cost, violation_minutes, schedule, truck_summary, loader_summary, plan)[source]

Bases: object

Cost and schedule evaluation of one route solution.

Parameters:

  • objective (float)

  • fixed_cost (float)

  • variable_cost (float)

  • overtime_cost (float)

  • violation_cost (float)

  • violation_minutes (float)

  • schedule (DataFrame)

  • truck_summary (DataFrame)

  • loader_summary (DataFrame)

  • plan (SchedulePlan)

objective

Total cost including fixed, variable, overtime, and overtime-violation penalty.

Type:

float

fixed_cost / variable_cost / overtime_cost / violation_cost

Cost components.

violation_minutes

Total minutes above regular_minutes + max_overtime_minutes.

Type:

float

schedule / truck_summary / loader_summary

DataFrames with the Phase II output schemas.

plan

The decoded route and loader-order plan.

Type:

fhops.hauling.schedule.SchedulePlan

fixed_cost: float
loader_summary: DataFrame
objective: float
overtime_cost: float
plan: SchedulePlan
schedule: DataFrame
truck_summary: DataFrame
variable_cost: float
violation_cost: float
violation_minutes: float
fhops.hauling.heuristic.decode_routes(problem, routes)[source]

Decode a route solution into a loader assignment and SPT loader order.

Parameters:

  • problem (ScheduleProblem) – Scheduling data.

  • routes (dict[str, tuple[str, ...]]) – truck_id -> ordered tuple of load ids. Every load must appear exactly once.

Returns:

Loader assignments and pickup/delivery orders derived from projected arrival times.

Return type:

SchedulePlan

Raises:

ValueError – When a route references an unknown load, a load is missing, or a load appears more than once.

fhops.hauling.heuristic.evaluate_routes(problem, routes, config=None)[source]

Evaluate a route solution with the SPT decoder and cost model.

Parameters:

  • problem (ScheduleProblem) – Scheduling data.

  • routes (dict[str, tuple[str, ...]]) – Truck routes.

  • config (SAConfig | None) – Optional SA configuration for the overtime-violation penalty. None uses the scenario’s driver policy.

Returns:

Cost components, schedule, truck/loader summaries, and the decoded plan.

Return type:

SolutionEvaluation

fhops.hauling.heuristic.initial_solution(problem, rng)[source]

Build a randomized feasible initial route solution.

Loads are visited in a stable order and assigned to the compatible truck with the fewest loads (ties broken randomly). This keeps every load assigned to a compatible truck and spreads work across the fleet before searching.

Parameters:

  • problem (ScheduleProblem)

  • rng (Random)

Return type:

dict[str, tuple[str, …]]

fhops.hauling.heuristic.solve_sa(problem, config=None)[source]

Run the simulated-annealing scheduler.

Parameters:

  • problem (ScheduleProblem) – Scheduling data.

  • config (SAConfig | None) – SA configuration. None uses the thesis-derived defaults.

Returns:

Best schedule, cost components, summaries, search provenance, and history.

Return type:

HaulingSAResult

fhops.hauling.heuristic.summarize_sa_result(result)[source]

Return a compact human-readable SA summary.

Parameters:

result (HaulingSAResult)

Return type:

str

Bridges from FHOPS operational/tactical plans to the hauling contract.

The hauling layer needs supply (volume by block and log sort) and demand (volume by yard and log sort). The operational and tactical models provide these in different shapes. This module turns them into a validated hauling scenario and reports any assumption or residual explicitly; it never silently invents a sort split or a missing arc.

class fhops.hauling.integration.HaulingIntegrationResult(scenario, warnings=<factory>, residuals=<factory>)[source]

Bases: object

A hauling scenario plus the assumptions and residuals behind it.

Parameters:

  • scenario (Scenario)

  • warnings (list[str])

  • residuals (pd.DataFrame)

scenario

Scenario with the updated hauling section.

Type:

Scenario

warnings

Human-readable assumptions, skipped records, and unsupported mappings.

Type:

list[str]

residuals

One row per log sort with total supply, demand, and the residual.

Type:

pd.DataFrame

residuals: pd.DataFrame
scenario: Scenario
warnings: list[str]
fhops.hauling.integration.residual_report(scenario)[source]

Return total supply, total demand, and the residual per log sort.

Parameters:

scenario (Scenario) – Scenario with a hauling section.

Returns:

Columns log_sort_id, supply_m3, demand_m3, and residual_m3. A positive residual means demand exceeds supply.

Return type:

pandas.DataFrame

fhops.hauling.integration.scenario_from_operational_plan(scenario, assignments, *, sort_shares=None, production_column='production')[source]

Build hauling supply from an operational assignment table.

Parameters:

  • scenario (Scenario) – Base scenario with a hauling section.

  • assignments (pd.DataFrame) – DataFrame with block_id and production columns. If production is absent, work_required is used for blocks that appear in the assignments.

  • sort_shares (dict[str, float] | None) – Optional log_sort_id -> fraction split. Fractions must be non-negative and sum to one. When omitted, each log sort receives an equal share of the block volume; this is recorded as a warning because the operational model does not carry product yields.

  • production_column (str) – Name of the production column. Defaults to production.

Returns:

Updated scenario, warnings, and supply/demand residuals.

Return type:

HaulingIntegrationResult

Raises:

ValueError – When the scenario has no hauling section, the assignment table lacks block_id, or the sort split is invalid.

fhops.hauling.integration.scenario_from_tactical_plan(scenario, tactical, *, period_id, production=None)[source]

Build hauling supply and demand from a tactical-operational period.

Parameters:

  • scenario (Scenario) – Base scenario with a hauling section.

  • tactical (TacticalOperationalScenario) – Tactical-operational scenario.

  • period_id (str) – Period to translate.

  • production (pd.DataFrame | None) – Optional tactical production table with block_id, product_id, period_id, and volume_m3. When omitted, supply is left unchanged and a warning is recorded; the model does not guess harvest volume from potential yields.

Returns:

Updated scenario, warnings, and residuals.

Return type:

HaulingIntegrationResult

Raises:

ValueError – When the scenario has no hauling section, no sort yard links the tactical facilities, or the period is not known.

Stable output writers and reports for log-hauling runs.

The filenames in this module are the artefact contract used by the CLI and, later, the web UI. The writers do not recompute a result; they serialise the DataFrames and objective payloads produced by the flow, scheduling, and SA solvers.

fhops.hauling.reporting.build_markdown_report(run_dir)[source]

Build a compact Markdown report from a run directory.

Parameters:

run_dir (Path) – Directory containing haul_summary.json and the stable CSV/JSON artefacts.

Returns:

Markdown report. Missing artefacts are named rather than treated as zero.

Return type:

str

Raises:

FileNotFoundError – When haul_summary.json is absent.

fhops.hauling.reporting.write_flow_run(result, out_dir, *, scenario_name, start_day, run_type='plan')[source]

Write a Phase I result to the stable artefact names.

Parameters:

  • result (FlowSolveResult) – Phase I result.

  • out_dir (Path) – Directory to create/write.

  • scenario_name (str) – Scenario label for the summary.

  • start_day (int) – Parent-scenario start day of the hauling horizon.

  • run_type (str) – plan or solve.

Returns:

Logical name -> written path.

Return type:

dict[str, Path]

fhops.hauling.reporting.write_schedule_run(result, out_dir, *, scenario_name, day, run_type='schedule')[source]

Write a Phase II/III schedule to the stable artefact names.

Parameters:

  • result (ScheduleSolveResult | HaulingSAResult) – Exact or SA scheduling result.

  • out_dir (Path) – Directory to create/write.

  • scenario_name (str) – Scenario label for the summary.

  • day (int) – Relative hauling-horizon day.

  • run_type (str) – schedule or solve.

Returns:

Logical name -> written path.

Return type:

dict[str, Path]

fhops.hauling.reporting.write_solve_run(flow_result, schedule_result, out_dir, *, scenario_name, start_day, day)[source]

Write a combined Phase I + II/III run and merge the summaries.

Parameters:

  • flow_result (FlowSolveResult) – Phase I result.

  • schedule_result (ScheduleSolveResult | HaulingSAResult) – Phase II or III result.

  • out_dir (Path) – Directory to create/write.

  • scenario_name (str) – Scenario label.

  • start_day (int) – Parent-scenario start day of the hauling horizon.

  • day (int) – Relative hauling-horizon day of the schedule.

Returns:

Logical name -> written path, including both phases.

Return type:

dict[str, Path]