`fhops.web` Package

Run discovery and manifest handling for the FHOPS web service.

The service reads the artefacts the CLI writes. This module classifies a directory into one of the three planning lenses (operational, hauling, tactical-operational), collects the known files, and optionally writes a stable run.json manifest. It never invents a file or a KPI.

class fhops.web.discovery.RunRecord(run_id, path, domain, run_type, status, manifest=None, files=<factory>, warnings=<factory>)[source]

Bases: object

One discovered run directory.

Parameters:

run_id (str)
path (Path)
domain (str | None)
run_type (str | None)
status (str)
manifest (dict[str, Any] | None)
files (dict[str, str])
warnings (list[str])

run_id

Stable identifier used in API paths.

Type:: str

path

Absolute run directory.

Type:: pathlib.Path

domain

operational, hauling, or tactical_operational (None when unclassified).

Type:: str | None

run_type

Best-effort run kind, such as solve_heur, haul_solve, or tactical_operational.

Type:: str | None

status

Manifest status when present, otherwise unknown.

Type:: str

manifest

Parsed run.json when present.

Type:: dict[str, Any] | None

files

Logical artefact name -> path relative to the run directory.

Type:: dict[str, str]

warnings

Discovery/classification warnings.

Type:: list[str]

domain: str | None

files: dict[str, str]

manifest: dict[str, Any] | None

path: Path

run_id: str

run_type: str | None

status: str

to_dict()[source]

Return a JSON-ready payload.

Return type:: dict[str, Any]

warnings: list[str]

fhops.web.discovery.build_run_manifest(run_dir, *, domain=None, run_type=None, command=None, scenario=None, solver=None, seed=None, status=None, run_id=None, write=True)[source]

Build (and optionally write) a run.json manifest for a directory.

Parameters:

run_dir (Path) – Existing run directory.
run_type (str | None) – Explicit overrides. When omitted, the manifest is classified from the files.
status (str | None) – Provenance fields. Existing values are preserved unless overridden.
run_id (str | None) – Explicit run id. Defaults to the existing manifest value or the directory name/path slug.
write (bool) – When True, write run.json and return the written payload.
domain (str | None)
run_type
command (str | None)
scenario (str | None)
solver (str | None)
seed (int | None)
status

Returns:

Manifest payload.

Return type:

dict[str, Any]

fhops.web.discovery.classify_run(files)[source]

Classify a run from its logical file names.

Parameters:: files (dict[str, str]) – Logical artefact names from _scan_files() (or a manifest’s files map).
Returns:: (domain, run_type, warnings). Ambiguous combinations are reported rather than guessed.
Return type:: tuple[str | None, str | None, list[str]]

fhops.web.discovery.discover_runs(roots, *, max_depth=2)[source]

Discover run directories under one or more roots.

Parameters:

roots (Sequence[Path]) – Directories to scan. A root may itself be a run directory.
max_depth (int) – Maximum child depth below each root. 1 scans immediate children, 2 grandchildren.

Returns:

Sorted by run_id. Directories with no known artefact are skipped.

Return type:

list[RunRecord]

fhops.web.discovery.load_run_manifest(run_dir)[source]

Load run.json from a run directory, or None when absent.

Raises:: ValueError – When the manifest is not a JSON object.
Parameters:: run_dir (Path)
Return type:: dict[str, Any] | None

fhops.web.discovery.write_run_manifest(run_dir, **kwargs)[source]

Write a run.json manifest and return it. See build_run_manifest().

Parameters:

run_dir (Path)
kwargs (Any)

Return type:

dict[str, Any]

FastAPI application factory for the optional FHOPS web service.

This module imports FastAPI and is only imported when the fhops[web] extra is installed. The core CLI imports fhops.cli.web lazily, so a core install never needs FastAPI.

class fhops.web.app.ServiceConfig(roots=<factory>, host='127.0.0.1', port=8765, root_path='', allow_remote=False, title='FHOPS Web', geometry=None, geometry_crs=None, workspace=<factory>, cli_command=<factory>, allow_agent=False, agent_command=('opencode', 'run', '--format', 'json'), agent_timeout=120.0, lidar_bundles=<factory>, lidar_timeout=300.0, game_level=None, game_precompute=None, game_levels=<factory>, game_precomputes=<factory>, game_agent_model='opencode-go/deepseek-v4.1-flash', game_explanation_cache=None, game_session_store=None)[source]

Bases: object

Web service configuration.

Parameters:

roots (list[Path])
host (str)
port (int)
root_path (str)
allow_remote (bool)
title (str)
geometry (Path | None)
geometry_crs (str | None)
workspace (Path)
cli_command (tuple[str, ...])
allow_agent (bool)
agent_command (tuple[str, ...])
agent_timeout (float)
lidar_bundles (list[Path])
lidar_timeout (float)
game_level (Path | None)
game_precompute (Path | None)
game_levels (list[Path])
game_precomputes (list[Path])
game_agent_model (str)
game_explanation_cache (Path | None)
game_session_store (Path | None)

roots

Directories scanned for run artefacts.

Type:: list[pathlib.Path]

host / port: Bind address. Loopback is the default; a non-loopback host needs allow_remote.

root_path

Optional proxy prefix (for example a notebook proxy path).

Type:: str

allow_remote

Explicit opt-in for a non-loopback bind.

Type:: bool

title

Page/service title.

Type:: str

geometry

Optional block GeoJSON path. When omitted, the service tries the scenario’s geo_block_path and a blocks.geojson file inside the run directory.

Type:: pathlib.Path | None

geometry_crs

Optional source CRS for the geometry. When omitted, the scenario geo_crs or the GeoJSON crs member is used; coordinates are transformed to WGS84 for MapLibre.

Type:: str | None

workspace

Working directory for CLI subprocesses and for resolving scenario paths.

Type:: pathlib.Path

cli_command

Command prefix used to invoke FHOPS. Defaults to the current Python interpreter running -m fhops.cli.main.

Type:: tuple[str, …]

allow_agent

Enable the optional local assistant. Off by default; the assistant runs the installed opencode CLI with the service user’s shell permissions.

Type:: bool

agent_command

Command prefix for the assistant.

Type:: tuple[str, …]

agent_timeout

Assistant subprocess timeout in seconds.

Type:: float

lidar_bundles

Allowlisted roots containing FEMIC LiDAR bundle directories.

Type:: list[pathlib.Path]

lidar_timeout

Timeout for LiDAR mapping subprocesses.

Type:: float

game_level

Optional stand-level game manifest. Must be paired with game_precompute.

Type:: pathlib.Path | None

game_precompute

Optional complete stand precompute artifact.

Type:: pathlib.Path | None

game_levels / game_precomputes: Optional ordered list of stand levels and matching precomputes.

game_agent_model

Model used for optional post-level LLM explanations.

Type:: str

game_explanation_cache

Optional JSON cache for level+choice explanations.

Type:: pathlib.Path | None

game_session_store

Optional JSON file used to persist tournament sessions across restarts.

Type:: pathlib.Path | None

agent_command: tuple[str, ...]

agent_timeout: float

allow_agent: bool

allow_remote: bool

cli_command: tuple[str, ...]

game_agent_model: str

game_explanation_cache: Path | None

game_level: Path | None

game_levels: list[Path]

game_precompute: Path | None

game_precomputes: list[Path]

game_session_store: Path | None

geometry: Path | None

geometry_crs: str | None

host: str

lidar_bundles: list[Path]

lidar_timeout: float

port: int

root_path: str

roots: list[Path]

title: str

workspace: Path

fhops.web.app.create_app(config)[source]

Create the FastAPI application.

Parameters:: config (ServiceConfig) – Service configuration.
Returns:: Application with health, run discovery, run detail, and the static shell.
Return type:: fastapi.FastAPI

fhops.web.app.is_loopback(host)[source]

Return whether a host string is a loopback address.

Parameters:: host (str)
Return type:: bool

fhops.web.app.run_server(config)[source]

Run the service with uvicorn.

Parameters:: config (ServiceConfig) – Service configuration.
Raises:: ValueError – When the bind is refused or uvicorn is unavailable.
Return type:: None

fhops.web.app.validate_bind(host, *, allow_remote)[source]

Refuse a non-loopback bind unless explicitly allowed.

Raises:

ValueError – When host is not loopback and allow_remote is false.

Parameters:

host (str)
allow_remote (bool)

Return type:

None

Skin manifests for the FHOPS web application.

A skin is a JSON manifest over one application. The loader checks every panel, layer, and action against a registry and refuses an unknown name with that name. read_only removes every mutating action.

class fhops.web.skins.SkinManifest(id, title, panels, layers, actions, read_only)[source]

Bases: object

Validated skin manifest.

Parameters:

id (str)
title (str)
panels (tuple[str, ...])
layers (tuple[str, ...])
actions (tuple[str, ...])
read_only (bool)

id / title: Identifier and display label.

panels / layers / actions: Names checked against PANELS, LAYERS, and ACTIONS.

read_only

Remove every mutating action.

Type:: bool

actions: tuple[str, ...]

id: str

layers: tuple[str, ...]

panels: tuple[str, ...]

read_only: bool

title: str

to_dict()[source]

Return a JSON-ready payload.

Return type:: dict[str, Any]

fhops.web.skins.load_skin(path)[source]

Load and validate one skin JSON file.

Parameters:: path (Path)
Return type:: SkinManifest

fhops.web.skins.load_skins(directory=None)[source]

Load every skin JSON file from a directory, sorted by id.

Parameters:: directory (Path | None)
Return type:: list[SkinManifest]

fhops.web.skins.validate_skin(payload)[source]

Validate a skin payload.

Parameters:: payload (dict[str, Any]) – Parsed JSON object with id, title, panels, layers, actions, and read_only.
Returns:: Validated manifest.
Return type:: SkinManifest
Raises:: ValueError – When an id/title is missing, a panel/layer/action is unknown, or a read-only skin contains a mutating action.

Map payload construction for the FHOPS web service.

The map is built from the scenario’s block geometry plus the run’s own assignment, flow, or harvest artefacts. It never invents geometry for a yard, mill, depot, or road; those layers are reported as unavailable until a geometry source is supplied.

fhops.web.mapdata.layer_registry(record, geometry, *, workspace=None)[source]

Return every known layer with availability and a legend/reason.

Parameters:

record (RunRecord)
geometry (dict[str, dict[str, Any]])
workspace (Path | None)

Return type:

list[dict[str, Any]]

fhops.web.mapdata.load_block_geometry(record, geometry_path=None, geometry_crs=None)[source]

Load block geometry indexed by block_id.

Parameters:

record (RunRecord) – Discovered run.
geometry_path (Path | None) – Optional GeoJSON path from --geometry.
geometry_crs (str | None) – Optional source CRS for --geometry or a run-local GeoJSON. When omitted, the scenario geo_crs or the GeoJSON top-level crs is used; the default is EPSG:4326. Coordinates are transformed to EPSG:4326 for MapLibre.

Returns:

Mapping from block id to GeoJSON geometry. An empty mapping means no geometry is available.

Return type:

dict[str, dict[str, Any]]

Raises:

ValueError – When a candidate file exists but is not a FeatureCollection.

fhops.web.mapdata.map_payload(record, geometry, *, day=None, period=None, shift=None, layers=None, workspace=None)[source]

Build a GeoJSON FeatureCollection for the requested layers.

Raises:

ValueError – When a requested layer is unknown or unavailable.

Parameters:

record (RunRecord)
geometry (dict[str, dict[str, Any]])
day (int | None)
period (str | None)
shift (str | None)
layers (list[str] | None)
workspace (Path | None)

Return type:

dict[str, Any]

fhops.web.mapdata.time_payload(record)[source]

Return the domain’s time control values.

Parameters:: record (RunRecord)
Return type:: dict[str, Any]

fhops.web Package

`fhops.web` Package