necst.web.status_model

Shared read-only status model for NECST operator/progress UIs.

This module intentionally contains no telescope command path. It reads the progress sidecar files and optional live telemetry snapshots, then returns a canonical dictionary that can be consumed by both the diagnostic progress monitor and the future operator console.

safe_name(value)

Parameters:

value (Any) –

Return type:

str

progress_root(value=None)

Parameters:

value (Optional[str]) –

Return type:

Path

live_status_max_age_sec()

Maximum age for live ROS status before it is treated as stale.

Return type:

float

strip_record_file_header(text)

Remove leading NECST FileWriter comment headers if present.

Parameters:

text (str) –

Return type:

str

read_json(path)

Parameters:

path (Optional[Path]) –

Return type:

Optional[Dict[str, Any]]

read_jsonl(path, limit=None)

Parameters:

• path (Optional[Path]) –

• limit (Optional[int]) –

Return type:

List[Dict[str, Any]]

current_snapshot_path(root)

Parameters:

root (Path) –

Return type:

Path

current_record_name(root)

Parameters:

root (Path) –

Return type:

Optional[str]

progress_record_dir(root, record_name)

Return the progress sidecar directory for a record name, if known.

Parameters:

• root (Path) –

• record_name (Optional[Any]) –

Return type:

Optional[Path]

default_record_root()

Return the NECST FileWriter root used by recorder.py.

The recorder defaults to $NECST_RECORD_ROOT or ~/data. This helper mirrors that logic for display only; it does not create directories.

Return type:

Path

record_path_display_mode()

Return the preferred path kind for operator-facing record display.

The operator usually copies the path into an analysis notebook on the local PC or mounted filesystem, so the default is local. Set NECST_CONSOLE_RECORD_PATH_MODE=container (or NECST_CONSOLE_SHOW_CONTAINER_RECORD_PATH=1) only when the Docker/container path itself should be displayed.

Return type:

str

local_record_root()

Return the local-PC/mounted record root used for operator display.

A configured host-side mount point is preferred. If none is configured, use the recorder root as the best available path instead of hiding the field. This keeps the UI useful by default; deployments that explicitly want the Docker/container path can set NECST_CONSOLE_RECORD_PATH_MODE=container.

Return type:

Path

recording_data_dir(record_name)

Return the expected observation data directory for a record name.

NECST RecorderController calls Recorder(record_root).start_recording(name). When name is relative, files are written under the record root; if name is already absolute, pathlib keeps it absolute.

Parameters:

record_name (Optional[Any]) –

Return type:

Optional[Path]

local_recording_data_dir(record_name)

Return the local-PC/mounted display path for a record name.

Parameters:

record_name (Optional[Any]) –

Return type:

Optional[Path]

latest_events_path(root, snapshot=None)

Parameters:

• root (Path) –

• snapshot (Optional[Mapping[str, Any]]) –

Return type:

Optional[Path]

latest_plan_path(root, snapshot=None)

Parameters:

• root (Path) –

• snapshot (Optional[Mapping[str, Any]]) –

Return type:

Optional[Path]

read_recent_events(path, limit)

Parameters:

• path (Optional[Path]) –

• limit (int) –

Return type:

List[Dict[str, Any]]

read_all_events(path)

Parameters:

path (Optional[Path]) –

Return type:

List[Dict[str, Any]]

apply_dynamic_remaining(snapshot, *, now_unix=None)

Return a display/API snapshot with ETA counted down from completion time.

Parameters:

• snapshot (Optional[Dict[str, Any]]) –

• now_unix (Optional[float]) –

Return type:

Optional[Dict[str, Any]]

make_idle_live_snapshot(live_payload, *, now_unix=None)

Build a minimal display-only snapshot for idle live telemetry.

This snapshot is used only by progress.py. It is not written back to current_observation_progress.json and therefore does not affect the observation runtime.

Parameters:

• live_payload (Mapping[str, Any]) –

• now_unix (Optional[float]) –

Return type:

Dict[str, Any]

merge_live_telemetry(snapshot, live)

Return a display-only snapshot augmented with live ROS telemetry.

The progress JSON files remain the source of truth while an observation is active. Live ROS values are merged only in the CLI/web process so the observation runtime stays free of high-frequency subscriptions.

If no progress snapshot exists but live antenna position telemetry is available, return a minimal idle snapshot so observers can still see current encoder Az/El before starting an observation. This idle snapshot is display-only and is never written to disk.

Parameters:

• snapshot (Optional[Dict[str, Any]]) –

• live (Optional[Mapping[str, Any]]) –

Return type:

Optional[Dict[str, Any]]

build_operator_status(snapshot, *, plan=None, events=None, paths=None, server_time_unix=None)

Build a compact, read-only status summary for operator-facing UIs.

The output is deliberately conservative: missing fields are represented as None or unknown rather than guessed. It must be safe to call while the telescope is idle and while ROS/neclib are unavailable.

Parameters:

• snapshot (Optional[Mapping[str, Any]]) –

• plan (Optional[Mapping[str, Any]]) –

• events (Optional[List[Mapping[str, Any]]]) –

• paths (Optional[Mapping[str, Any]]) –

• server_time_unix (Optional[float]) –

Return type:

Dict[str, Any]

build_progress_status_state(root, *, events_limit=12, live_payload=None, time_sync_payload=None, server_time_unix=None)

Read progress sidecars and return raw inputs plus compact status.

This is the shared read-only boundary between progress.py and the future operator console. It does not render HTML and it does not send commands.

Parameters:

• root (Path) –

• events_limit (int) –

• live_payload (Optional[Mapping[str, Any]]) –

• time_sync_payload (Optional[Mapping[str, Any]]) –

• server_time_unix (Optional[float]) –

Return type:

Dict[str, Any]

previous

necst.web.site_config

next

necst.az_unwrap_limits