necst.rx.spectral_recording_setup

Parser/resolver for NECST/XFFTS spectral recording setup snapshots.

This module is intentionally self-contained and depends only on the Python standard library plus an optional TOML reader (tomllib, tomli or toml). It implements the PR1 foundation for the spectral recording redesign:

  • parse human-edited lo_profile.toml, recording_window_setup.toml and beam_model.toml;

  • expand strict two-level stream_groups into a normalized stream table;

  • resolve per-stream recording mode and channel ranges;

  • validate primary keys, references, and DB table paths;

  • emit a deterministic spectral_recording_snapshot.toml.

Runtime recording, ROS services, and converter/sunscan adapters are deliberately left to later PRs. The classes and functions here are written so they can be unit-tested without ROS imports.

exception SpectralRecordingSetupError[source]

Bases: ValueError

Raised when a setup file is structurally invalid.

exception SpectralRecordingValidationError[source]

Bases: SpectralRecordingSetupError

Raised when references or resolved values fail validation.

read_toml(path)[source]
Parameters:

path (os.PathLike[str] | str) –

Return type:

Dict[str, Any]

sha256_file(path)[source]
Parameters:

path (os.PathLike[str] | str) –

Return type:

str

sha256_text(text)[source]
Parameters:

text (str) –

Return type:

str

dumps_toml(data)[source]

Write a deterministic, minimal TOML representation.

The writer supports scalars, lists of scalars, nested tables, and arrays of tables. It is deliberately small so that canonical hashing is independent of third-party TOML writer formatting.

Parameters:

data (Mapping[str, Any]) –

Return type:

str

velocity_to_frequency_hz(rest_frequency_hz, velocity_kms, definition)[source]
Parameters:

  • rest_frequency_hz (float) –

  • velocity_kms (float) –

  • definition (str) –

Return type:

float

resolve_velocity_window_to_channels(*, rest_frequency_hz, velocity_definition, vmin_kms, vmax_kms, margin_kms, frequency_hz, clip_policy='strict', requested_velocity_frame='LSRK', vlsrk_correction_kms=None)[source]

Resolve a velocity window to an inclusive/exclusive full-channel range.

Direction definitions:

  • full channel index is zero-based;

  • output computed_ch_start is inclusive;

  • output computed_ch_stop is exclusive;

  • any channel whose center falls within the effective frequency interval is included. The result is always sorted by full channel index, independent of increasing/decreasing frequency order.

Parameters:

  • rest_frequency_hz (float) –

  • velocity_definition (str) –

  • vmin_kms (float) –

  • vmax_kms (float) –

  • margin_kms (float) –

  • frequency_hz (Sequence[float]) –

  • clip_policy (str) –

  • requested_velocity_frame (str) –

  • vlsrk_correction_kms (Optional[float]) –

Return type:

Dict[str, Any]

resolve_velocity_window_to_channels_linear(*, rest_frequency_hz, velocity_definition, vmin_kms, vmax_kms, margin_kms, start_hz, step_hz, nchan, clip_policy='strict', requested_velocity_frame='LSRK', vlsrk_correction_kms=None)[source]

Resolve a velocity window on a linear WCS without allocating an array.

Parameters:

  • rest_frequency_hz (float) –

  • velocity_definition (str) –

  • vmin_kms (float) –

  • vmax_kms (float) –

  • margin_kms (float) –

  • start_hz (float) –

  • step_hz (float) –

  • nchan (int) –

  • clip_policy (str) –

  • requested_velocity_frame (str) –

  • vlsrk_correction_kms (Optional[float]) –

Return type:

Dict[str, Any]

resolve_spectral_recording_setup(*, lo_profile, recording_setup, beam_model, setup_id, input_paths=None, setup_override_policy='strict', created_utc=None, velocity_reference_context=None)[source]
Parameters:

  • lo_profile (Mapping[str, Any]) –

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

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

  • setup_id (str) –

  • input_paths (Optional[Mapping[str, os.PathLike[str] | str]]) –

  • setup_override_policy (str) –

  • created_utc (Optional[str]) –

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

Return type:

Dict[str, Any]

load_and_resolve_spectral_recording_setup(*, lo_profile_path, recording_window_setup_path, beam_model_path, setup_id, setup_override_policy='strict', velocity_reference_context=None)[source]
Parameters:

  • lo_profile_path (os.PathLike[str] | str) –

  • recording_window_setup_path (Optional[Union[PathLike[str], str]]) –

  • beam_model_path (Optional[Union[PathLike[str], str]]) –

  • setup_id (str) –

  • setup_override_policy (str) –

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

Return type:

Dict[str, Any]

validate_snapshot(snapshot)[source]
Parameters:

snapshot (Mapping[str, Any]) –

Return type:

None

local_channel_to_full_channel(saved_ch_start, c_local)[source]
Parameters:

  • saved_ch_start (int) –

  • c_local (int) –

Return type:

int

if_frequency_at_full_channel(axis, c_full)[source]
Parameters:

  • axis (Mapping[str, Any]) –

  • c_full (int) –

Return type:

float

if_frequency_at_saved_local_channel(axis, saved_ch_start, c_local)[source]
Parameters:

  • axis (Mapping[str, Any]) –

  • saved_ch_start (int) –

  • c_local (int) –

Return type:

float

write_snapshot(path, snapshot)[source]
Parameters:

  • path (os.PathLike[str] | str) –

  • snapshot (Mapping[str, Any]) –

Return type:

None

main_resolve(argv=None)[source]
Parameters:

argv (Optional[Sequence[str]]) –

Return type:

int

main_validate(argv=None)[source]
Parameters:

argv (Optional[Sequence[str]]) –

Return type:

int

previous

necst.rx.spectral_recording_runtime

next

necst.rx.spectral_recording_sg