fhops.scenario Package

This package hosts the scenario contract, IO utilities, and synthetic dataset generator. Use it when authoring datasets, validating inputs, or programmatically instantiating fhops.scenario.contract.Problem objects before passing them to solvers. The typical workflow is:

  1. Define CSV tables + YAML metadata (see Data Contract Guide).

  2. Load scenarios with fhops.scenario.io.load_scenario().

  3. Create a fhops.scenario.contract.Problem via Problem.from_scenario for use with MIP/heuristics.

  4. Optionally call fhops.scenario.synthetic helpers to generate benchmark datasets.

Scenario package exposing contracts, IO helpers, and synthetic generators.

class fhops.scenario.Block(*, id, landing_id, work_required, earliest_start=1, latest_finish=None, harvest_system_id=None, avg_stem_size_m3=None, volume_per_ha_m3=None, volume_per_ha_m3_sigma=None, stem_density_per_ha=None, stem_density_per_ha_sigma=None, ground_slope_percent=None, salvage_processing_mode=None)[source]

Bases: BaseModel

Harvest block metadata and scheduling window.

Parameters:

  • id (str)

  • landing_id (str)

  • work_required (float)

  • earliest_start (int | None)

  • latest_finish (int | None)

  • harvest_system_id (str | None)

  • avg_stem_size_m3 (float | None)

  • volume_per_ha_m3 (float | None)

  • volume_per_ha_m3_sigma (float | None)

  • stem_density_per_ha (float | None)

  • stem_density_per_ha_sigma (float | None)

  • ground_slope_percent (float | None)

  • salvage_processing_mode (SalvageProcessingMode | None)

id

Unique block identifier (referenced by production rates and assignments).

Type:

str

landing_id

Landing where wood is forwarded; constrains landing daily capacity.

Type:

str

work_required

Volume (m³) that must be delivered from the block to complete it — the terminal (e.g., loader) output measured in the same units as ProductionRate.rate. Every role of an explicit harvest system processes up to this volume; loader truckload batching, playback production, KPIs (total_production, remaining_work_total) and all reference datasets treat it as m³. Must be non-negative.

Type:

float

earliest_start

Optional earliest day (inclusive, 1-indexed) when the block can begin.

Type:

Day | None

latest_finish

Optional latest day (inclusive) when the block must finish.

Type:

Day | None

harvest_system_id

Optional harvest system definition that restricts machine roles per block. Scenario validation requires the id to be a key of Scenario.harvest_systems or of fhops.scheduling.systems.default_system_registry().

Type:

str | None

avg_stem_size_m3 / volume_per_ha_m3 / volume_per_ha_m3_sigma

Stand descriptors (cubic metres) surfaced in analytics and productivity lookups.

stem_density_per_ha / stem_density_per_ha_sigma

Stems per hectare statistics used by some productivity models.

ground_slope_percent

Mean slope (%) for the block — used by productivity heuristics and diagnostics.

Type:

float | None

salvage_processing_mode

Enum describing downstream salvage processing (affects evaluation notes).

Type:

SalvageProcessingMode | None

avg_stem_size_m3: float | None
earliest_start: Day | None
ground_slope_percent: float | None
harvest_system_id: str | None
id: str
landing_id: str
latest_finish: Day | None
model_config = {}

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

salvage_processing_mode: SalvageProcessingMode | None
stem_density_per_ha: float | None
stem_density_per_ha_sigma: float | None
volume_per_ha_m3: float | None
volume_per_ha_m3_sigma: float | None
work_required: float
class fhops.scenario.BlockOperability(block_id, landing_id, terrain_class, current_system_id, current_system_class, current_system_compatible, recommended_system_id, evidence_source, fallbacks, slope_percent, roughness_m, canopy_cover_percent, aspect_degrees, landing_slope_percent, landing_roughness_m, move_in_multiplier, landing_multiplier, road_multiplier, reasons=<factory>)[source]

Bases: object

Advisory operability result for one block.

Parameters:

  • block_id (str)

  • landing_id (str)

  • terrain_class (Literal['ground', 'cable', 'helicopter'])

  • current_system_id (str | None)

  • current_system_class (Literal['ground', 'cable', 'helicopter'] | None)

  • current_system_compatible (bool | None)

  • recommended_system_id (str)

  • evidence_source (str)

  • fallbacks (tuple[str, ...])

  • slope_percent (float)

  • roughness_m (float)

  • canopy_cover_percent (float | None)

  • aspect_degrees (float | None)

  • landing_slope_percent (float | None)

  • landing_roughness_m (float | None)

  • move_in_multiplier (float)

  • landing_multiplier (float)

  • road_multiplier (float)

  • reasons (tuple[str, ...])

aspect_degrees: float | None
block_id: str
canopy_cover_percent: float | None
current_system_class: Literal['ground', 'cable', 'helicopter'] | None
current_system_compatible: bool | None
current_system_id: str | None
evidence_source: str
fallbacks: tuple[str, ...]
landing_id: str
landing_multiplier: float
landing_roughness_m: float | None
landing_slope_percent: float | None
move_in_multiplier: float
reasons: tuple[str, ...]
recommended_system_id: str
road_multiplier: float
roughness_m: float
slope_percent: float
terrain_class: Literal['ground', 'cable', 'helicopter']
to_dict()[source]

Return a JSON-ready payload.

Return type:

dict[str, object]

class fhops.scenario.CalendarEntry(*, machine_id, day, available=1)[source]

Bases: BaseModel

Day-level availability for a machine.

Parameters:
machine_id

Identifier of the machine whose availability is being set.

Type:

str

day

One-indexed day number relative to the scenario horizon.

Type:

Day

available

Binary flag (1 available, 0 unavailable) controlling day-level assignment eligibility.

Type:

int

available: int
day: Day
machine_id: str
model_config = {}

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

fhops.scenario.Day

alias of int

class fhops.scenario.FemicHydrologyArtifact(role, relative_path, schema_version, media_type, size_bytes, sha256, description)[source]

Bases: object

One pinned artifact from a FEMIC hydrology manifest.

Parameters:

  • role (str)

  • relative_path (str)

  • schema_version (str)

  • media_type (str)

  • size_bytes (int)

  • sha256 (str)

  • description (str)

role

Producer artifact role.

Type:

str

relative_path

Path relative to the bundle root.

Type:

str

schema_version

Artifact schema recorded by the producer.

Type:

str

media_type

Artifact media type.

Type:

str

size_bytes

Declared artifact size.

Type:

int

sha256

Declared artifact SHA-256 digest.

Type:

str

description

Producer description.

Type:

str

description: str
media_type: str
relative_path: str
role: str
schema_version: str
sha256: str
size_bytes: int
exception fhops.scenario.FemicHydrologyAssetError[source]

Bases: RuntimeError

Raised when a road/culvert asset table cannot be consumed or mapped.

class fhops.scenario.FemicHydrologyBundle(bundle_dir, manifest_path, bundle_id, bundle_version, generated_utc, domain_id, ledger_id, response_surface_id, manifest_sha256, artifacts, response_surface, flood_risk, applicability_domain, consumer_contract, caveats, total_bytes)[source]

Bases: object

Validated and pinned FEMIC hydrology bundle metadata.

Parameters:
bundle_dir

Resolved bundle root.

Type:

Path

manifest_path

Resolved manifest path.

Type:

Path

bundle_id

Producer bundle identifier.

Type:

str

bundle_version

Producer bundle version.

Type:

str

generated_utc

Producer generation timestamp.

Type:

str

domain_id

FEMIC domain identifier.

Type:

str

ledger_id

FEMIC disturbance-ledger identifier.

Type:

str

response_surface_id

FEMIC response-surface identifier.

Type:

str

manifest_sha256

SHA-256 digest of the manifest bytes.

Type:

str

artifacts

Artifacts keyed by role.

Type:

dict[str, FemicHydrologyArtifact]

response_surface

Producer response-surface summary.

Type:

dict[str, object]

flood_risk

Producer flood-risk summary.

Type:

dict[str, object]

applicability_domain

Producer applicability domain.

Type:

dict[str, object]

consumer_contract

Producer consumer-contract metadata.

Type:

dict[str, object]

caveats

Producer caveats.

Type:

tuple[str, …]

total_bytes

Sum of declared artifact sizes.

Type:

int

applicability_domain: dict[str, object]
artifacts: dict[str, FemicHydrologyArtifact]
bundle_dir: Path
bundle_id: str
bundle_version: str
caveats: tuple[str, ...]
consumer_contract: dict[str, object]
domain_id: str
flood_risk: dict[str, object]
generated_utc: str
ledger_id: str
manifest_path: Path
manifest_sha256: str
response_surface: dict[str, object]
response_surface_id: str
total_bytes: int
exception fhops.scenario.FemicHydrologyConstraintError[source]

Bases: RuntimeError

A schedule cannot be checked against hazard evidence.

exception fhops.scenario.FemicHydrologyError[source]

Bases: RuntimeError

Raised when a FEMIC hydrology bundle cannot be consumed.

exception fhops.scenario.FemicHydrologyExposureError[source]

Bases: RuntimeError

Raised when a road/culvert exposure table cannot be consumed.

exception fhops.scenario.FemicHydrologyHazardError[source]

Bases: RuntimeError

Raised when a hazard-window table cannot be consumed.

exception fhops.scenario.FemicHydrologyMappingError[source]

Bases: RuntimeError

A hydrology crosswalk cannot be loaded or applied to a scenario.

class fhops.scenario.FemicHydrologyReadiness(status, validation_report_id, model_card_id, review_status, full_process_pilot_ready, hydrologist_review_recorded, tail_check_failures, usable_fits, benchmark_cases, benchmark_passed, benchmark_failed, benchmark_not_evaluated, warnings, caveats, evidence_sha256)[source]

Bases: object

Validation/model-card evidence for a pinned hydrology bundle.

Parameters:

  • status (str)

  • validation_report_id (str | None)

  • model_card_id (str | None)

  • review_status (str | None)

  • full_process_pilot_ready (bool | None)

  • hydrologist_review_recorded (bool | None)

  • tail_check_failures (dict[str, list[str]])

  • usable_fits (dict[str, list[str]])

  • benchmark_cases (int)

  • benchmark_passed (int)

  • benchmark_failed (int)

  • benchmark_not_evaluated (int)

  • warnings (tuple[str, ...])

  • caveats (tuple[str, ...])

  • evidence_sha256 (str)

status

not_available, provisional, conditionally_ready, ready, or blocked.

Type:

str

validation_report_id

Validation-report identifier, when packaged.

Type:

str | None

model_card_id

Model-card identifier, when packaged.

Type:

str | None

review_status

Producer model-card review status.

Type:

str | None

full_process_pilot_ready

Producer full-pilot readiness flag.

Type:

bool | None

hydrologist_review_recorded

Producer review-record flag.

Type:

bool | None

tail_check_failures

Tail-check failures keyed by gauge/station.

Type:

dict[str, list[str]]

usable_fits

Usable fit methods keyed by gauge/station.

Type:

dict[str, list[str]]

benchmark_cases

Published benchmark count.

Type:

int

benchmark_passed

Passed benchmark count.

Type:

int

benchmark_failed

Failed benchmark count.

Type:

int

benchmark_not_evaluated

Not-evaluated benchmark count.

Type:

int

warnings

Non-fatal evidence warnings.

Type:

tuple[str, …]

caveats

Evidence caveats.

Type:

tuple[str, …]

evidence_sha256

Deterministic digest of the evidence record.

Type:

str

benchmark_cases: int
benchmark_failed: int
benchmark_not_evaluated: int
benchmark_passed: int
caveats: tuple[str, ...]
evidence_sha256: str
full_process_pilot_ready: bool | None
hydrologist_review_recorded: bool | None
model_card_id: str | None
review_status: str | None
status: str
tail_check_failures: dict[str, list[str]]
to_dict()[source]

Return the readiness report as a JSON-serializable mapping.

Returns:

Status, evidence identifiers, failures, fits, benchmarks, and caveats.

Return type:

dict[str, object]

usable_fits: dict[str, list[str]]
validation_report_id: str | None
warnings: tuple[str, ...]
exception fhops.scenario.FemicHydrologyRegistrationError[source]

Bases: RuntimeError

A hydrology bundle cannot be bound to a scenario or written safely.

exception fhops.scenario.FemicHydrologyScreenError[source]

Bases: RuntimeError

A schedule cannot be screened against the hydrology evidence.

class fhops.scenario.FemicHydrologyValidation(bundle_dir, is_valid, checked_utc, errors, warnings, artifact_count, total_bytes)[source]

Bases: object

Validation report for a FEMIC hydrology bundle.

Parameters:
bundle_dir

Resolved bundle root.

Type:

Path

is_valid

Whether every required check passed.

Type:

bool

checked_utc

Report timestamp.

Type:

str

errors

Blocking findings.

Type:

tuple[str, …]

warnings

Non-blocking findings.

Type:

tuple[str, …]

artifact_count

Number of artifacts inspected.

Type:

int

total_bytes

Sum of declared artifact sizes.

Type:

int

artifact_count: int
bundle_dir: Path
checked_utc: str
errors: tuple[str, ...]
is_valid: bool
to_dict()[source]

Return the validation report as a JSON-serializable mapping.

Returns:

Bundle path, validity, findings, counts, and size.

Return type:

dict[str, object]

total_bytes: int
warnings: tuple[str, ...]
class fhops.scenario.FemicLidarBundle(bundle_dir, manifest_path, bundle_id, bundle_version, source, selection, processing, products, manifest_sha256, total_bytes)[source]

Bases: object

Pinned FEMIC LidarBC bundle metadata.

Parameters:
bundle_dir: Path
bundle_id: str
bundle_version: str
manifest_path: Path
manifest_sha256: str
processing: dict[str, object]
products: dict[str, FemicLidarProduct]
selection: dict[str, object]
source: dict[str, object]
total_bytes: int
exception fhops.scenario.FemicLidarError[source]

Bases: RuntimeError

Raised when a FEMIC LidarBC bundle cannot be consumed.

class fhops.scenario.FemicLidarProduct(product, relative_path, sha256, size_bytes, crs, resolution_m, vertical_datum, licence_id, source_tile_ids, derived, lineage)[source]

Bases: object

One pinned product record from a FEMIC bundle manifest.

Parameters:
crs: str
derived: bool
licence_id: str
lineage: tuple[str, ...]
product: str
relative_path: str
resolution_m: float | None
sha256: str
size_bytes: int
source_tile_ids: tuple[str, ...]
vertical_datum: str
class fhops.scenario.FemicLidarValidation(bundle_dir, is_valid, checked_utc, errors, warnings, product_count, total_bytes)[source]

Bases: object

Validation report for a FEMIC LidarBC bundle.

Parameters:
bundle_dir: Path
checked_utc: str
errors: tuple[str, ...]
is_valid: bool
product_count: int
to_dict()[source]

Return the validation report as a JSON-serializable mapping.

Return type:

dict[str, object]

total_bytes: int
warnings: tuple[str, ...]
class fhops.scenario.HydrologyAsset(asset_id, asset_type, watershed_id, road_job_id, vulnerability_class, notes)[source]

Bases: object

One explicit road or culvert asset.

Parameters:

  • asset_id (str)

  • asset_type (str)

  • watershed_id (str)

  • road_job_id (str | None)

  • vulnerability_class (str)

  • notes (str)

asset_id

Stable asset identifier.

Type:

str

asset_type

road or culvert.

Type:

str

watershed_id

FEMIC watershed identifier.

Type:

str

road_job_id

Scenario road-construction id for a road asset.

Type:

str | None

vulnerability_class

low, moderate, high, or unknown.

Type:

str

notes

Human notes.

Type:

str

asset_id: str
asset_type: str
notes: str
road_job_id: str | None
to_dict()[source]

Return the asset as a JSON-serializable mapping.

Returns:

Asset fields.

Return type:

dict[str, object]

vulnerability_class: str
watershed_id: str
class fhops.scenario.HydrologyAssetMapping(inventory_id, generated_utc, source_id, review_status, reviewer_id, assets, road_job_ids, watershed_ids, warnings, caveats, assets_sha256)[source]

Bases: object

Assets resolved against a scenario and hydrology mapping.

Parameters:
inventory_id

Asset inventory identifier.

Type:

str

generated_utc

Inventory timestamp.

Type:

str

source_id

Asset source.

Type:

str

review_status

Review state.

Type:

str

reviewer_id

Reviewer when reviewed.

Type:

str | None

assets

Validated assets.

Type:

tuple[HydrologyAsset, …]

road_job_ids

Scenario road-job ids referenced by road assets.

Type:

tuple[str, …]

watershed_ids

Watersheds referenced by assets.

Type:

tuple[str, …]

warnings

Non-fatal observations.

Type:

tuple[str, …]

caveats

Evidence caveats.

Type:

tuple[str, …]

assets_sha256

Asset file digest.

Type:

str

assets: tuple[HydrologyAsset, ...]
assets_sha256: str
caveats: tuple[str, ...]
generated_utc: str
inventory_id: str
review_status: str
reviewer_id: str | None
road_job_ids: tuple[str, ...]
source_id: str
to_dict()[source]

Return the mapped inventory as a JSON-serializable mapping.

Returns:

Inventory, assets, coverage, warnings, and caveats.

Return type:

dict[str, object]

warnings: tuple[str, ...]
watershed_ids: tuple[str, ...]
class fhops.scenario.HydrologyAssetValidation(path, is_valid, checked_utc, errors, warnings, asset_count, assets_sha256)[source]

Bases: object

Validation report for an asset table.

Parameters:
path

Asset JSON path.

Type:

Path

is_valid

Whether every required check passed.

Type:

bool

checked_utc

Report timestamp.

Type:

str

errors

Blocking findings.

Type:

tuple[str, …]

warnings

Non-blocking findings.

Type:

tuple[str, …]

asset_count

Parsed asset count.

Type:

int

assets_sha256

File digest.

Type:

str

asset_count: int
assets_sha256: str
checked_utc: str
errors: tuple[str, ...]
is_valid: bool
path: Path
to_dict()[source]

Return the validation report as a JSON-serializable mapping.

Returns:

Path, validity, findings, count, and digest.

Return type:

dict[str, object]

warnings: tuple[str, ...]
class fhops.scenario.HydrologyConstraintCheck(check_id, generated_utc, scenario_path, hazard_id, mapping_id, schedule_sha256, events, failed, failure_reasons, coverage, warnings, caveats)[source]

Bases: object

Result of checking one schedule against hazard windows.

Parameters:
check_id

Deterministic check identifier.

Type:

str

generated_utc

Check timestamp.

Type:

str

scenario_path

Scenario YAML.

Type:

Path

hazard_id

Hazard table identifier.

Type:

str

mapping_id

Hydrology mapping identifier.

Type:

str

schedule_sha256

Assignments CSV digest.

Type:

str

events

All watch/warning/closure events.

Type:

tuple[HydrologyConstraintEvent, …]

failed

Whether the selected policy treats an event as a failure.

Type:

bool

failure_reasons

Reasons the check failed.

Type:

tuple[str, …]

coverage

Assignment, day, block, and event coverage.

Type:

dict[str, object]

warnings

Non-fatal observations.

Type:

tuple[str, …]

caveats

Operational-only caveats.

Type:

tuple[str, …]

caveats: tuple[str, ...]
check_id: str
coverage: dict[str, object]
events: tuple[HydrologyConstraintEvent, ...]
failed: bool
failure_reasons: tuple[str, ...]
generated_utc: str
hazard_id: str
mapping_id: str
scenario_path: Path
schedule_sha256: str
to_dict()[source]

Return the check result as a JSON-serializable mapping.

Returns:

Identity, events, failure state, coverage, and caveats.

Return type:

dict[str, object]

warnings: tuple[str, ...]
class fhops.scenario.HydrologyConstraintEvent(day, shift_id, machine_id, block_id, watershed_id, severity, asset_ids, reason)[source]

Bases: object

One operational activity inside a hazard window.

Parameters:

  • day (int)

  • shift_id (str | None)

  • machine_id (str | None)

  • block_id (str)

  • watershed_id (str)

  • severity (str)

  • asset_ids (tuple[str, ...])

  • reason (str)

day

One-indexed schedule day.

Type:

int

shift_id

Optional shift label.

Type:

str | None

machine_id

Optional machine identifier.

Type:

str | None

block_id

Scheduled block identifier.

Type:

str

watershed_id

FEMIC watershed identifier.

Type:

str

severity

watch, warning, or closure.

Type:

str

asset_ids

Assets in the same watershed.

Type:

tuple[str, …]

reason

Human-readable exposure reason.

Type:

str

asset_ids: tuple[str, ...]
block_id: str
day: int
machine_id: str | None
reason: str
severity: str
shift_id: str | None
to_dict()[source]

Return the event as a JSON-serializable mapping.

Returns:

Event fields.

Return type:

dict[str, object]

watershed_id: str
class fhops.scenario.HydrologyCrosswalk(path, sha256, rows, metadata)[source]

Bases: object

Loaded and checksummed explicit crosswalk.

Parameters:

  • path (Path)

  • sha256 (str)

  • rows (tuple[HydrologyCrosswalkRow, ...])

  • metadata (HydrologyCrosswalkMetadata)

path

Crosswalk CSV path.

Type:

Path

sha256

SHA-256 of the CSV bytes.

Type:

str

rows

Mapping rows.

Type:

tuple[HydrologyCrosswalkRow, …]

metadata

Review metadata.

Type:

HydrologyCrosswalkMetadata

metadata: HydrologyCrosswalkMetadata
path: Path
rows: tuple[HydrologyCrosswalkRow, ...]
sha256: str
class fhops.scenario.HydrologyCrosswalkMetadata(crosswalk_id, created_utc, review_status, reviewed_by, method, source, caveats)[source]

Bases: object

Review metadata for an explicit hydrology crosswalk.

Parameters:

  • crosswalk_id (str)

  • created_utc (str)

  • review_status (str)

  • reviewed_by (str | None)

  • method (str)

  • source (str)

  • caveats (tuple[str, ...])

crosswalk_id

Stable crosswalk identifier.

Type:

str

created_utc

Creation timestamp.

Type:

str

review_status

not_reviewed, review_pending, or reviewed.

Type:

str

reviewed_by

Reviewer identifier when reviewed.

Type:

str | None

method

Mapping method or process.

Type:

str

source

Source document or process.

Type:

str

caveats

Review caveats.

Type:

tuple[str, …]

caveats: tuple[str, ...]
created_utc: str
crosswalk_id: str
method: str
review_status: str
reviewed_by: str | None
source: str
class fhops.scenario.HydrologyCrosswalkRow(block_id, landing_id, road_id, unit_id, watershed_id, receptor_id, notes)[source]

Bases: object

One explicit FHOPS support to FEMIC unit/watershed mapping.

Parameters:

  • block_id (str | None)

  • landing_id (str | None)

  • road_id (str | None)

  • unit_id (str)

  • watershed_id (str)

  • receptor_id (str | None)

  • notes (str)

block_id

FHOPS block identifier, when the row is block-level.

Type:

str | None

landing_id

FHOPS landing identifier, when the row is landing-level.

Type:

str | None

road_id

FHOPS road-construction identifier, when the row is road-level.

Type:

str | None

unit_id

FEMIC disturbance-ledger unit identifier.

Type:

str

watershed_id

FEMIC watershed identifier.

Type:

str

receptor_id

Optional FEMIC flood-risk receptor identifier.

Type:

str | None

notes

Human review notes.

Type:

str

block_id: str | None
landing_id: str | None
notes: str
receptor_id: str | None
road_id: str | None
support_key()[source]

Return the support unit and identifier.

Returns:

("block"|"landing"|"road", id).

Return type:

tuple[str, str]

unit_id: str
watershed_id: str
class fhops.scenario.HydrologyExposure(path, sha256, exposure_id, generated_utc, source_id, review_status, reviewer_id, schedule_sha256, watershed_ids, periods, rows, caveats)[source]

Bases: object

Loaded and checksummed schedule-bound exposure table.

Parameters:

  • path (Path)

  • sha256 (str)

  • exposure_id (str)

  • generated_utc (str)

  • source_id (str)

  • review_status (str)

  • reviewer_id (str | None)

  • schedule_sha256 (str)

  • watershed_ids (tuple[str, ...])

  • periods (tuple[int, ...])

  • rows (tuple[HydrologyExposureRow, ...])

  • caveats (tuple[str, ...])

path

Exposure JSON path.

Type:

Path

sha256

SHA-256 digest of the file bytes.

Type:

str

exposure_id

Stable exposure identifier.

Type:

str

generated_utc

Producer generation timestamp.

Type:

str

source_id

Source or process that produced the table.

Type:

str

review_status

not_reviewed, review_pending, or reviewed.

Type:

str

reviewer_id

Reviewer identifier when reviewed.

Type:

str | None

schedule_sha256

SHA-256 of the bound FHOPS schedule file.

Type:

str

watershed_ids

Declared watersheds.

Type:

tuple[str, …]

periods

Declared periods, sorted ascending.

Type:

tuple[int, …]

rows

Complete watershed/period grid.

Type:

tuple[HydrologyExposureRow, …]

caveats

Interpretation caveats.

Type:

tuple[str, …]

caveats: tuple[str, ...]
exposure_id: str
features_for(watershed_id, period, *, watershed_area_ha=None)[source]

Return raw and optional density features for one watershed/period.

Parameters:

  • watershed_id (str) – Declared watershed identifier.

  • period (int) – Declared period.

  • watershed_area_ha (float | None, optional) – FEMIC watershed area. When positive, calculated densities are added.

Returns:

Raw exposure features and, when an area is supplied, density features.

Return type:

dict[str, float]

Raises:

FemicHydrologyExposureError – If the row is missing or the area is not positive when supplied.

generated_utc: str
path: Path
periods: tuple[int, ...]
review_status: str
reviewer_id: str | None
row(watershed_id, period)[source]

Return one watershed/period row.

Parameters:

  • watershed_id (str) – Declared watershed identifier.

  • period (int) – Declared period.

Returns:

Matching row.

Return type:

HydrologyExposureRow

Raises:

FemicHydrologyExposureError – If the watershed/period pair is not declared.

rows: tuple[HydrologyExposureRow, ...]
schedule_sha256: str
sha256: str
source_id: str
to_dict()[source]

Return the exposure table as a JSON-serializable mapping.

Returns:

Schema, identity, review, schedule binding, rows, and caveats.

Return type:

dict[str, object]

watershed_ids: tuple[str, ...]
class fhops.scenario.HydrologyExposureRow(watershed_id, period, road_built_km, road_used_km, road_decommissioned_km, culvert_count, culvert_exposed_count)[source]

Bases: object

One watershed/period exposure record.

Parameters:

  • watershed_id (str)

  • period (int)

  • road_built_km (float)

  • road_used_km (float)

  • road_decommissioned_km (float)

  • culvert_count (int)

  • culvert_exposed_count (int)

watershed_id

FEMIC watershed identifier.

Type:

str

period

Non-negative screening period.

Type:

int

road_built_km

Road length built in the period.

Type:

float

road_used_km

Road length used in the period.

Type:

float

road_decommissioned_km

Road length decommissioned in the period.

Type:

float

culvert_count

Culvert count in the watershed.

Type:

int

culvert_exposed_count

Culverts exposed to the flood path.

Type:

int

culvert_count: int
culvert_exposed_count: int
period: int
road_built_km: float
road_decommissioned_km: float
road_used_km: float
to_dict()[source]

Return the row in the shared exposure-table shape.

Returns:

Watershed, period, road lengths, and culvert counts.

Return type:

dict[str, object]

watershed_id: str
class fhops.scenario.HydrologyExposureValidation(path, is_valid, checked_utc, errors, warnings, row_count, watershed_count, period_count, exposure_sha256, schedule_sha256)[source]

Bases: object

Validation report for a road/culvert exposure table.

Parameters:
path

Exposure path.

Type:

Path

is_valid

Whether every required check passed.

Type:

bool

checked_utc

Report timestamp.

Type:

str

errors

Blocking findings.

Type:

tuple[str, …]

warnings

Non-blocking findings.

Type:

tuple[str, …]

row_count

Number of parsed rows.

Type:

int

watershed_count

Number of declared watersheds.

Type:

int

period_count

Number of declared periods.

Type:

int

exposure_sha256

File digest.

Type:

str

schedule_sha256

Declared schedule digest.

Type:

str | None

checked_utc: str
errors: tuple[str, ...]
exposure_sha256: str
is_valid: bool
path: Path
period_count: int
row_count: int
schedule_sha256: str | None
to_dict()[source]

Return the validation report as a JSON-serializable mapping.

Returns:

Path, validity, findings, counts, and digests.

Return type:

dict[str, object]

warnings: tuple[str, ...]
watershed_count: int
class fhops.scenario.HydrologyHazard(path, sha256, hazard_id, generated_utc, source_id, review_status, reviewer_id, windows, caveats)[source]

Bases: object

Loaded and checksummed hazard-window table.

Parameters:

  • path (Path)

  • sha256 (str)

  • hazard_id (str)

  • generated_utc (str)

  • source_id (str)

  • review_status (str)

  • reviewer_id (str | None)

  • windows (tuple[HydrologyHazardWindow, ...])

  • caveats (tuple[str, ...])

path

Hazard JSON path.

Type:

Path

sha256

SHA-256 digest of the file bytes.

Type:

str

hazard_id

Stable hazard identifier.

Type:

str

generated_utc

Producer generation timestamp.

Type:

str

source_id

Source or process that produced the windows.

Type:

str

review_status

not_reviewed, review_pending, or reviewed.

Type:

str

reviewer_id

Reviewer identifier when reviewed.

Type:

str | None

windows

Validated windows.

Type:

tuple[HydrologyHazardWindow, …]

caveats

Interpretation caveats.

Type:

tuple[str, …]

caveats: tuple[str, ...]
generated_utc: str
hazard_id: str
path: Path
review_status: str
reviewer_id: str | None
severity_for(watershed_id, day)[source]

Return the highest severity covering a watershed/day.

Parameters:

  • watershed_id (str) – FEMIC watershed identifier.

  • day (int) – One-indexed day.

Returns:

closure, warning, watch, or None when no window covers the pair.

Return type:

str | None

sha256: str
source_id: str
to_dict()[source]

Return the hazard table as a JSON-serializable mapping.

Returns:

Schema, identity, review, windows, and caveats.

Return type:

dict[str, object]

windows: tuple[HydrologyHazardWindow, ...]
class fhops.scenario.HydrologyHazardValidation(path, is_valid, checked_utc, errors, warnings, window_count, watershed_count, hazard_sha256)[source]

Bases: object

Validation report for a hazard-window table.

Parameters:
path

Hazard path.

Type:

Path

is_valid

Whether every required check passed.

Type:

bool

checked_utc

Report timestamp.

Type:

str

errors

Blocking findings.

Type:

tuple[str, …]

warnings

Non-blocking findings.

Type:

tuple[str, …]

window_count

Number of parsed windows.

Type:

int

watershed_count

Number of distinct watersheds.

Type:

int

hazard_sha256

File digest.

Type:

str

checked_utc: str
errors: tuple[str, ...]
hazard_sha256: str
is_valid: bool
path: Path
to_dict()[source]

Return the validation report as a JSON-serializable mapping.

Returns:

Path, validity, findings, counts, and digest.

Return type:

dict[str, object]

warnings: tuple[str, ...]
watershed_count: int
window_count: int
class fhops.scenario.HydrologyHazardWindow(watershed_id, start_day, end_day, severity, return_period_years, source_id, notes)[source]

Bases: object

One watershed/day hazard window.

Parameters:

  • watershed_id (str)

  • start_day (int)

  • end_day (int)

  • severity (str)

  • return_period_years (float | None)

  • source_id (str | None)

  • notes (str)

watershed_id

FEMIC watershed identifier.

Type:

str

start_day

Inclusive first day (1-indexed).

Type:

int

end_day

Inclusive last day.

Type:

int

severity

watch, warning, or closure.

Type:

str

return_period_years

Optional return period associated with the hazard.

Type:

float | None

source_id

Optional source identifier for this window.

Type:

str | None

notes

Human notes.

Type:

str

covers(day)[source]

Return whether the window covers a day.

Parameters:

day (int) – One-indexed day.

Returns:

True when start_day <= day <= end_day.

Return type:

bool

end_day: int
notes: str
return_period_years: float | None
severity: str
source_id: str | None
start_day: int
to_dict()[source]

Return the window as a JSON-serializable mapping.

Returns:

Window fields.

Return type:

dict[str, object]

watershed_id: str
class fhops.scenario.HydrologyMapping(mapping_id, generated_utc, scenario_path, bundle_id, bundle_version, manifest_sha256, crosswalk_id, crosswalk_sha256, review_status, rows, coverage, warnings, caveats)[source]

Bases: object

Resolved scenario-to-FEMIC mapping and coverage.

Parameters:
mapping_id

Deterministic mapping identifier.

Type:

str

generated_utc

Mapping timestamp.

Type:

str

scenario_path

Scenario YAML.

Type:

Path

bundle_id

FEMIC bundle identifier.

Type:

str

bundle_version

FEMIC bundle version.

Type:

str

manifest_sha256

FEMIC manifest digest.

Type:

str

crosswalk_id

Crosswalk identifier.

Type:

str

crosswalk_sha256

Crosswalk CSV digest.

Type:

str

review_status

Crosswalk review state.

Type:

str

rows

Resolved rows.

Type:

tuple[HydrologyMappingRow, …]

coverage

Counts and unmapped support ids.

Type:

dict[str, object]

warnings

Non-fatal observations.

Type:

tuple[str, …]

caveats

Mapping and screening caveats.

Type:

tuple[str, …]

bundle_id: str
bundle_version: str
caveats: tuple[str, ...]
coverage: dict[str, object]
crosswalk_id: str
crosswalk_sha256: str
generated_utc: str
manifest_sha256: str
mapping_id: str
review_status: str
rows: tuple[HydrologyMappingRow, ...]
scenario_path: Path
to_dict()[source]

Return the mapping manifest as a JSON-serializable mapping.

Returns:

Identity, row count, rows, coverage, warnings, and caveats.

Return type:

dict[str, object]

warnings: tuple[str, ...]
class fhops.scenario.HydrologyMappingRow(support_unit, support_id, unit_id, watershed_id, receptor_id, notes)[source]

Bases: object

One resolved support mapping.

Parameters:

  • support_unit (str)

  • support_id (str)

  • unit_id (str)

  • watershed_id (str)

  • receptor_id (str | None)

  • notes (str)

support_unit

block, landing, or road.

Type:

str

support_id

FHOPS support identifier.

Type:

str

unit_id

FEMIC unit identifier.

Type:

str

watershed_id

FEMIC watershed identifier.

Type:

str

receptor_id

Optional FEMIC receptor identifier.

Type:

str | None

notes

Human review notes.

Type:

str

notes: str
receptor_id: str | None
support_id: str
support_unit: str
to_dict()[source]

Return the mapping row as a JSON-serializable mapping.

Returns:

Support unit/id, FEMIC IDs, receptor, and notes.

Return type:

dict[str, object]

unit_id: str
watershed_id: str
fhops.scenario.HydrologyRegistrationResult

alias of RegistrationResult

class fhops.scenario.HydrologyScreen(screen_id, generated_utc, scenario_path, bundle_id, bundle_version, manifest_sha256, mapping_id, exposure_id, exposure_sha256, schedule_sha256, readiness_status, policy, rows, refusals, coverage, warnings, caveats)[source]

Bases: object

One-schedule hydrology screening result.

Parameters:
screen_id

Deterministic screening identifier.

Type:

str

generated_utc

Screening timestamp.

Type:

str

scenario_path

Scenario YAML.

Type:

Path

bundle_id

FEMIC bundle identifier.

Type:

str

bundle_version

FEMIC bundle version.

Type:

str

manifest_sha256

FEMIC manifest digest.

Type:

str

mapping_id

Crosswalk mapping identifier.

Type:

str

exposure_id

Exposure table identifier.

Type:

str

schedule_sha256

Assignments CSV digest.

Type:

str

readiness_status

Bundle readiness state at screening time.

Type:

str

policy

Readiness policy applied.

Type:

str

rows

Inside-domain screening rows.

Type:

tuple[dict[str, object], …]

refusals

Refused schedule/watershed/return-period combinations.

Type:

tuple[dict[str, object], …]

coverage

Period, watershed, block, and row coverage.

Type:

dict[str, object]

warnings

Non-fatal observations.

Type:

tuple[str, …]

caveats

Screening-only caveats.

Type:

tuple[str, …]

bundle_id: str
bundle_version: str
caveats: tuple[str, ...]
coverage: dict[str, object]
exposure_id: str
exposure_sha256: str
generated_utc: str
manifest_sha256: str
mapping_id: str
policy: str
readiness_status: str
refusals: tuple[dict[str, object], ...]
rows: tuple[dict[str, object], ...]
scenario_path: Path
schedule_sha256: str
screen_id: str
to_dict()[source]

Return the screening manifest as a JSON-serializable mapping.

Returns:

Identity, readiness, rows, refusals, coverage, and caveats.

Return type:

dict[str, object]

warnings: tuple[str, ...]
class fhops.scenario.Landing(*, id, daily_capacity=2)[source]

Bases: BaseModel

Landing metadata including its machine capacity.

Parameters:

  • id (str)

  • daily_capacity (int)

id

Landing identifier referenced by blocks and mobilisation logic.

Type:

str

daily_capacity

Maximum number of machines that can work the landing’s blocks concurrently, counted per shift slot (day, shift_id) (on single-shift scenarios: per day). The name is kept for compatibility. The heuristics and the operational MILP apply the same rule: when ObjectiveWeights.landing_surplus is 0 (default) the capacity is hard (heuristics: the repair keeps landings within capacity and any remaining extra machine costs the hard violation penalty, at least 1000, #140; MILP: constraint), otherwise each slot’s k-th machine beyond capacity costs k × landing_surplus. Before FHOPS 1.0.1 the operational MILP counted machine-shifts per day with a slack that was free at weight 0 (#125).

Type:

int

daily_capacity: int
id: str
model_config = {}

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

class fhops.scenario.LidarPolygonSummary(polygon_id, product, valid_pixels, statistics)[source]

Bases: object

One polygon-level support summary from a FEMIC product.

Parameters:
polygon_id: str
product: str
statistics: dict[str, float]
to_dict()[source]

Return the summary as a JSON-serializable mapping.

Return type:

dict[str, object]

valid_pixels: int
class fhops.scenario.LidarProductPin(*, product, relative_path, sha256, crs, resolution_m=None, vertical_datum, source_tile_ids=<factory>)[source]

Bases: BaseModel

One product file recorded in a scenario bundle pin.

Parameters:
crs: str
model_config = {}

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

product: str
relative_path: str
resolution_m: float | None
sha256: str
source_tile_ids: list[str]
vertical_datum: str
class fhops.scenario.LidarProductRef(*, bundle_id, bundle_version, manifest_sha256=None, licence_id=None, manifest_path=None, crs=None, aoi_bbox=None, selection=None, products=None)[source]

Bases: BaseModel

A scenario-level pin to one FEMIC bundle version.

The fields mirror fhops.scenario.lidar.bundle_pin(); manifest_sha256 is the digest of the bundle manifest, not of an individual raster.

Parameters:
aoi_bbox: tuple[float, float, float, float] | None
bundle_id: str
bundle_version: str
crs: str | None
licence_id: str | None
manifest_path: str | None
manifest_sha256: str | None
model_config = {}

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

products: list[LidarProductPin] | None
selection: dict[str, object] | None
class fhops.scenario.LidarSupport(*, support_unit, support_id, bundle_id, bundle_version, product, aggregation='area_weighted_mean', native_resolution_m, support_area_m2=None, valid_pixels=None, slope_percent=None, aspect_degrees=None, roughness_m=None, canopy_cover_percent=None, canopy_height_m=None, vertical_structure_index=None, vertical_structure_class=None, provenance=None)[source]

Bases: BaseModel

Aggregated LidarBC evidence for one block, landing, or road.

The values are the output of a FEMIC raster summary (for example fhops.scenario.lidar.summarize_femic_lidar_for_polygons()). Native resolution, product name, aggregation, valid-pixel count, and support footprint preserve how the summary was made.

Parameters:
aggregation: AggregationMethod
aspect_degrees: float | None
bundle_id: str
bundle_version: str
canopy_cover_percent: float | None
canopy_height_m: float | None
model_config = {}

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

native_resolution_m: float
product: str
provenance: str | None
roughness_m: float | None
slope_percent: float | None
support_area_m2: float | None
support_id: str
support_unit: LidarSupportUnit
valid_pixels: int | None
vertical_structure_class: str | None
vertical_structure_index: float | None
class fhops.scenario.Machine(*, id, crew=None, daily_hours=24.0, operating_cost=0.0, role=None, repair_usage_hours=None)[source]

Bases: BaseModel

Machine definition (identifier, crew, availability, and costing metadata).

Parameters:

  • id (str)

  • crew (str | None)

  • daily_hours (float)

  • operating_cost (float)

  • role (str | None)

  • repair_usage_hours (int | None)

id

Unique machine identifier referenced throughout calendars/assignments.

Type:

str

crew

Optional crew label for reporting/telemetry grouping.

Type:

str | None

daily_hours

Maximum hours the machine can operate per day (defaults to 24).

Type:

float

operating_cost

Cost per scheduled machine hour (SMH) expressed in scenario currency units.

Type:

float

role

Optional machine role string (normalised via normalize_machine_role) used by harvest systems and rental-rate lookups.

Type:

str | None

repair_usage_hours

Optional cumulative repair hours that influences the rental-rate defaults.

Type:

int | None

crew: str | None
daily_hours: float
id: str
model_config = {}

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

operating_cost: float
repair_usage_hours: int | None
role: str | None
class fhops.scenario.MappingResult(scenario_path, bundle_id, bundle_version, rows, warnings, written)[source]

Bases: object

Outcome of mapping FEMIC products into a scenario.

Parameters:
bundle_id: str
bundle_version: str
rows: tuple[LidarSupport, ...]
scenario_path: Path
to_dict()[source]

Return a JSON-ready summary.

Return type:

dict[str, object]

warnings: tuple[str, ...]
written: bool
class fhops.scenario.OperabilityRules(ground_max_slope_percent=35.0, ground_max_roughness_m=0.35, cable_max_slope_percent=60.0, cable_max_roughness_m=0.55, fallback_slope_percent=20.0, fallback_roughness_m=0.2)[source]

Bases: object

Assumption thresholds used by the advisory classifier.

These are planning assumptions, not field-validated rules. They are intentionally conservative and visible in every report.

Parameters:

  • ground_max_slope_percent (float)

  • ground_max_roughness_m (float)

  • cable_max_slope_percent (float)

  • cable_max_roughness_m (float)

  • fallback_slope_percent (float)

  • fallback_roughness_m (float)

cable_max_roughness_m: float = 0.55
cable_max_slope_percent: float = 60.0
fallback_roughness_m: float = 0.2
fallback_slope_percent: float = 20.0
ground_max_roughness_m: float = 0.35
ground_max_slope_percent: float = 35.0
class fhops.scenario.Problem(**data)[source]

Bases: BaseModel

Runtime representation of a scenario used by solvers.

Problem wraps a validated Scenario and expands it into concrete days and shifts so optimisation code can iterate over deterministic index sets without repeatedly querying the Scenario. Problem.from_scenario is the canonical constructor; it injects the default harvest-system registry (when necessary) and synthesises single-shift calendars for legacy day-indexed inputs.

Parameters:

data (Any)

scenario

Back-reference to the source Scenario.

Type:

Scenario

days

List of integer day indices derived from scenario.num_days.

Type:

list[Day]

shifts

List of ShiftInstance entries representing every (day, shift_id) slot the solver should consider.

Type:

list[ShiftInstance]

Notes

Any code that builds Pyomo models or heuristic plans should accept a Problem rather than the raw Scenario to avoid recomputing shift/day metadata.

days: list[Day]
classmethod from_scenario(scenario)[source]
Parameters:

scenario (Scenario)

Return type:

Problem

model_config = {}

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

scenario: Scenario
shifts: list[ShiftInstance]
class fhops.scenario.ProductionRate(*, machine_id, block_id, rate)[source]

Bases: BaseModel

Production rate (m³ per assignment) for a machine/block pair.

Parameters:
machine_id

Machine identifier (must exist in Scenario.machines).

Type:

str

block_id

Block identifier (must exist in Scenario.blocks).

Type:

str

rate

Volume (m³) produced per full shift assignment (per day in single-shift scenarios), in the same units as Block.work_required. Must be non-negative.

Type:

float

block_id: str
machine_id: str
model_config = {}

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

rate: float
class fhops.scenario.RegistrationResult(scenario_path, bundle_id, bundle_version, manifest_sha256, manifest_path, aoi_bbox, warnings, written, pin)[source]

Bases: object

Outcome of a bundle registration.

Parameters:
aoi_bbox: tuple[float, float, float, float] | None
bundle_id: str
bundle_version: str
manifest_path: str
manifest_sha256: str
pin: LidarProductRef
scenario_path: Path
to_dict()[source]

Return a JSON-ready summary.

Return type:

dict[str, object]

warnings: tuple[str, ...]
written: bool
class fhops.scenario.Scenario(**data)[source]

Bases: BaseModel

Top-level container for the FHOPS data contract.

The model mirrors the CSV/YAML inputs documented in docs/howto/data_contract.rst and is the object returned by fhops.scenario.io.load_scenario(). Only validated, horizon-bounded data reaches this point, which means downstream solvers (MIP + heuristics) can rely on:

  • every block referencing a known landing/harvest system,

  • machine calendars/shift calendars never exceeding num_days,

  • mobilisation tables referencing existing blocks/machines, and

  • optional extras (crew assignments, road construction, GeoJSON metadata) being present only when fully specified.

Parameters:

data (Any)

name

Human-readable scenario label surfaced in CLI/Evaluation outputs.

Type:

str

num_days

Planning horizon length (integer number of days).

Type:

int

schema_version

Version of the input schema; used to guard loader compatibility.

Type:

str

start_date

Optional ISO date string used for timestamped exports.

Type:

date | None

blocks / machines / landings

Validated lists of the corresponding Pydantic models.

calendar / shift_calendar

Availability tables. shift_calendar may be None for day-level scenarios.

production_rates

Machine/block productivity table measured in m³ per shift assignment.

Type:

list[ProductionRate]

timeline

Optional TimelineConfig describing shifts, blackout windows, etc.

Type:

TimelineConfig | None

mobilisation

Optional MobilisationConfig describing distances and per-machine parameters.

Type:

MobilisationConfig | None

harvest_systems

Optional registry mapping harvest-system IDs to HarvestSystem definitions. Entries are overlaid on fhops.scheduling.systems.default_system_registry(); every Block.harvest_system_id must name a system of that combined registry (unknown ids are rejected since 1.0.1; 1.0.0 accepted them).

Type:

dict[str, HarvestSystem] | None

geo

Optional GeoMetadata with GeoJSON lookups.

Type:

GeoMetadata | None

crew_assignments

Optional list mapping crew IDs to machines for reporting/telemetry.

Type:

list[CrewAssignment] | None

locked_assignments

Optional list of ScheduleLock entries that pin machines to blocks on specific days (or specific day/shift slots when ScheduleLock.shift_id is set). Enforced by the operational MILP, the legacy MIP builder, and the heuristics. Validation rejects unknown machine/block ids, days outside the horizon or the block’s [earliest_start, latest_finish] window, blackout days, unknown shift labels (see shift_labels()), duplicate or mixed day/shift locks, and machines whose role is not a role of the block’s explicit harvest system. Files using shift_id require fhops>=1.0.1 (1.0.0 reads such entries as whole-day locks).

Type:

list[ScheduleLock] | None

objective_weights

Optional ObjectiveWeights overriding default solver weights.

Type:

ObjectiveWeights | None

road_construction

Optional list of RoadConstruction entries used by telemetry/costing exports.

Type:

list[RoadConstruction] | None

lidar_products

Optional pins to FEMIC-compiled LidarBC bundle versions. A pin records the bundle id, version, manifest checksum, licence, and optional manifest path. FHOPS consumes the FEMIC bundle; it never discovers or materializes source tiles itself.

Type:

list[LidarProductRef] | None

hydrology_products

Optional pins to FEMIC hydrology response bundles. A pin records the bundle id, version, manifest checksum, domain/ledger/response-surface ids, artifact checksums, readiness state, and caveats. FHOPS consumes the bundle; it does not copy or re-run FEMIC.

Type:

list[HydrologyBundleRef] | None

tree_inventory

Optional reference to a validated individual-tree inventory. The reference records the inventory and metadata checksums, CRS, review state, producer, method, tree count, block memberships, and caveats. The inventory is referenced, not copied.

Type:

TreeInventoryRef | None

lidar_support

Optional aggregated LidarBC evidence for blocks, landings, and roads. Each row states the FEMIC product, native resolution, aggregation, support footprint, and provenance; existing scenarios without LiDAR fields remain valid.

Type:

list[LidarSupport] | None

hauling

Optional HaulingConfig with log sorts, contractors, truck types and trucks, depots, sort yards, haul arcs, loader bindings, and the driver policy. None (default) means the scenario has no hauling layer. Files using this field require schema 1.1.0 (the loader still accepts 1.0.0).

Type:

HaulingConfig | None

initial_state

Optional ScenarioInitialState carrying staged inter-role inventory, remaining per-role output, head-start shift counts, and each machine’s last block at the start of the horizon. None (default) starts from the 1.0.0 initial conditions (empty staged inventory, full work_required per role, no prior machine position); it does not undo other 1.0.1 behaviour changes. Validation requires known block/machine ids, non-negative values, role keys drawn from the block’s harvest-system roles, role_remaining values not above the block’s work_required, and last_block_id values that are scenario blocks. Files using this field require fhops>=1.0.1 (1.0.0 ignores it).

Type:

ScenarioInitialState | None

Notes

The helper methods (machine_ids(), window_for(), etc.) are convenience routines for the solver/evaluation layers and are intentionally lightweight so they can be used in tight loops.

block_ids()[source]

Return the list of block identifiers defined in the scenario.

Return type:

list[str]

blocks: list[Block]
calendar: list[CalendarEntry]
crew_assignments: list[CrewAssignment] | None
geo: GeoMetadata | None
harvest_systems: dict[str, HarvestSystem] | None
hauling: HaulingConfig | None
hydrology_products: list[HydrologyBundleRef] | None
initial_state: ScenarioInitialState | None
landing_ids()[source]

Return the list of landing identifiers defined in the scenario.

Return type:

list[str]

landings: list[Landing]
lidar_products: list[LidarProductRef] | None
lidar_support: list[LidarSupport] | None
locked_assignments: list[ScheduleLock] | None
machine_ids()[source]

Return the list of machine identifiers defined in the scenario.

Return type:

list[str]

machines: list[Machine]
mobilisation: MobilisationConfig | None
model_config = {}

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

name: str
num_days: int
objective_weights: ObjectiveWeights | None
production_rates: list[ProductionRate]
road_construction: list[RoadConstruction] | None
schema_version: str
shift_calendar: list[ShiftCalendarEntry] | None
shift_labels()[source]

Return the shift labels of the scenario’s shift grid.

These are the labels accepted for ScheduleLock.shift_id and used for the shift_id column of solver and playback outputs. The source mirrors Problem.from_scenario().

Returns:

shift_calendar labels when a shift calendar is present (whether or not each entry is available), otherwise the timeline shift names, otherwise {"S1"} (the synthetic single shift used for day-indexed scenarios).

Return type:

set[str]

Examples

>>> from fhops.scenario.io import load_scenario
>>> load_scenario("examples/tiny7/scenario.yaml").shift_labels()
{'S1'}
start_date: date | None
timeline: TimelineConfig | None
tree_inventory: TreeInventoryRef | None
window_for(block_id)[source]

Return the inclusive (earliest, latest) day window for the target block.

Parameters:

block_id (str)

Return type:

tuple[int, int]

fhops.scenario.assess_block_operability(scenario, block_id, *, rules=None)[source]

Classify one block from LiDAR support, block fields, and explicit fallbacks.

Parameters:

  • scenario (Scenario)

  • block_id (str)

  • rules (OperabilityRules | None)

Return type:

BlockOperability

fhops.scenario.assess_femic_hydrology_readiness(bundle)[source]

Read optional validation-report and model-card evidence.

Parameters:

bundle (FemicHydrologyBundle) – Validated bundle.

Returns:

Review, tail-check, fit, and benchmark evidence. Missing evidence is reported explicitly and never treated as approval.

Return type:

FemicHydrologyReadiness

Raises:

FemicHydrologyError – If a packaged evidence artifact is malformed.

fhops.scenario.assess_scenario_operability(scenario, *, rules=None)[source]

Assess every block in a scenario.

Parameters:

  • scenario (Scenario)

  • rules (OperabilityRules | None)

Return type:

tuple[BlockOperability, …]

fhops.scenario.bundle_pin(bundle)[source]

Return the stable pin fields FHOPS records in its scenario provenance.

Parameters:

bundle (FemicLidarBundle)

Return type:

dict[str, object]

fhops.scenario.check_fhops_hydrology_constraints(scenario_path, mapping_path, schedule_path, hazard_path, *, assets_path=None, fail_on_closure=True, fail_on_warning=False, generated_utc=None)[source]

Check one schedule for operational hazard-window exposure.

Parameters:

  • scenario_path (str | Path) – FHOPS scenario YAML.

  • mapping_path (str | Path) – Hydrology mapping directory or manifest.

  • schedule_path (str | Path) – FHOPS assignments CSV.

  • hazard_path (str | Path) – Operational hazard JSON.

  • assets_path (str | Path | None, optional) – Explicit road/culvert asset JSON.

  • fail_on_closure (bool, optional) – Treat closure events as failures. Defaults to True.

  • fail_on_warning (bool, optional) – Treat warning events as failures. Defaults to False.

  • generated_utc (str | None, optional) – Check timestamp.

Returns:

Events, failure state, coverage, warnings, and caveats.

Return type:

HydrologyConstraintCheck

Raises:

FemicHydrologyConstraintError – If evidence is missing or a scheduled block has no mapping.

fhops.scenario.enforce_hydrology_readiness(readiness, *, policy='require_reviewed')[source]

Enforce a readiness policy for one-schedule screening.

Parameters:

  • readiness (FemicHydrologyReadiness) – Bundle readiness report.

  • policy (str, optional) – allow_provisional, require_reviewed, or require_ready.

Raises:

FemicHydrologyScreenError – If the policy is unknown or the evidence is insufficient.

Return type:

None

fhops.scenario.hydrology_bundle_pin(bundle)

Return a stable downstream pin for a validated bundle.

Parameters:

bundle (FemicHydrologyBundle) – Validated bundle.

Returns:

Bundle identity, manifest digest, IDs, artifact pins, caveats, and the producer consumer contract.

Return type:

dict[str, object]

fhops.scenario.load_femic_hydrology_bundle(bundle_dir, *, expected_bundle_id=None, expected_bundle_version=None, verify_checksums=True)[source]

Load and validate a FEMIC hydrology bundle.

Parameters:

  • bundle_dir (Path) – Bundle directory or manifest path.

  • expected_bundle_id (str | None, optional) – Require this bundle identifier.

  • expected_bundle_version (str | None, optional) – Require this bundle version.

  • verify_checksums (bool, optional) – Verify every declared size and SHA-256. Defaults to True.

Returns:

Validated bundle with artifact pins and summaries.

Return type:

FemicHydrologyBundle

Raises:

FemicHydrologyError – If the bundle is invalid.

fhops.scenario.load_femic_lidar_bundle(bundle_dir, *, expected_bundle_id=None, expected_bundle_version=None, verify_checksums=True)[source]

Load and validate a FEMIC LidarBC bundle for consumption.

Parameters:

  • bundle_dir (Path)

  • expected_bundle_id (str | None)

  • expected_bundle_version (str | None)

  • verify_checksums (bool)

Return type:

FemicLidarBundle

fhops.scenario.load_hydrology_assets(path)[source]

Load and validate an asset table.

Parameters:

path (Path) – Asset JSON path.

Returns:

Loaded assets and their validation report.

Return type:

tuple[HydrologyAssetMapping, HydrologyAssetValidation]

Raises:

FemicHydrologyAssetError – If the table is invalid.

fhops.scenario.load_hydrology_crosswalk(path, *, metadata_path=None)[source]

Load and validate an explicit hydrology crosswalk.

Parameters:

  • path (Path) – Crosswalk CSV with one support key per row.

  • metadata_path (Path | None, optional) – Crosswalk review metadata JSON.

Returns:

Checksummed rows and review metadata.

Return type:

HydrologyCrosswalk

Raises:

FemicHydrologyMappingError – If the CSV or metadata is malformed.

fhops.scenario.load_hydrology_exposure(path, *, schedule_path=None)[source]

Load and validate a shared exposure table.

Parameters:

  • path (Path) – Exposure JSON path.

  • schedule_path (Path | None, optional) – FHOPS assignments CSV to compare with schedule_sha256.

Returns:

Validated exposure table.

Return type:

HydrologyExposure

Raises:

FemicHydrologyExposureError – If the table is invalid or bound to a different schedule.

fhops.scenario.load_hydrology_hazard(path)[source]

Load and validate a hazard-window table.

Parameters:

path (Path) – Hazard JSON path.

Returns:

Validated hazard windows.

Return type:

HydrologyHazard

Raises:

FemicHydrologyHazardError – If the table is invalid.

fhops.scenario.load_hydrology_mapping(path)[source]

Load a mapping manifest written by write_hydrology_mapping().

Parameters:

path (Path) – Mapping directory or hydrology_mapping.json.

Returns:

Loaded mapping with rows, coverage, and review state.

Return type:

HydrologyMapping

Raises:

FemicHydrologyMappingError – If the manifest is missing, malformed, or inconsistent.

fhops.scenario.load_scenario(yaml_path)[source]

Load a Scenario from the YAML metadata + CSV bundle.

Parameters:

yaml_path (str | Path) – Path to the scenario.yaml file that references the component CSVs.

Returns:

Fully validated Pydantic model ready to be converted into a fhops.scenario.contract.Problem.

Return type:

Scenario

Notes

The loader performs several quality-of-life tasks that callers usually forget:

  • normalises optional string columns (e.g., harvest_system_id blanks → None),

  • back-fills mobilisation distance matrices from *_block_distances.csv whenever present,

  • accepts inline YAML overrides for optional tables (road construction, shift calendar, crew map),

  • re-roots GeoJSON paths relative to the scenario directory,

  • ensures every optional extra (timeline, mobilisation config, objective weights) is copied into the resulting Scenario instance,

  • parses an optional inline initial_state mapping (see fhops.scenario.contract.ScenarioInitialState) and validates it against the loaded blocks, machines, and harvest systems, and

  • re-validates the assembled scenario once all optional sections are attached, so YAML locked_assignments (machine/block/day/shift_id checks, duplicates, day+shift mixes, blackout clashes), mobilisation, crew assignments, and initial_state get the same fhops.scenario.contract.Scenario cross-validation as Python-constructed scenarios. Core tables (and the harvest_systems section, which block harvest_system_id values are checked against together with the default registry) are validated first, so their errors are reported before optional-section errors.

Examples

An inline initial_state section in scenario.yaml:

initial_state:
  blocks:
    - block_id: B01
      role_remaining: {feller_buncher: 0.0}
      staged_inventory: {feller_buncher: 150.0}
  machines:
    - machine_id: H3
      last_block_id: B02
Raises:

ValueError – (Pydantic ValidationError) when any table, the harvest_systems, locked_assignments or initial_state section, or the scenario cross-validation fails (including a block harvest_system_id that is neither in harvest_systems nor in fhops.scheduling.systems.default_system_registry()).

Parameters:

yaml_path (str | Path)

Return type:

Scenario

fhops.scenario.map_femic_hydrology_to_scenario(scenario_path, bundle_dir, crosswalk_path, *, metadata_path=None, allow_partial=False, generated_utc=None)[source]

Map explicit support rows to FEMIC watersheds and receptors.

Parameters:

  • scenario_path (str | Path) – FHOPS scenario YAML with a registered hydrology pin.

  • bundle_dir (str | Path) – FEMIC hydrology bundle directory or manifest.

  • crosswalk_path (str | Path) – Explicit crosswalk CSV.

  • metadata_path (str | Path | None, optional) – Crosswalk review metadata JSON.

  • allow_partial (bool, optional) – Record unmapped supports instead of failing. Defaults to False.

  • generated_utc (str | None, optional) – Mapping timestamp.

Returns:

Resolved rows, coverage, review state, warnings, and caveats.

Return type:

HydrologyMapping

Raises:

FemicHydrologyMappingError – If the bundle is not pinned, a support or FEMIC ID is unknown, a mapping conflicts, or coverage is incomplete without allow_partial.

fhops.scenario.map_femic_lidar_to_scenario(scenario_path, bundle_dir, *, products=None, statistic='mean', include_blocks=True, include_landings=True, roads_geojson=None, road_id_field='road_id', dry_run=False)[source]

Summarize FEMIC products over scenario support and write lidar_support rows.

Parameters:
Return type:

MappingResult

fhops.scenario.map_hydrology_assets(scenario_path, mapping_path, assets_path)[source]

Map explicit assets to scenario road jobs and FEMIC watersheds.

Parameters:

  • scenario_path (str | Path) – FHOPS scenario YAML.

  • mapping_path (str | Path) – Hydrology mapping directory or manifest.

  • assets_path (str | Path) – Asset JSON path.

Returns:

Resolved assets with coverage and caveats.

Return type:

HydrologyAssetMapping

Raises:

FemicHydrologyAssetError – If a road job or watershed is unknown or the mapping cannot load.

fhops.scenario.read_csv(path)[source]

Load a CSV file using pandas with UTF-8 defaults.

Parameters:

path (Path)

Return type:

DataFrame

fhops.scenario.read_femic_hydrology_domain(bundle)[source]

Return the pinned FEMIC domain artifact.

Parameters:

bundle (FemicHydrologyBundle) – Validated bundle.

Returns:

Domain payload with watersheds and receptors.

Return type:

dict[str, object]

Raises:

FemicHydrologyError – If the domain artifact is missing or malformed.

fhops.scenario.read_femic_hydrology_flood_risk(bundle)[source]

Return the pinned FEMIC flood-risk artifact.

Parameters:

bundle (FemicHydrologyBundle) – Validated bundle.

Returns:

Flood-risk payload with receptor rows.

Return type:

dict[str, object]

Raises:

FemicHydrologyError – If the artifact is missing or malformed.

fhops.scenario.read_femic_hydrology_ledger(bundle)[source]

Return the pinned FEMIC disturbance-ledger artifact.

Parameters:

bundle (FemicHydrologyBundle) – Validated bundle.

Returns:

Ledger payload with units and watershed assignments.

Return type:

dict[str, object]

Raises:

FemicHydrologyError – If the ledger artifact is missing or malformed.

fhops.scenario.read_femic_hydrology_response_surface(bundle)[source]

Return the pinned FEMIC response-surface artifact.

Parameters:

bundle (FemicHydrologyBundle) – Validated bundle.

Returns:

Response-surface payload with scenarios and rows.

Return type:

dict[str, object]

Raises:

FemicHydrologyError – If the artifact is missing or malformed.

fhops.scenario.register_femic_hydrology_bundle(scenario_path, bundle_dir, *, expected_bundle_id=None, expected_bundle_version=None, verify_checksums=True, dry_run=False)[source]

Validate a hydrology bundle and pin it in a scenario.

Parameters:

  • scenario_path (str | Path) – Scenario YAML to update.

  • bundle_dir (str | Path) – FEMIC hydrology bundle directory or manifest path.

  • expected_bundle_id (str | None, optional) – Require this bundle identifier.

  • expected_bundle_version (str | None, optional) – Require this bundle version.

  • verify_checksums (bool, optional) – Verify artifact sizes and SHA-256 digests. Defaults to True.

  • dry_run (bool, optional) – Validate and return the pin without writing the scenario.

Returns:

Stable pin, readiness state, warnings, and whether the scenario changed.

Return type:

RegistrationResult

Raises:

FemicHydrologyRegistrationError – If the bundle or scenario cannot be loaded, the existing pin conflicts, or the updated scenario does not reload.

fhops.scenario.register_femic_lidar_bundle(scenario_path, bundle_dir, *, expected_bundle_id=None, expected_bundle_version=None, verify_checksums=True, dry_run=False)[source]

Validate a FEMIC bundle and pin it in a scenario’s lidar_products section.

Parameters:

  • scenario_path (str | Path)

  • bundle_dir (str | Path)

  • expected_bundle_id (str | None)

  • expected_bundle_version (str | None)

  • verify_checksums (bool)

  • dry_run (bool)

Return type:

RegistrationResult

fhops.scenario.schedule_sha256(path)[source]

Return the SHA-256 digest of an assignments CSV.

Parameters:

path (Path) – FHOPS assignments CSV.

Returns:

Lowercase SHA-256 hex digest.

Return type:

str

Raises:

FemicHydrologyExposureError – If the file cannot be read.

fhops.scenario.screen_fhops_hydrology(scenario_path, bundle_dir, mapping_path, schedule_path, exposure_path, *, policy='require_reviewed', scenario_tolerance=0.05, allow_refused=False, generated_utc=None)[source]

Screen one FHOPS assignments schedule against a hydrology bundle.

Parameters:

  • scenario_path (str | Path) – FHOPS scenario YAML with a hydrology pin and block geometry.

  • bundle_dir (str | Path) – FEMIC hydrology bundle directory or manifest.

  • mapping_path (str | Path) – Mapping directory or manifest from scenario hydrology map.

  • schedule_path (str | Path) – FHOPS assignments CSV with block_id and day.

  • exposure_path (str | Path) – Schedule-bound exposure JSON.

  • policy (str, optional) – Readiness policy. Defaults to require_reviewed.

  • scenario_tolerance (float, optional) – Maximum feature difference for response-scenario selection.

  • allow_refused (bool, optional) – Return inside-domain rows plus refusals instead of failing.

  • generated_utc (str | None, optional) – Screening timestamp.

Returns:

Screening rows, refusals, coverage, warnings, and caveats.

Return type:

HydrologyScreen

Raises:

FemicHydrologyScreenError – If evidence is missing, the policy fails, the schedule is invalid, or a refusal occurs without allow_refused.

fhops.scenario.summarize_femic_lidar_for_polygons(bundle, polygons_path, *, product='slope', id_field=None, statistics=('mean', 'p95'))[source]

Summarize one bundle raster over polygon support.

Geometry and raster CRS must already match. No reprojection or resampling is performed.

Parameters:

  • bundle (FemicLidarBundle)

  • polygons_path (Path)

  • product (str)

  • id_field (str | None)

  • statistics (tuple[str, ...])

Return type:

tuple[LidarPolygonSummary, …]

fhops.scenario.validate_femic_hydrology_bundle(bundle_dir, *, expected_bundle_id=None, expected_bundle_version=None, verify_checksums=True)[source]

Validate a FEMIC hydrology response bundle without raising.

Parameters:

  • bundle_dir (Path) – Bundle directory or manifest path.

  • expected_bundle_id (str | None, optional) – Require this bundle identifier.

  • expected_bundle_version (str | None, optional) – Require this bundle version.

  • verify_checksums (bool, optional) – Verify every declared size and SHA-256. Defaults to True.

Returns:

Validation report with specific errors and warnings.

Return type:

FemicHydrologyValidation

fhops.scenario.validate_femic_lidar_bundle(bundle_dir, *, verify_checksums=True)[source]

Validate a FEMIC LidarBC bundle without raising on contract errors.

Parameters:

  • bundle_dir (Path)

  • verify_checksums (bool)

Return type:

FemicLidarValidation

fhops.scenario.validate_hydrology_assets(path)[source]

Validate an asset table without raising.

Parameters:

path (Path) – Asset JSON path.

Returns:

Validation report with specific errors and warnings.

Return type:

HydrologyAssetValidation

fhops.scenario.validate_hydrology_exposure(path, *, schedule_path=None)[source]

Validate a shared exposure table without raising.

Parameters:

  • path (Path) – Exposure JSON path.

  • schedule_path (Path | None, optional) – FHOPS assignments CSV to compare with the declared schedule digest.

Returns:

Validation report with specific errors and warnings.

Return type:

HydrologyExposureValidation

fhops.scenario.validate_hydrology_hazard(path)[source]

Validate a hazard-window table without raising.

Parameters:

path (Path) – Hazard JSON path.

Returns:

Validation report with specific errors and warnings.

Return type:

HydrologyHazardValidation

fhops.scenario.write_hydrology_constraint_check(result, out_dir)[source]

Write constraint-check events and manifest.

Parameters:

  • result (HydrologyConstraintCheck) – Check result.

  • out_dir (Path) – Output directory.

Returns:

Paths keyed by events and manifest.

Return type:

dict[str, Path]

fhops.scenario.write_hydrology_mapping(mapping, out_dir)[source]

Write the mapping CSV and manifest.

Parameters:

  • mapping (HydrologyMapping) – Resolved mapping.

  • out_dir (Path) – Output directory.

Returns:

Paths keyed by rows and manifest.

Return type:

dict[str, Path]

fhops.scenario.write_hydrology_screen(screen, out_dir)[source]

Write screening rows and manifest.

Parameters:

  • screen (HydrologyScreen) – Screening result.

  • out_dir (Path) – Output directory.

Returns:

Paths keyed by rows and manifest.

Return type:

dict[str, Path]

Scenario contract models (Pydantic schemas, validators).

class fhops.scenario.contract.Block(*, id, landing_id, work_required, earliest_start=1, latest_finish=None, harvest_system_id=None, avg_stem_size_m3=None, volume_per_ha_m3=None, volume_per_ha_m3_sigma=None, stem_density_per_ha=None, stem_density_per_ha_sigma=None, ground_slope_percent=None, salvage_processing_mode=None)[source]

Bases: BaseModel

Harvest block metadata and scheduling window.

Parameters:

  • id (str)

  • landing_id (str)

  • work_required (float)

  • earliest_start (int | None)

  • latest_finish (int | None)

  • harvest_system_id (str | None)

  • avg_stem_size_m3 (float | None)

  • volume_per_ha_m3 (float | None)

  • volume_per_ha_m3_sigma (float | None)

  • stem_density_per_ha (float | None)

  • stem_density_per_ha_sigma (float | None)

  • ground_slope_percent (float | None)

  • salvage_processing_mode (SalvageProcessingMode | None)

id

Unique block identifier (referenced by production rates and assignments).

Type:

str

landing_id

Landing where wood is forwarded; constrains landing daily capacity.

Type:

str

work_required

Volume (m³) that must be delivered from the block to complete it — the terminal (e.g., loader) output measured in the same units as ProductionRate.rate. Every role of an explicit harvest system processes up to this volume; loader truckload batching, playback production, KPIs (total_production, remaining_work_total) and all reference datasets treat it as m³. Must be non-negative.

Type:

float

earliest_start

Optional earliest day (inclusive, 1-indexed) when the block can begin.

Type:

Day | None

latest_finish

Optional latest day (inclusive) when the block must finish.

Type:

Day | None

harvest_system_id

Optional harvest system definition that restricts machine roles per block. Scenario validation requires the id to be a key of Scenario.harvest_systems or of fhops.scheduling.systems.default_system_registry().

Type:

str | None

avg_stem_size_m3 / volume_per_ha_m3 / volume_per_ha_m3_sigma

Stand descriptors (cubic metres) surfaced in analytics and productivity lookups.

stem_density_per_ha / stem_density_per_ha_sigma

Stems per hectare statistics used by some productivity models.

ground_slope_percent

Mean slope (%) for the block — used by productivity heuristics and diagnostics.

Type:

float | None

salvage_processing_mode

Enum describing downstream salvage processing (affects evaluation notes).

Type:

SalvageProcessingMode | None

avg_stem_size_m3: float | None
earliest_start: Day | None
ground_slope_percent: float | None
harvest_system_id: str | None
id: str
landing_id: str
latest_finish: Day | None
model_config = {}

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

salvage_processing_mode: SalvageProcessingMode | None
stem_density_per_ha: float | None
stem_density_per_ha_sigma: float | None
volume_per_ha_m3: float | None
volume_per_ha_m3_sigma: float | None
work_required: float
class fhops.scenario.contract.BlockInitialState(*, block_id, role_remaining=<factory>, staged_inventory=<factory>, role_shift_counts=<factory>)[source]

Bases: BaseModel

Carried-over sequencing state for one block at the start of the horizon.

Used to resume planning mid-operation (e.g. a rolling-horizon window that starts after some volume has already been felled, extracted, or processed). The terminal volume still to be delivered is expressed through Block.work_required itself; this model captures the intermediate (non-terminal) state of the harvest-system roles.

Parameters:
block_id

Identifier of a block in Scenario.blocks.

Type:

str

role_remaining

Mapping role -> volume (m³) the role may still output on this block, i.e. work_required at the original start minus the role’s cumulative output. Roles omitted here default to the block’s work_required (the v1.0.0 initialisation). Values must be non-negative.

Type:

dict[str, float]

staged_inventory

Mapping role -> volume (m³) output by that role on this block but not yet consumed by its downstream role(s) (for example, felled-but-unskidded wood is keyed by the felling role). Staged volume is kept per upstream role (since 1.0.1): a downstream role consumes its output from the staged volume of every one of its upstream roles, so it can process at most the smallest of them, and the downstream roles of an upstream role that feeds several roles share (split) its staged volume. Roles omitted here start with zero staged volume. Values must be non-negative.

Type:

dict[str, float]

role_shift_counts

Mapping role -> shifts already worked on this block. Informational: it seeds fhops.evaluation.sequencing.SequencingTracker.role_counts_total (reported by the rolling-horizon carry-forward) but no longer affects sequencing checks, because head-start buffers (role_headstart_shifts) are enforced as staged upstream volume (staged_inventory) in the MILP, the heuristics, and playback alike (#109). Roles omitted here start at zero. Values must be non-negative integers.

Type:

dict[str, int]

Notes

Role keys are normalised with fhops.costing.machine_rates.normalize_machine_role() (case-insensitive, punctuation → _), so "Feller-Buncher" becomes feller_buncher. Scenario-level validation requires every role key to be a role of the block’s harvest system, which means the block must declare harvest_system_id when any role-keyed state is given.

block_id: str
model_config = {}

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

role_remaining: dict[str, float]
role_shift_counts: dict[str, int]
staged_inventory: dict[str, float]
class fhops.scenario.contract.CalendarEntry(*, machine_id, day, available=1)[source]

Bases: BaseModel

Day-level availability for a machine.

Parameters:
machine_id

Identifier of the machine whose availability is being set.

Type:

str

day

One-indexed day number relative to the scenario horizon.

Type:

Day

available

Binary flag (1 available, 0 unavailable) controlling day-level assignment eligibility.

Type:

int

available: int
day: Day
machine_id: str
model_config = {}

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

class fhops.scenario.contract.CrewAssignment(*, crew_id, machine_id, primary_role=None, notes=None)[source]

Bases: BaseModel

Optional mapping of crews to machines/roles.

Parameters:

  • crew_id (str)

  • machine_id (str)

  • primary_role (str | None)

  • notes (str | None)

crew_id

Unique crew identifier.

Type:

str

machine_id

Machine assigned to the crew.

Type:

str

primary_role

Optional role label associated with the crew (e.g., fallers, processors).

Type:

str | None

notes

Additional metadata surfaced in telemetry exports.

Type:

str | None

crew_id: str
machine_id: str
model_config = {}

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

notes: str | None
primary_role: str | None
fhops.scenario.contract.Day

alias of int

class fhops.scenario.contract.HydrologyArtifactPin(*, role, relative_path, schema_version, media_type, size_bytes, sha256)[source]

Bases: BaseModel

One artifact recorded in a hydrology bundle pin.

Parameters:
media_type: str
model_config = {}

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

relative_path: str
role: str
schema_version: str
sha256: str
size_bytes: int
class fhops.scenario.contract.HydrologyBundleRef(*, bundle_id, bundle_version, manifest_sha256, domain_id, ledger_id, response_surface_id, manifest_path=None, readiness_status='not_available', readiness_evidence_sha256=None, artifacts=<factory>, caveats=<factory>, consumer_contract=<factory>)[source]

Bases: BaseModel

A scenario-level pin to one FEMIC hydrology bundle version.

The fields mirror fhops.scenario.hydrology.bundle_pin(). The pin is a reference, not a copy of the bundle.

Parameters:

  • bundle_id (Annotated[str, MinLen(min_length=1)])

  • bundle_version (Annotated[str, MinLen(min_length=1)])

  • manifest_sha256 (str)

  • domain_id (Annotated[str, MinLen(min_length=1)])

  • ledger_id (Annotated[str, MinLen(min_length=1)])

  • response_surface_id (Annotated[str, MinLen(min_length=1)])

  • manifest_path (str | None)

  • readiness_status (Literal['not_available', 'provisional', 'conditionally_ready', 'ready', 'blocked'])

  • readiness_evidence_sha256 (str | None)

  • artifacts (list[HydrologyArtifactPin])

  • caveats (list[str])

  • consumer_contract (dict[str, object])

artifacts: list[HydrologyArtifactPin]
bundle_id: str
bundle_version: str
caveats: list[str]
consumer_contract: dict[str, object]
domain_id: str
ledger_id: str
manifest_path: str | None
manifest_sha256: str
model_config = {}

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

readiness_evidence_sha256: str | None
readiness_status: HydrologyReadinessStatus
response_surface_id: str
class fhops.scenario.contract.Landing(*, id, daily_capacity=2)[source]

Bases: BaseModel

Landing metadata including its machine capacity.

Parameters:

  • id (str)

  • daily_capacity (int)

id

Landing identifier referenced by blocks and mobilisation logic.

Type:

str

daily_capacity

Maximum number of machines that can work the landing’s blocks concurrently, counted per shift slot (day, shift_id) (on single-shift scenarios: per day). The name is kept for compatibility. The heuristics and the operational MILP apply the same rule: when ObjectiveWeights.landing_surplus is 0 (default) the capacity is hard (heuristics: the repair keeps landings within capacity and any remaining extra machine costs the hard violation penalty, at least 1000, #140; MILP: constraint), otherwise each slot’s k-th machine beyond capacity costs k × landing_surplus. Before FHOPS 1.0.1 the operational MILP counted machine-shifts per day with a slack that was free at weight 0 (#125).

Type:

int

daily_capacity: int
id: str
model_config = {}

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

class fhops.scenario.contract.LidarProductPin(*, product, relative_path, sha256, crs, resolution_m=None, vertical_datum, source_tile_ids=<factory>)[source]

Bases: BaseModel

One product file recorded in a scenario bundle pin.

Parameters:
crs: str
model_config = {}

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

product: str
relative_path: str
resolution_m: float | None
sha256: str
source_tile_ids: list[str]
vertical_datum: str
class fhops.scenario.contract.LidarProductRef(*, bundle_id, bundle_version, manifest_sha256=None, licence_id=None, manifest_path=None, crs=None, aoi_bbox=None, selection=None, products=None)[source]

Bases: BaseModel

A scenario-level pin to one FEMIC bundle version.

The fields mirror fhops.scenario.lidar.bundle_pin(); manifest_sha256 is the digest of the bundle manifest, not of an individual raster.

Parameters:
aoi_bbox: tuple[float, float, float, float] | None
bundle_id: str
bundle_version: str
crs: str | None
licence_id: str | None
manifest_path: str | None
manifest_sha256: str | None
model_config = {}

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

products: list[LidarProductPin] | None
selection: dict[str, object] | None
class fhops.scenario.contract.LidarSupport(*, support_unit, support_id, bundle_id, bundle_version, product, aggregation='area_weighted_mean', native_resolution_m, support_area_m2=None, valid_pixels=None, slope_percent=None, aspect_degrees=None, roughness_m=None, canopy_cover_percent=None, canopy_height_m=None, vertical_structure_index=None, vertical_structure_class=None, provenance=None)[source]

Bases: BaseModel

Aggregated LidarBC evidence for one block, landing, or road.

The values are the output of a FEMIC raster summary (for example fhops.scenario.lidar.summarize_femic_lidar_for_polygons()). Native resolution, product name, aggregation, valid-pixel count, and support footprint preserve how the summary was made.

Parameters:
aggregation: AggregationMethod
aspect_degrees: float | None
bundle_id: str
bundle_version: str
canopy_cover_percent: float | None
canopy_height_m: float | None
model_config = {}

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

native_resolution_m: float
product: str
provenance: str | None
roughness_m: float | None
slope_percent: float | None
support_area_m2: float | None
support_id: str
support_unit: LidarSupportUnit
valid_pixels: int | None
vertical_structure_class: str | None
vertical_structure_index: float | None
class fhops.scenario.contract.Machine(*, id, crew=None, daily_hours=24.0, operating_cost=0.0, role=None, repair_usage_hours=None)[source]

Bases: BaseModel

Machine definition (identifier, crew, availability, and costing metadata).

Parameters:

  • id (str)

  • crew (str | None)

  • daily_hours (float)

  • operating_cost (float)

  • role (str | None)

  • repair_usage_hours (int | None)

id

Unique machine identifier referenced throughout calendars/assignments.

Type:

str

crew

Optional crew label for reporting/telemetry grouping.

Type:

str | None

daily_hours

Maximum hours the machine can operate per day (defaults to 24).

Type:

float

operating_cost

Cost per scheduled machine hour (SMH) expressed in scenario currency units.

Type:

float

role

Optional machine role string (normalised via normalize_machine_role) used by harvest systems and rental-rate lookups.

Type:

str | None

repair_usage_hours

Optional cumulative repair hours that influences the rental-rate defaults.

Type:

int | None

crew: str | None
daily_hours: float
id: str
model_config = {}

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

operating_cost: float
repair_usage_hours: int | None
role: str | None
class fhops.scenario.contract.MachineInitialState(*, machine_id, last_block_id=None)[source]

Bases: BaseModel

Carried-over position of one machine at the start of the horizon.

Parameters:

  • machine_id (str)

  • last_block_id (str | None)

machine_id

Identifier of a machine in Scenario.machines.

Type:

str

last_block_id

Block the machine occupied in its last worked slot before the horizon, or None when unknown (the v1.0.0 behaviour: the first move is free). When set, the machine’s first assignment to a different block is charged mobilisation (MobilisationConfig) and a transition by the operational MILP, the heuristics, and playback.

Type:

str | None

last_block_id: str | None
machine_id: str
model_config = {}

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

class fhops.scenario.contract.Problem(*, scenario, days, shifts)[source]

Bases: BaseModel

Runtime representation of a scenario used by solvers.

Problem wraps a validated Scenario and expands it into concrete days and shifts so optimisation code can iterate over deterministic index sets without repeatedly querying the Scenario. Problem.from_scenario is the canonical constructor; it injects the default harvest-system registry (when necessary) and synthesises single-shift calendars for legacy day-indexed inputs.

Parameters:

  • scenario (Scenario)

  • days (list[int])

  • shifts (list[ShiftInstance])

scenario

Back-reference to the source Scenario.

Type:

Scenario

days

List of integer day indices derived from scenario.num_days.

Type:

list[Day]

shifts

List of ShiftInstance entries representing every (day, shift_id) slot the solver should consider.

Type:

list[ShiftInstance]

Notes

Any code that builds Pyomo models or heuristic plans should accept a Problem rather than the raw Scenario to avoid recomputing shift/day metadata.

days: list[Day]
classmethod from_scenario(scenario)[source]
Parameters:

scenario (Scenario)

Return type:

Problem

model_config = {}

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

scenario: Scenario
shifts: list[ShiftInstance]
class fhops.scenario.contract.ProductionRate(*, machine_id, block_id, rate)[source]

Bases: BaseModel

Production rate (m³ per assignment) for a machine/block pair.

Parameters:
machine_id

Machine identifier (must exist in Scenario.machines).

Type:

str

block_id

Block identifier (must exist in Scenario.blocks).

Type:

str

rate

Volume (m³) produced per full shift assignment (per day in single-shift scenarios), in the same units as Block.work_required. Must be non-negative.

Type:

float

block_id: str
machine_id: str
model_config = {}

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

rate: float
class fhops.scenario.contract.RoadConstruction(*, id, machine_slug, road_length_m, include_mobilisation=True, soil_profile_ids=None, notes=None)[source]

Bases: BaseModel

Road/subgrade construction job describing TR-28 soil profiles and costing metadata.

Parameters:

  • id (str)

  • machine_slug (str)

  • road_length_m (float)

  • include_mobilisation (bool)

  • soil_profile_ids (list[str] | None)

  • notes (str | None)

id

Unique job identifier referenced in telemetry and costing exports.

Type:

str

machine_slug

Machine rate slug (tr28 index) used to determine construction costs.

Type:

str

road_length_m

Length of the road section (metres) to construct.

Type:

float

include_mobilisation

When True, mobilisation costs are included in the estimate.

Type:

bool

soil_profile_ids

Optional list of TR-28 soil profile identifiers associated with the job.

Type:

list[str] | None

notes

Free-form comments surfaced in CLI summaries.

Type:

str | None

id: str
include_mobilisation: bool
machine_slug: str
model_config = {}

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

notes: str | None
road_length_m: float
soil_profile_ids: list[str] | None
class fhops.scenario.contract.SalvageProcessingMode(*values)[source]

Bases: StrEnum

IN_WOODS_CHIPPING = 'in_woods_chipping'
PORTABLE_MILL = 'portable_mill'
STANDARD_MILL = 'standard_mill'
class fhops.scenario.contract.Scenario(*, name, num_days, schema_version='1.0.0', start_date=None, blocks, machines, landings, calendar, shift_calendar=None, production_rates, timeline=None, mobilisation=None, harvest_systems=None, geo=None, crew_assignments=None, locked_assignments=None, objective_weights=None, road_construction=None, initial_state=None, hauling=None, lidar_products=None, lidar_support=None, hydrology_products=None, tree_inventory=None)[source]

Bases: BaseModel

Top-level container for the FHOPS data contract.

The model mirrors the CSV/YAML inputs documented in docs/howto/data_contract.rst and is the object returned by fhops.scenario.io.load_scenario(). Only validated, horizon-bounded data reaches this point, which means downstream solvers (MIP + heuristics) can rely on:

  • every block referencing a known landing/harvest system,

  • machine calendars/shift calendars never exceeding num_days,

  • mobilisation tables referencing existing blocks/machines, and

  • optional extras (crew assignments, road construction, GeoJSON metadata) being present only when fully specified.

Parameters:

  • name (str)

  • num_days (int)

  • schema_version (str)

  • start_date (date | None)

  • blocks (list[Block])

  • machines (list[Machine])

  • landings (list[Landing])

  • calendar (list[CalendarEntry])

  • shift_calendar (list[ShiftCalendarEntry] | None)

  • production_rates (list[ProductionRate])

  • timeline (TimelineConfig | None)

  • mobilisation (MobilisationConfig | None)

  • harvest_systems (dict[str, HarvestSystem] | None)

  • geo (GeoMetadata | None)

  • crew_assignments (list[CrewAssignment] | None)

  • locked_assignments (list[ScheduleLock] | None)

  • objective_weights (ObjectiveWeights | None)

  • road_construction (list[RoadConstruction] | None)

  • initial_state (ScenarioInitialState | None)

  • hauling (HaulingConfig | None)

  • lidar_products (list[LidarProductRef] | None)

  • lidar_support (list[LidarSupport] | None)

  • hydrology_products (list[HydrologyBundleRef] | None)

  • tree_inventory (TreeInventoryRef | None)

name

Human-readable scenario label surfaced in CLI/Evaluation outputs.

Type:

str

num_days

Planning horizon length (integer number of days).

Type:

int

schema_version

Version of the input schema; used to guard loader compatibility.

Type:

str

start_date

Optional ISO date string used for timestamped exports.

Type:

date | None

blocks / machines / landings

Validated lists of the corresponding Pydantic models.

calendar / shift_calendar

Availability tables. shift_calendar may be None for day-level scenarios.

production_rates

Machine/block productivity table measured in m³ per shift assignment.

Type:

list[ProductionRate]

timeline

Optional TimelineConfig describing shifts, blackout windows, etc.

Type:

TimelineConfig | None

mobilisation

Optional MobilisationConfig describing distances and per-machine parameters.

Type:

MobilisationConfig | None

harvest_systems

Optional registry mapping harvest-system IDs to HarvestSystem definitions. Entries are overlaid on fhops.scheduling.systems.default_system_registry(); every Block.harvest_system_id must name a system of that combined registry (unknown ids are rejected since 1.0.1; 1.0.0 accepted them).

Type:

dict[str, HarvestSystem] | None

geo

Optional GeoMetadata with GeoJSON lookups.

Type:

GeoMetadata | None

crew_assignments

Optional list mapping crew IDs to machines for reporting/telemetry.

Type:

list[CrewAssignment] | None

locked_assignments

Optional list of ScheduleLock entries that pin machines to blocks on specific days (or specific day/shift slots when ScheduleLock.shift_id is set). Enforced by the operational MILP, the legacy MIP builder, and the heuristics. Validation rejects unknown machine/block ids, days outside the horizon or the block’s [earliest_start, latest_finish] window, blackout days, unknown shift labels (see shift_labels()), duplicate or mixed day/shift locks, and machines whose role is not a role of the block’s explicit harvest system. Files using shift_id require fhops>=1.0.1 (1.0.0 reads such entries as whole-day locks).

Type:

list[ScheduleLock] | None

objective_weights

Optional ObjectiveWeights overriding default solver weights.

Type:

ObjectiveWeights | None

road_construction

Optional list of RoadConstruction entries used by telemetry/costing exports.

Type:

list[RoadConstruction] | None

lidar_products

Optional pins to FEMIC-compiled LidarBC bundle versions. A pin records the bundle id, version, manifest checksum, licence, and optional manifest path. FHOPS consumes the FEMIC bundle; it never discovers or materializes source tiles itself.

Type:

list[LidarProductRef] | None

hydrology_products

Optional pins to FEMIC hydrology response bundles. A pin records the bundle id, version, manifest checksum, domain/ledger/response-surface ids, artifact checksums, readiness state, and caveats. FHOPS consumes the bundle; it does not copy or re-run FEMIC.

Type:

list[HydrologyBundleRef] | None

tree_inventory

Optional reference to a validated individual-tree inventory. The reference records the inventory and metadata checksums, CRS, review state, producer, method, tree count, block memberships, and caveats. The inventory is referenced, not copied.

Type:

TreeInventoryRef | None

lidar_support

Optional aggregated LidarBC evidence for blocks, landings, and roads. Each row states the FEMIC product, native resolution, aggregation, support footprint, and provenance; existing scenarios without LiDAR fields remain valid.

Type:

list[LidarSupport] | None

hauling

Optional HaulingConfig with log sorts, contractors, truck types and trucks, depots, sort yards, haul arcs, loader bindings, and the driver policy. None (default) means the scenario has no hauling layer. Files using this field require schema 1.1.0 (the loader still accepts 1.0.0).

Type:

HaulingConfig | None

initial_state

Optional ScenarioInitialState carrying staged inter-role inventory, remaining per-role output, head-start shift counts, and each machine’s last block at the start of the horizon. None (default) starts from the 1.0.0 initial conditions (empty staged inventory, full work_required per role, no prior machine position); it does not undo other 1.0.1 behaviour changes. Validation requires known block/machine ids, non-negative values, role keys drawn from the block’s harvest-system roles, role_remaining values not above the block’s work_required, and last_block_id values that are scenario blocks. Files using this field require fhops>=1.0.1 (1.0.0 ignores it).

Type:

ScenarioInitialState | None

Notes

The helper methods (machine_ids(), window_for(), etc.) are convenience routines for the solver/evaluation layers and are intentionally lightweight so they can be used in tight loops.

block_ids()[source]

Return the list of block identifiers defined in the scenario.

Return type:

list[str]

blocks: list[Block]
calendar: list[CalendarEntry]
crew_assignments: list[CrewAssignment] | None
geo: GeoMetadata | None
harvest_systems: dict[str, HarvestSystem] | None
hauling: HaulingConfig | None
hydrology_products: list[HydrologyBundleRef] | None
initial_state: ScenarioInitialState | None
landing_ids()[source]

Return the list of landing identifiers defined in the scenario.

Return type:

list[str]

landings: list[Landing]
lidar_products: list[LidarProductRef] | None
lidar_support: list[LidarSupport] | None
locked_assignments: list[ScheduleLock] | None
machine_ids()[source]

Return the list of machine identifiers defined in the scenario.

Return type:

list[str]

machines: list[Machine]
mobilisation: MobilisationConfig | None
model_config = {}

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

name: str
num_days: int
objective_weights: ObjectiveWeights | None
production_rates: list[ProductionRate]
road_construction: list[RoadConstruction] | None
schema_version: str
shift_calendar: list[ShiftCalendarEntry] | None
shift_labels()[source]

Return the shift labels of the scenario’s shift grid.

These are the labels accepted for ScheduleLock.shift_id and used for the shift_id column of solver and playback outputs. The source mirrors Problem.from_scenario().

Returns:

shift_calendar labels when a shift calendar is present (whether or not each entry is available), otherwise the timeline shift names, otherwise {"S1"} (the synthetic single shift used for day-indexed scenarios).

Return type:

set[str]

Examples

>>> from fhops.scenario.io import load_scenario
>>> load_scenario("examples/tiny7/scenario.yaml").shift_labels()
{'S1'}
start_date: date | None
timeline: TimelineConfig | None
tree_inventory: TreeInventoryRef | None
window_for(block_id)[source]

Return the inclusive (earliest, latest) day window for the target block.

Parameters:

block_id (str)

Return type:

tuple[int, int]

class fhops.scenario.contract.ScenarioInitialState(*, blocks=<factory>, machines=<factory>)[source]

Bases: BaseModel

Optional initial state for a scenario (resume planning mid-operation).

When Scenario.initial_state is None (default) the horizon starts from the FHOPS 1.0.0 initial conditions: staged inventories and role shift counts start at zero, each role may output the full work_required, and machines have no prior position (their first move is free). Solver and playback results can still differ from 1.0.0 because of other 1.0.1 fixes (MILP/playback sequencing alignment, MILP blackouts, stochastic playback events; see docs/releases/v1.0.1.md).

Scenario files that contain initial_state require fhops>=1.0.1: FHOPS 1.0.0 silently ignores the section (schema_version is unchanged at 1.0.0).

Parameters:

  • blocks (list[BlockInitialState])

  • machines (list[MachineInitialState])

blocks

Per-block BlockInitialState entries (at most one per block_id). Blocks not listed start from the default (empty) state.

Type:

list[BlockInitialState]

machines

Per-machine MachineInitialState entries (at most one per machine_id).

Type:

list[MachineInitialState]

Examples

>>> state = ScenarioInitialState(
...     blocks=[
...         BlockInitialState(
...             block_id="B01",
...             role_remaining={"feller_buncher": 0.0},
...             staged_inventory={"feller_buncher": 120.0},
...         )
...     ],
...     machines=[MachineInitialState(machine_id="H3", last_block_id="B02")],
... )
>>> state.block_state("B01").staged_inventory
{'feller_buncher': 120.0}
block_state(block_id)[source]

Return the BlockInitialState for block_id (None when absent).

Parameters:

block_id (str)

Return type:

BlockInitialState | None

blocks: list[BlockInitialState]
last_block_by_machine()[source]

Return machine_id -> last_block_id for machines with a known prior position.

Return type:

dict[str, str]

machines: list[MachineInitialState]
model_config = {}

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

class fhops.scenario.contract.ScheduleLock(**data)[source]

Bases: BaseModel

Immutable assignment of a machine to a block on a specific day (or day/shift slot).

Parameters:

data (Any)

machine_id

Identifier of the machine being locked (must exist in Scenario.machines).

Type:

str

block_id

Identifier of the block that must be worked during the lock.

Type:

str

day

One-indexed day within the planning horizon where the lock applies.

Type:

Day

shift_id

Optional shift label. None (default) locks every available shift of day to block_id (the v1.0.0 behaviour). When set, only the (day, shift_id) slot is locked; the label must belong to the scenario’s shift grid (shift_calendar labels, timeline shift names, or S1 for day-indexed scenarios).

Type:

str | None

production

Optional planned production (m³, >= 0) of the locked slot. It is used when a lock table is replayed (rolling-horizon carry-forward and fhops.planning.compute_rolling_kpis()), so a stitched operational-MILP plan is evaluated with the production the MILP planned rather than the full production rate. None (default) replays with min(rate, remaining). The heuristics (SA, ILS, Tabu) and playback of plans without a production column use it as an upper bound on the locked slot’s production (0 = idle; since #158); a locked slot never produces more than its staged input allows and idles without a violation when it has none. The operational MILP ignores it.

Type:

float | None

Notes

A day-level lock and a shift-level lock for the same machine/day are rejected by Scenario validation, as are duplicate (machine_id, day, shift_id) entries.

block_id: str
day: Day
machine_id: str
model_config = {}

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

production: float | None
shift_id: str | None
class fhops.scenario.contract.TreeInventoryRef(*, inventory_id, inventory_sha256, metadata_sha256, inventory_path, metadata_path, crs, review_status='not_reviewed', reviewed_by=None, source_id, producer, method, tree_count, block_ids=<factory>, caveats=<factory>)[source]

Bases: BaseModel

A scenario-level reference to one validated tree inventory.

The inventory is referenced, not copied. inventory_sha256 and metadata_sha256 pin the CSV and metadata bytes; the remaining fields preserve identity, CRS, review, and producer provenance.

Parameters:
inventory_id

Stable inventory identifier from the metadata.

Type:

str

inventory_sha256

SHA-256 digest of the inventory CSV.

Type:

str

metadata_sha256

SHA-256 digest of the metadata JSON.

Type:

str

inventory_path

Inventory CSV path, relative to the scenario when possible.

Type:

str

metadata_path

Metadata JSON path, relative to the scenario when possible.

Type:

str

crs

Inventory coordinate reference system.

Type:

str

review_status

not_reviewed, review_pending, reviewed, or reviewed_with_conditions.

Type:

str

reviewed_by

Reviewer identifier when reviewed.

Type:

str | None

source_id

Producer source identifier.

Type:

str

producer

Producer name.

Type:

str

method

Detection/segmentation method.

Type:

str

tree_count

Number of tree records.

Type:

int

block_ids

Declared block memberships.

Type:

list[str]

caveats

Inventory caveats.

Type:

list[str]

block_ids: list[str]
caveats: list[str]
crs: str
inventory_id: str
inventory_path: str
inventory_sha256: str
metadata_path: str
metadata_sha256: str
method: str
model_config = {}

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

producer: str
review_status: TreeInventoryReviewStatus
reviewed_by: str | None
source_id: str
tree_count: int
fhops.scenario.contract.validate_initial_state(scenario)[source]

Validate scenario.initial_state against the scenario’s blocks, machines, and systems.

Called automatically by Scenario validation (including the final re-validation in fhops.scenario.io.load_scenario()). Call it directly after attaching an initial_state with model_copy(update=...), which bypasses Pydantic validators.

Parameters:

scenario (Scenario) – Scenario whose initial_state should be checked. None initial state is accepted.

Raises:

ValueError – If a block/machine id is unknown, a last_block_id is not a scenario block, a block with role-keyed state has no harvest_system_id, a role key is not a role of the block’s harvest system, or a terminal role_remaining value exceeds the block’s work_required by more than ROLE_REMAINING_TOLERANCE (1e-6 m³). Non-terminal roles may exceed the terminal remainder in multi-terminal systems or when staged inventory is carried (#157/N3). Harvest systems are looked up in scenario.harvest_systems (falling back to fhops.scheduling.systems.default_system_registry(), mirroring Problem.from_scenario()).

Return type:

None

Quickstart

from fhops.scenario.io import load_scenario
from fhops.scenario.contract import Problem

scenario = load_scenario("examples/tiny7/scenario.yaml")
problem = Problem.from_scenario(scenario)
print(problem.days, len(problem.shifts))

Key models:

  • fhops.scenario.contract.Scenario – Pydantic model for inputs (blocks, machines, timelines).

  • fhops.scenario.contract.Problem – Derived object used by solvers (days, shifts, scenario).

  • fhops.scenario.contract.MobilisationConfig / TimelineConfig – optional extras for mobilisation/shift data.

  • fhops.scenario.contract.ScenarioInitialState – optional carried-in state (staged inventory, remaining role output, head-start shift counts, last block per machine); see Initial State (Resuming Mid-Operation).

  • fhops.scenario.contract.ScheduleLock – machine/block/day locks with an optional shift_id.

  • fhops.hauling.contract.HaulingConfig – optional log-hauling extension (schema 1.1.0); see Log Hauling.

Scenario IO helpers (YAML/CSV loaders, schema checks).

fhops.scenario.io.load_scenario(yaml_path)[source]

Load a Scenario from the YAML metadata + CSV bundle.

Parameters:

yaml_path (str | Path) – Path to the scenario.yaml file that references the component CSVs.

Returns:

Fully validated Pydantic model ready to be converted into a fhops.scenario.contract.Problem.

Return type:

Scenario

Notes

The loader performs several quality-of-life tasks that callers usually forget:

  • normalises optional string columns (e.g., harvest_system_id blanks → None),

  • back-fills mobilisation distance matrices from *_block_distances.csv whenever present,

  • accepts inline YAML overrides for optional tables (road construction, shift calendar, crew map),

  • re-roots GeoJSON paths relative to the scenario directory,

  • ensures every optional extra (timeline, mobilisation config, objective weights) is copied into the resulting Scenario instance,

  • parses an optional inline initial_state mapping (see fhops.scenario.contract.ScenarioInitialState) and validates it against the loaded blocks, machines, and harvest systems, and

  • re-validates the assembled scenario once all optional sections are attached, so YAML locked_assignments (machine/block/day/shift_id checks, duplicates, day+shift mixes, blackout clashes), mobilisation, crew assignments, and initial_state get the same fhops.scenario.contract.Scenario cross-validation as Python-constructed scenarios. Core tables (and the harvest_systems section, which block harvest_system_id values are checked against together with the default registry) are validated first, so their errors are reported before optional-section errors.

Examples

An inline initial_state section in scenario.yaml:

initial_state:
  blocks:
    - block_id: B01
      role_remaining: {feller_buncher: 0.0}
      staged_inventory: {feller_buncher: 150.0}
  machines:
    - machine_id: H3
      last_block_id: B02
Raises:

ValueError – (Pydantic ValidationError) when any table, the harvest_systems, locked_assignments or initial_state section, or the scenario cross-validation fails (including a block harvest_system_id that is neither in harvest_systems nor in fhops.scheduling.systems.default_system_registry()).

Parameters:

yaml_path (str | Path)

Return type:

Scenario

fhops.scenario.io.read_csv(path)[source]

Load a CSV file using pandas with UTF-8 defaults.

Parameters:

path (Path)

Return type:

DataFrame

Synthetic scenario generators.

class fhops.scenario.synthetic.BlackoutBias(start_day, end_day, probability, duration=None)[source]

Bases: object

Bias blackout probabilities for specific windows.

Parameters:
duration: tuple[int, int] | int | None = None
end_day: int
probability: float
start_day: int
class fhops.scenario.synthetic.SyntheticDatasetBundle(scenario, blocks, machines, landings, calendar, production_rates, metadata=None)[source]

Bases: object

Container for generated scenario tables and helpers to persist them.

Parameters:

  • scenario (Scenario)

  • blocks (DataFrame)

  • machines (DataFrame)

  • landings (DataFrame)

  • calendar (DataFrame)

  • production_rates (DataFrame)

  • metadata (dict[str, object] | None)

blocks: DataFrame
calendar: DataFrame
landings: DataFrame
machines: DataFrame
metadata: dict[str, object] | None = None
production_rates: DataFrame
scenario: Scenario
write(out_dir, *, include_yaml=True, metadata_path=None)[source]
Parameters:

  • out_dir (Path)

  • include_yaml (bool)

  • metadata_path (Path | None)

Return type:

Path

class fhops.scenario.synthetic.SyntheticDatasetConfig(name, num_blocks, num_days, num_machines, num_landings=1, shift_hours=(8.0, 12.0), shifts_per_day=1, machine_daily_hours=24.0, landing_capacity=(1, 3), work_required=(6.0, 18.0), production_rate=(6.0, 18.0), availability_probability=0.9, blackout_probability=0.1, blackout_duration=(1, 2), role_pool=<factory>, tier=None, terrain_pool=<factory>, terrain_weights=None, prescription_pool=<factory>, prescription_weights=None, crew_pool=<factory>, capability_pool=<factory>, crew_capability_span=(1, 2), system_mix=None, blackout_biases=<factory>, sampling_overrides=None, block_metric_section='daily')[source]

Bases: object

Configuration for generating random synthetic datasets.

Notes

work_required and production_rate are sampling ranges in m³ (block volume to deliver and volume per shift assignment), matching Block.work_required and ProductionRate.rate; shift_hours and machine_daily_hours are hours.

Parameters:
availability_probability: float = 0.9
blackout_biases: list[BlackoutBias]
blackout_duration: tuple[int, int] | int = (1, 2)
blackout_probability: float = 0.1
block_metric_section: str = 'daily'
capability_pool: list[str]
crew_capability_span: tuple[int, int] = (1, 2)
crew_pool: list[str]
landing_capacity: tuple[int, int] | int = (1, 3)
machine_daily_hours: float = 24.0
name: str
num_blocks: tuple[int, int] | int
num_days: tuple[int, int] | int
num_landings: tuple[int, int] | int = 1
num_machines: tuple[int, int] | int
prescription_pool: list[str]
prescription_weights: list[float] | None = None
production_rate: tuple[float, float] = (6.0, 18.0)
role_pool: list[str]
sampling_overrides: dict[str, object] | None = None
shift_hours: tuple[float, float] = (8.0, 12.0)
shifts_per_day: int = 1
system_mix: dict[str, float] | None = None
terrain_pool: list[str]
terrain_weights: list[float] | None = None
tier: str | None = None
work_required: tuple[float, float] = (6.0, 18.0)
class fhops.scenario.synthetic.SyntheticScenarioSpec(num_blocks, num_days, num_machines, landing_capacity=1, blackout_days=None)[source]

Bases: object

Configuration for generating synthetic scenarios.

Parameters:

  • num_blocks (int)

  • num_days (int)

  • num_machines (int)

  • landing_capacity (int)

  • blackout_days (list[int] | None)

blackout_days: list[int] | None = None
landing_capacity: int = 1
num_blocks: int
num_days: int
num_machines: int
fhops.scenario.synthetic.generate_basic(spec)[source]

Generate a minimal scenario matching the supplied specification.

Parameters:

spec (SyntheticScenarioSpec)

Return type:

Scenario

fhops.scenario.synthetic.generate_random_dataset(config, *, seed=123, systems=None)[source]

Generate a random synthetic dataset bundle (scenario + CSV tables).

Parameters:

  • config (SyntheticDatasetConfig) – Sizes, sampling ranges and tier presets (tier fills unset pools and system_mix).

  • seed (int, default 123) – RNG seed; the same config, seed and systems give the same bundle.

  • systems (dict[str, HarvestSystem] | None) – Harvest systems the blocks may use (fhops synth passes the default registry). When given, the fleet is built from them (since 1.0.1, #158): systems with a positive system_mix weight are visited in a weighted random order and kept while the distinct roles of the kept systems fit in the sampled machine count; machines take those roles round-robin (upstream roles first; role_pool is ignored) and every block draws its system from the kept systems by system_mix weight (round-robin without weights). Every block can therefore be delivered unless no system fits the fleet (then the system with the fewest roles is used and some of its roles have no machine). The chosen systems and roles are recorded in metadata (fleet_systems, fleet_roles). Without systems, machines take role_pool roles and blocks have no harvest system.

Returns:

Scenario, CSV tables and metadata.

Return type:

SyntheticDatasetBundle

fhops.scenario.synthetic.generate_with_systems(spec, systems=None)[source]

Generate a scenario and assign blocks round-robin to harvest systems.

Parameters:
Return type:

Scenario

fhops.scenario.synthetic.sampling_config_for(config)[source]

Return the stochastic playback configuration for a synthetic dataset tier.

Parameters:

config (SyntheticDatasetConfig) – Dataset configuration; tier selects a SAMPLING_PRESETS entry (small, medium, large; unknown tiers keep the defaults) and sampling_overrides is deep-merged on top.

Returns:

SamplingConfig(samples=10) updated with the tier preset and overrides. Landing-shock presets are calibrated for per-landing-day shocks: the expected fraction of shocked landing-days is 1 - (1 - probability) ** duration_days (medium ≈ 4.9 %, large ≈ 10.1 %; the default config gives 0.1). Deprecated fields (correlated_days, seed_offset) are dropped before re-validation, so no deprecation warning is emitted.

Return type:

SamplingConfig