fhops.web Package

Run discovery and the run.json manifest are implemented in fhops.web.discovery. The FastAPI application factory lives in fhops.web.app and is imported only when the optional fhops[web] extra is installed.

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

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

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:
Return type:

None

Process management for live runs started by the web service.

The service starts a run by invoking the FHOPS CLI in a subprocess. It never imports the optimizer into the request process. The manager keeps only process state and a bounded output directory; all run values still come from the files the CLI writes.

class fhops.web.runs.RunControlRequest(*, action)[source]

Bases: BaseModel

Run control action.

Parameters:

action (Literal['pause', 'resume', 'stop'])

action: Literal['pause', 'resume', 'stop']
model_config = {}

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

class fhops.web.runs.RunManager[source]

Bases: object

Start, monitor, and control CLI runs.

control(run_id, action)[source]

Apply pause/resume/stop to a live run.

Parameters:
Return type:

dict[str, Any]

get(run_id)[source]

Return a live run by id.

Parameters:

run_id (str)

Return type:

RunProcess | None

shutdown()[source]

Terminate live subprocesses when the service shuts down.

Return type:

None

start(request, *, workspace, run_root, cli_command)[source]

Start a run and return its process state.

Parameters:

  • request (RunRequest)

  • workspace (Path)

  • run_root (Path)

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

Return type:

RunProcess

status(record)[source]

Return a JSON-ready status payload.

Parameters:

record (RunProcess)

Return type:

dict[str, Any]

async stream(run_id)[source]

Yield SSE-formatted events for a live run.

Parameters:

run_id (str)

class fhops.web.runs.RunRequest(*, domain, run_type, scenario, options=<factory>)[source]

Bases: BaseModel

Request to start a run.

Parameters:

  • domain (Literal['operational', 'hauling', 'tactical_operational'])

  • run_type (str)

  • scenario (str)

  • options (dict[str, Any])

domain

One of operational, hauling, or tactical_operational.

Type:

Domain

run_type

Domain-specific run kind: solve_heur, solve_mip_operational, haul_solve, or tactical_operational.

Type:

str

scenario

Scenario path, relative to the workspace or absolute under it.

Type:

str

options

Optional iterations, seed, time_limit, day, and solver values.

Type:

dict[str, Any]

domain: Domain
model_config = {}

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

options: dict[str, Any]
run_type: str
scenario: str
fhops.web.runs.build_command(request, *, scenario, out_dir, cli_command, workspace=None, run_root=None)[source]

Build the allowlisted CLI command for a run request.

Raises:

ValueError – For an unsupported run type or a missing/invalid option.

Parameters:

  • request (RunRequest)

  • scenario (Path)

  • out_dir (Path)

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

  • workspace (Path | None)

  • run_root (Path | None)

Return type:

list[str]

Validated scenario editing for the web UI.

The editor never builds a parallel scenario model. It edits the YAML that the CLI already reads, loads the result with the same loader, and only then replaces the file. Top-level sections are replaced in place so comments and ordering elsewhere in the scenario are preserved. Tactical edits are separate overlays; the base tactical bundle is never rewritten.

exception fhops.web.editing.ScenarioEditError[source]

Bases: ValueError

A scenario edit was refused by the FHOPS loader.

fhops.web.editing.read_editable(scenario_path, *, domain, overlay_root)[source]

Return the editable payload for one scenario.

Parameters:

  • scenario_path (Path)

  • domain (str | None)

  • overlay_root (Path)

Return type:

dict[str, Any]

fhops.web.editing.save_tactical_overlay(scenario_path, overlay, *, overlay_root)[source]

Validate and persist a tactical overlay without rewriting the base bundle.

Parameters:
Return type:

dict[str, Any]

fhops.web.editing.update_hauling(scenario_path, *, driver_policy=None, contractors=None)[source]

Update the inline hauling driver policy or contractor preferred workloads.

Parameters:
Return type:

dict[str, Any]

fhops.web.editing.update_locks(scenario_path, locks)[source]

Replace top-level locked_assignments after loader validation.

Parameters:
Return type:

dict[str, Any]

fhops.web.editing.update_objective(scenario_path, weights)[source]

Update top-level objective_weights after loader validation.

Parameters:
Return type:

dict[str, Any]

Read-only LiDAR evidence payloads for the web UI.

The service reads the scenario’s committed lidar_products and lidar_support sections and the P14.4 advisory operability report. It never discovers, downloads, or processes source LiDAR, and it does not import the optional GeoPandas/rasterio stack.

fhops.web.lidarview.lidar_map_values(record, workspace=None)[source]

Return block-level values for LiDAR map layers.

Parameters:

  • record (RunRecord)

  • workspace (Path | None)

Return type:

dict[str, dict[str, Any]]

fhops.web.lidarview.lidar_payload(record, workspace=None)[source]

Build the read-only LiDAR panel payload for one run.

Parameters:

  • record (RunRecord)

  • workspace (Path | None)

Return type:

dict[str, Any]

fhops.web.lidarview.scenario_path_for_run(record, workspace=None)[source]

Resolve the scenario path recorded in run.json.

Parameters:

  • record (RunRecord)

  • workspace (Path | None)

Return type:

Path | None

LiDAR bundle validation, registration, and mapping actions for the web service.

Bundle validation and scenario registration are core FHOPS operations. Mapping runs through the CLI subprocess so the request process never imports GeoPandas/rasterio. Bundle roots and scenario paths are allowlisted.

class fhops.web.lidarworkflow.LidarBundleRequest(*, bundle, expected_bundle_id=None, expected_bundle_version=None)[source]

Bases: BaseModel

A bundle action request.

Parameters:

  • bundle (str)

  • expected_bundle_id (str | None)

  • expected_bundle_version (str | None)

bundle: str
expected_bundle_id: str | None
expected_bundle_version: str | None
model_config = {}

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

class fhops.web.lidarworkflow.LidarMapRequest(*, bundle, expected_bundle_id=None, expected_bundle_version=None, products=None, statistic='mean', include_blocks=True, include_landings=True, roads_geojson=None, dry_run=False)[source]

Bases: LidarBundleRequest

Map bundle products into the scenario’s support rows.

Parameters:

  • bundle (str)

  • expected_bundle_id (str | None)

  • expected_bundle_version (str | None)

  • products (dict[str, str] | None)

  • statistic (str)

  • include_blocks (bool)

  • include_landings (bool)

  • roads_geojson (str | None)

  • dry_run (bool)

dry_run: bool
include_blocks: bool
include_landings: bool
model_config = {}

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

products: dict[str, str] | None
roads_geojson: str | None
statistic: str
exception fhops.web.lidarworkflow.LidarWorkflowError[source]

Bases: RuntimeError

A LiDAR web action cannot be completed safely.

fhops.web.lidarworkflow.list_bundles(roots)[source]

List manifest-bearing bundle directories under the configured roots.

Parameters:

roots (list[Path])

Return type:

dict[str, Any]

fhops.web.lidarworkflow.map_bundle_products(scenario_path, bundle_path, *, workspace, cli_command, request, timeout=300.0)[source]

Run fhops scenario lidar map in a subprocess and return its JSON result.

Parameters:

  • scenario_path (Path)

  • bundle_path (Path)

  • workspace (Path)

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

  • request (LidarMapRequest)

  • timeout (float)

Return type:

dict[str, Any]

fhops.web.lidarworkflow.register_bundle(scenario_path, bundle_path, *, expected_bundle_id=None, expected_bundle_version=None, dry_run=False)[source]

Register a bundle pin in a scenario.

Parameters:

  • scenario_path (Path)

  • bundle_path (Path)

  • expected_bundle_id (str | None)

  • expected_bundle_version (str | None)

  • dry_run (bool)

Return type:

dict[str, Any]

fhops.web.lidarworkflow.resolve_bundle(value, roots)[source]

Resolve a bundle name under the configured roots.

Parameters:
Return type:

Path

fhops.web.lidarworkflow.validate_bundle(bundle_path, *, expected_bundle_id=None, expected_bundle_version=None)[source]

Validate a bundle and return the report.

Parameters:

  • bundle_path (Path)

  • expected_bundle_id (str | None)

  • expected_bundle_version (str | None)

Return type:

dict[str, Any]

Optional local assistant bridge.

The assistant is disabled unless the service is started with --allow-agent. When enabled it runs the installed opencode CLI as a subprocess in the configured workspace; the service never reads provider keys and never supplies a shell.

exception fhops.web.agent.AgentError[source]

Bases: RuntimeError

The assistant subprocess failed.

class fhops.web.agent.AgentRequest(*, message, model=None)[source]

Bases: BaseModel

A single assistant prompt.

Parameters:

  • message (Annotated[str, MinLen(min_length=1), MaxLen(max_length=4000)])

  • model (Annotated[str | None, MaxLen(max_length=200)])

message: str
model: str | None
model_config = {}

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

fhops.web.agent.run_agent(request, *, workspace, command, timeout)[source]

Run the assistant CLI and return its textual response.

Parameters:
Return type:

dict[str, Any]

Key objects

  • fhops.web.discovery.RunRecord – one discovered run directory.

  • fhops.web.discovery.discover_runs() – scan roots and classify operational, hauling, and tactical runs.

  • fhops.web.discovery.build_run_manifest() / fhops.web.discovery.write_run_manifest() – build or write run.json.

  • fhops.web.app.ServiceConfig – roots, bind, proxy prefix, remote-bind opt-in, geometry, geometry CRS, workspace, and CLI command prefix.

  • fhops.web.runs.RunManager / fhops.web.runs.RunRequest – start, monitor, stream, and control CLI subprocesses without importing the optimizer.

  • fhops.web.editing.update_objective() / update_locks / update_hauling / save_tactical_overlay – validated edits through the existing loader.

  • fhops.web.lidarview.lidar_payload() – read-only scenario LiDAR evidence for the panel.

  • fhops.web.lidarview.lidar_map_values() – block values for LiDAR map layers.

  • fhops.web.lidarworkflow.resolve_bundle() / validate_bundle / register_bundle / map_bundle_products – allowlisted LiDAR workflow actions.

  • fhops.web.agent.run_agent() – opt-in bridge to the installed opencode CLI.

  • fhops.web.app.create_app() – FastAPI factory.

  • fhops.web.app.validate_bind() – loopback/remote-bind guard.

  • fhops.web.skins.load_skins() / fhops.web.skins.validate_skin() – skin registry.

  • fhops.web.skins.SkinManifest – validated domain skin.

  • fhops.web.mapdata.load_block_geometry() – resolve block geometry from --geometry, the scenario, or a run-local GeoJSON.

  • fhops.web.mapdata.layer_registry() – available/unavailable map layers with reasons.

  • fhops.web.mapdata.time_payload() – day/shift/period control values.

  • fhops.web.mapdata.map_payload() – GeoJSON FeatureCollection for the requested layers.

CLI

fhops web --runs runs/ --workspace /path/to/fhops-checkout
fhops web --runs runs/ --allow-agent
fhops web --runs runs/ --lidar-bundles /path/to/femic/bundles
fhops web manifest runs/run-a --status optimal
fhops web discover runs/