necst.procedures.observations.observation_base

exception ObservationAbort[source]

Bases: RuntimeError

Raised when a running observation is asked to abort cooperatively.

class Observation(record_name=None, /, **kwargs)[source]

Bases: ABC

Observation runner.

Parameters:
  • record_name (Optional[str]) – Record name. This will prefixed by auto-generated observation identifier necst_{start_datetime}_{observation_type}.

  • **kwargs – Keyword arguments passed to run().

Examples

>>> obs = necst.procedures.Observation(...)
>>> obs.execute()
observation_type: str
target: Optional[str] = None
execute()[source]
Return type:

None

property start_datetime: Optional[str]
ros2env()[source]
Return type:

Generator[None, None, None]

property record_name: str
before_record_controls()[source]

Hook after privilege acquisition and before legacy record controls.

Return type:

None

after_record_start()[source]

Hook after recorder start and default parameter-file sidecars.

Return type:

None

before_record_stop()[source]

Hook before recorder stop in the cleanup path.

Implementations must not clear active spectral-recording setup here, because the recorder may still be running until record("stop") returns. Use after_record_stop() for deactivation that is safe only after recorder stop.

Return type:

None

after_record_stop()[source]

Hook after recorder stop in the cleanup path.

Return type:

None

allow_legacy_recording_cleanup_controls()[source]

Whether cleanup may send legacy TP/binning reset commands.

File-based spectral-recording setup observations override this to prevent legacy controls from being sent while an active setup remains uncleared after an uncertain recorder-stop failure.

Return type:

bool

allow_legacy_recording_startup_controls()[source]

Whether startup may send legacy recorder-control commands.

File-based spectral-recording setup mode overrides this after setup application so inactive CLI defaults such as ch=None or tp_mode=False are consumed without sending old recorder controls.

Return type:

bool

record_parameter_files()[source]
Return type:

None

abstract run(*args, **kwargs)[source]
Return type:

None

hot(integ_time, id, *, preserve_tracking=False, phase='HOT', location_context=None, geometry=None)[source]
Parameters:
  • integ_time (Union[int, float]) –

  • id (Any) –

  • preserve_tracking (bool) –

  • phase (str) –

  • location_context (Optional[str]) –

  • geometry (Optional[dict]) –

Return type:

None

sky(integ_time, id, *, phase='SKY', location_context=None, geometry=None)[source]
Parameters:
  • integ_time (Union[int, float]) –

  • id (Any) –

  • phase (str) –

  • location_context (Optional[str]) –

  • geometry (Optional[dict]) –

Return type:

None

off(integ_time, id, *, phase='OFF', location_context=None, geometry=None)[source]
Parameters:
  • integ_time (Union[int, float]) –

  • id (Any) –

  • phase (str) –

  • location_context (Optional[str]) –

  • geometry (Optional[dict]) –

Return type:

None

on(integ_time, id, *, phase='ON', location_context=None, geometry=None)[source]
Parameters:
  • integ_time (Union[int, float]) –

  • id (Any) –

  • phase (str) –

  • location_context (Optional[str]) –

  • geometry (Optional[dict]) –

Return type:

None

binning(ch, spectrometer)[source]