necst.rx.spectral_recording_sg

Signal-generator apply/verify helpers for spectral recording LO profiles.

This module implements the PR3 SG layer without importing ROS at module import time. The pure functions are unit-testable with fake readback objects; the CLI imports necst.core.Commander only when an actual apply/verify command is requested.

Frequency convention

sg_set_frequency_hz in lo_profile.toml is the frequency sent to the physical signal generator. Commander currently publishes LocalSignal.freq in GHz, so readback objects with a freq attribute are interpreted as GHz unless an explicit frequency_hz/sg_readback_frequency_hz field exists.

exception SpectralRecordingSGError[source]

Bases: RuntimeError

Base error for SG apply/verify failures.

exception SpectralRecordingSGValidationError[source]

Bases: SpectralRecordingSGError

Raised when a profile/readback pair violates strict SG semantics.

sg_devices_from_profile(lo_profile)[source]
Parameters:

lo_profile (Mapping[str, Any]) –

Return type:

Dict[str, Dict[str, Any]]

build_sg_apply_plan(lo_profile, sg_id=None)[source]
Parameters:

  • lo_profile (Mapping[str, Any]) –

  • sg_id (Optional[str]) –

Return type:

Dict[str, Dict[str, Any]]

filter_sg_apply_plan(plan, sg_ids)[source]

Return a deterministic subset of an SG apply/verify plan.

Missing IDs are rejected so that a typo in .obs does not silently disable SG preflight checks.

Parameters:

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

  • sg_ids (Iterable[str]) –

Return type:

Dict[str, Dict[str, Any]]

active_sg_ids_from_snapshot(snapshot, *, fail_on_legacy_chain=True)[source]

Infer active SG IDs from a resolved spectral-recording snapshot.

The active path is streams -> lo_chains -> lo_roles -> sg_id. Fixed LO roles are intentionally skipped. Legacy inline LO chains (lo1_hz/sb1/…) do not encode which physical SG generated the LO; if such a chain is active and fail_on_legacy_chain is true, raise a clear error instead of giving a false sense that hardware was checked.

Parameters:

  • snapshot (Mapping[str, Any]) –

  • fail_on_legacy_chain (bool) –

Return type:

Tuple[str, …]

select_sg_preflight_plan(*, lo_profile, snapshot, scope='active_lo_chains', sg_ids=None)[source]

Build the SG subset to check/set for observation-time preflight.

Parameters:

  • lo_profile (Mapping[str, Any]) –

  • snapshot (Mapping[str, Any]) –

  • scope (str) –

  • sg_ids (Optional[Iterable[str]]) –

Return type:

Dict[str, Dict[str, Any]]

run_sg_preflight(commander, *, plan, policy, timeout_sec=10.0, allow_command_echo=False, poll_interval_sec=0.1)[source]

Execute observation-time SG preflight.

Policies: - none: do nothing. - verify_only: read and strictly verify fresh SG readbacks. - apply_and_verify: send set/stop according to the plan, then verify. - warn_only: verify fresh readbacks but return failures instead of raising.

Parameters:

  • commander (Any) –

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

  • policy (str) –

  • timeout_sec (float) –

  • allow_command_echo (bool) –

  • poll_interval_sec (float) –

Return type:

Dict[str, Dict[str, Any]]

apply_sg_plan_entry(commander, entry)[source]
Parameters:

  • commander (Any) –

  • entry (Mapping[str, Any]) –

Return type:

None

verify_sg_readback(*, sg_id, entry, readback, verify_started_at, strict=True, allow_command_echo=False)[source]

Validate one SG readback and return normalized values.

Strict mode requires an id match, non-stale timestamp, frequency tolerance, power tolerance when power is present, and a real device readback unless allow_command_echo is explicitly set.

Parameters:

  • sg_id (str) –

  • entry (Mapping[str, Any]) –

  • readback (Any) –

  • verify_started_at (float) –

  • strict (bool) –

  • allow_command_echo (bool) –

Return type:

Dict[str, Any]

poll_sg_readback(commander, *, sg_id, entry, verify_started_at, timeout_sec, strict=True, allow_command_echo=False, poll_interval_sec=0.1)[source]

Poll Commander until a fresh id-matching SG readback verifies or times out.

Parameters:

  • commander (Any) –

  • sg_id (str) –

  • entry (Mapping[str, Any]) –

  • verify_started_at (float) –

  • timeout_sec (float) –

  • strict (bool) –

  • allow_command_echo (bool) –

  • poll_interval_sec (float) –

Return type:

Dict[str, Any]

summarize_lo_profile(path)[source]
Parameters:

path (Path) –

Return type:

str

main_lo_profile(argv=None)[source]
Parameters:

argv (Optional[Iterable[str]]) –

Return type:

int

previous

necst.rx.spectral_recording_setup

next

necst.rx.spectrometer