pypeit.state.run_state module

Data model and I/O for the PypeIt reduction state.

While run_pypeit reduces data it records, per calibration group and per detector/mosaic, the status of each calibration step (plus step-specific metrics and per-slit/order detail). This state is held by the pydantic RunPypeItState model and serialized to <pypeit_root>_state.json in the reduction directory. It is a live record — updated at each step transition — and is consumed by tools rather than edited by hand: the PypeIt Dashboard (PypeIt Dashboard) and pypeit_status both read it (see PypeIt Reduction State in the documentation).

State I/O is deliberately non-essential to the reduction: it is written through RunPypeItState.safe_write() / safe_update_calib(), which log and swallow any error so that state bookkeeping can never abort a run.

Generated by JXP and Claude.

class pypeit.state.run_state.AlignCalibState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['align']

class pypeit.state.run_state.ArcCalibState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['arc']

class pypeit.state.run_state.BaseCalibState(**data)

Bases: BaseModel

_abc_impl = <_abc._abc_data object>

calib_id: int

det: Union[int, List[int]]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: str

class pypeit.state.run_state.BiasCalibState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

mean: Optional[float]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

status: Literal['complete', 'fail', 'undone', 'running', 'success']

std: Optional[float]

step: Literal['bias']

class pypeit.state.run_state.DarkCalibState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['dark']

class pypeit.state.run_state.FlatCorrectionMetric(**data)

Bases: BaseModel

Mean and scatter of one applied flat-field correction over a single slit’s good, on-slit pixels.

Generated by JXP and Claude.

_abc_impl = <_abc._abc_data object>

mean: Optional[float]

model_config: ClassVar[ConfigDict] = {}

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

rms: Optional[float]

class pypeit.state.run_state.FlatsSlit(**data)

Bases: BaseModel

Per-slit flat-field state: an overall status plus the per-correction mean/RMS metrics.

Generated by JXP and Claude.

_abc_impl = <_abc._abc_data object>

corrections: Optional[Dict[str, FlatCorrectionMetric]]

model_config: ClassVar[ConfigDict] = {}

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

status: Literal['success', 'fail', 'skip', 'undone']

class pypeit.state.run_state.FlatsState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

corrections: Optional[List[str]]

det: int | List[int]

illumflat_files: Optional[List[str]]

input_files: Optional[List[str]]

lampoff_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

pixelflat_files: Optional[List[str]]

pixelflat_source: Optional[str]

qa_files: Optional[List[str]]

required: bool

slits: Optional[Dict[int, FlatsSlit]]

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['flats']

class pypeit.state.run_state.RunPypeItState(**data)

Bases: BaseModel

The state of a PypeIt run.

This pydantic model records, for one .pypeit reduction, the status of every calibration step per calibration group and per detector/mosaic. Each calibration field (bias, dark, slits, arc, tiltimg, wv_calib, tilts, scattlight, flats, align) is a list of per-(calib_id, det) entries (subclasses of BaseCalibState) carrying that step’s status, required flag, input/output/QA files, and step-specific metrics (and, for slits/wv_calib/tilts/flats, per-slit/order detail).

It is written to <pypeit_root>_state.json (see outfile) and is updated incrementally during a run; use safe_write() / safe_update_calib() so state I/O can never abort the reduction, and get_status() for a tabular summary. It is read by the PypeIt Dashboard and pypeit_status; it is not meant to be edited by hand.

Generated by JXP and Claude.

Variables:

• pypeit_file (str) – The .pypeit file this state belongs to.

• current_step (str) – The step most recently updated.

• current_det (int) – The detector most recently updated.

• current_calibID (int) – The calibration group most recently updated.

• previous_step (str) – The step updated before current_step.

• path (str) – Optional explicit path for the state JSON file; if None, outfile is derived from pypeit_file.

_abc_impl = <_abc._abc_data object>

_print_science_status()

Print the per-exposure science-frame status table to stdout (the four macro-steps, object count, and product presence).

Generated by JXP and Claude.

add_or_get_science(frame, det, calib_id=-1, objtype='science', comb_id=None, bkg_id=None)

Find the science entry for (frame, det) or create it.

Generated by JXP and Claude.

Parameters:

• frame (str) – The exposure basename.

• det (int, tuple, list) – Detector or mosaic.

• calib_id (int, optional) – Calibration group ID.

• objtype (str, optional) – ‘science’ or ‘standard’.

• comb_id (int, optional) – Combination group ID.

• bkg_id (int, optional) – Background (A-B) group ID.

Returns:

The existing or newly-created entry.

Return type:

ScienceFrameState

align: Optional[List[AlignCalibState]]

arc: Optional[List[ArcCalibState]]

bias: Optional[List[BiasCalibState]]

current_calibID: int

current_det: int

current_step: str

dark: Optional[List[DarkCalibState]]

flats: Optional[List[FlatsState]]

get_science_status()

Return a pandas DataFrame summarizing the science-frame state: one row per (frame, det) with the four macro-step statuses, object count, and product presence.

Generated by JXP and Claude.

Returns:

None if there are no science entries.

Return type:

pandas.DataFrame or None

get_status()

Summarize the state as a tabular, per-step status overview.

This is intentionally less detailed than the full serialized state (write()): it omits the per-slit detail and metrics, giving a scannable health overview (used by print_status() and the dashboard’s Status view).

Generated by JXP and Claude.

Returns:

One row per (calibration_group, detector, step) with columns calibration_group, detector, steps, required, status, and output_file; missing entries are filled with "--". Returns None if no calibration entries exist.

Return type:

pandas.DataFrame or None

load(path=None)

Load the state from outfile, if it exists.

Generated by JXP and Claude.

Parameters:

path (str, optional) – Unused; retained for backward compatibility (the file read is outfile).

Returns:

The state validated from the JSON file, or self unchanged if no state file is present.

Return type:

RunPypeItState

merge_from_disk()

Overlay the calibration and science statuses from the existing on-disk state file onto this state, matching entries by (calib_id, det) and (frame, det).

A step-runner script (pypeit_run_to_calibstep / pypeit_reduce_by_step) starts with a fresh state — its calibrations are the required (undone) set and it has no science — so writing it would reset the other portion of the shared *_state.json (e.g. a science step-build would blank out the calibration statuses a prior calibration build wrote, and vice versa). Calling this first preserves whatever the other runner already recorded. This instance’s outfile (and pypeit_file) is left untouched, so the subsequent write goes to the same file. Best-effort: a missing or unreadable state file leaves self unchanged.

Generated by JXP and Claude.

Returns:

self (updated in place).

Return type:

RunPypeItState

model_config: ClassVar[ConfigDict] = {}

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

property outfile

Path to the state JSON file.

Generated by JXP and Claude.

Returns:

path if set, else the .pypeit file name with the .pypeit extension replaced by _state.json.

Return type:

str

path: Optional[str]

previous_step: str

print_status()

Pretty-print the reduction state to the terminal: a per-(group, detector) calibration table (from get_status()) followed by the science-frame status (_print_science_status()). Used by pypeit_status.

Generated by JXP and Claude.

Returns:

None.

pypeit_file: str

safe_update_calib(step, calib_id, det, key, value, slit=None)

Wrapper on update_calib() that catches and logs any error so that state bookkeeping can never crash a PypeIt reduction.

Generated by JXP and Claude.

Parameters:

• step (str) – Calibration step name.

• calib_id (int) – Calibration group ID.

• det (int, tuple, list) – Detector or detector mosaic.

• key (str) – Field to set on the step (or per-slit) entry.

• value – Value to assign (or append, for list-valued fields).

• slit (int, optional) – Slit/order ID for a per-slit update.

Returns:

True if the update succeeded, False otherwise.

Return type:

bool

safe_update_science(frame, det, step=None, status=None, calib_id=-1, objtype='science', **fields)

Get-or-create a science entry and update a step status and/or top-level fields, catching and logging any error so science bookkeeping can never crash a reduction.

Generated by JXP and Claude.

Parameters:

• frame (str) – The exposure basename.

• det (int, tuple, list) – Detector or mosaic.

• step (str, optional) – One of science_steps; if given, its status is set to status.

• status (str, optional) – Status to assign to step.

• calib_id (int, optional) – Calibration group (on create).

• objtype (str, optional) – ‘science’/’standard’ (on create).

• **fields – Other top-level ScienceFrameState fields to set (e.g. nobj, spec1d_file, spec2d_file).

Returns:

The entry (None on failure).

Return type:

ScienceFrameState or None

safe_write()

Write the state, catching and logging any error so that state I/O can never crash a PypeIt reduction.

Generated by JXP and Claude.

Returns:

True if the write succeeded, False if it failed (in which case a warning is logged and the run continues).

Return type:

bool

scattlight: Optional[List[ScattLightCalibState]]

science: Optional[List[ScienceFrameState]]

science_entry(frame, det)

Return the science entry matching (frame, det), or None.

Generated by JXP and Claude.

Parameters:

• frame (str) – The exposure basename.

• det (int, tuple, list) – Detector or mosaic.

Returns:

ScienceFrameState or None.

slits: Optional[List[SlitEdgesState]]

tiltimg: Optional[List[TiltImgCalibState]]

tilts: Optional[List[TiltsState]]

update_calib(step, calib_id, det, key, value, slit=None)

Create or update the state entry for a single calibration step.

Looks up the entry matching (calib_id, det) for the given step; if none exists it is created. The requested key is then set to value (appended if the field is a list), either on the step entry or, when slit is given, on the per-slit/order sub-entry.

Generated by JXP and Claude.

Parameters:

• step (str) – Calibration step name (must be a key of calib_classes).

• calib_id (int) – Calibration group ID.

• det (int, tuple, list) – Detector (int) or detector mosaic (tuple/list of ints).

• key (str) – Name of the field to set on the step (or per-slit) entry.

• value – Value to assign. A list/tuple replaces the field with a flat list (e.g. input_files); a scalar appended to a field that already holds a list accumulates (e.g. flats types); otherwise the scalar is assigned.

• slit (int, optional) – Slit/order ID. If provided, key is set on the per-slit sub-entry rather than the step entry itself.

write()

Serialize the state to its JSON file (self.outfile).

Generated by JXP and Claude.

This may raise on I/O or serialization errors; use safe_write() from within a reduction so a failure here can never abort the run.

wv_calib: Optional[List[WvCalibState]]

class pypeit.state.run_state.ScattLightCalibState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['scattlight']

class pypeit.state.run_state.ScienceFrameState(**data)

Bases: BaseModel

State of one reduced science/standard exposure on one detector/mosaic.

Generated by JXP and Claude.

_abc_impl = <_abc._abc_data object>

bkg_id: Optional[int]

calib_id: int

comb_id: Optional[int]

det: Union[int, List[int]]

extract: ScienceStep

findobj: ScienceStep

frame: str

model_config: ClassVar[ConfigDict] = {}

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

nobj: Optional[int]

objects: Optional[List[ScienceObj]]

objtype: str

process: ScienceStep

raw_files: Optional[List[str]]

skysub: ScienceStep

slits: Optional[Dict[int, ScienceSlit]]

spec1d_file: Optional[str]

spec2d_file: Optional[str]

class pypeit.state.run_state.ScienceObj(**data)

Bases: BaseModel

One detected (and possibly extracted) object on a science exposure.

Generated by JXP and Claude.

_abc_impl = <_abc._abc_data object>

extracted: bool

fwhm: Optional[float]

model_config: ClassVar[ConfigDict] = {}

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

objid: int

s2n: Optional[float]

sign: Optional[int]

slitid: Optional[int]

snr_find: Optional[float]

spat_pixpos: Optional[float]

class pypeit.state.run_state.ScienceSlit(**data)

Bases: BaseModel

Per-slit science status (from the slit bitmask).

Generated by JXP and Claude.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}

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

nobj: Optional[int]

status: Literal['success', 'fail', 'skip', 'undone']

class pypeit.state.run_state.ScienceStep(**data)

Bases: BaseModel

Status of one macro-step (process/findobj/skysub/extract) of a science exposure.

Generated by JXP and Claude.

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}

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

status: Literal['undone', 'running', 'success', 'fail']

class pypeit.state.run_state.SlitEdges(**data)

Bases: BaseModel

_abc_impl = <_abc._abc_data object>

center: Optional[float]

model_config: ClassVar[ConfigDict] = {}

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

slitord_id: Optional[int]

status: Literal['success', 'fail', 'undone']

class pypeit.state.run_state.SlitEdgesState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

nslits: Optional[int]

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

slits: Optional[Dict[int, SlitEdges]]

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['slits']

class pypeit.state.run_state.TiltImgCalibState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['tiltimg']

class pypeit.state.run_state.TiltsSlit(**data)

Bases: BaseModel

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}

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

rms: Optional[float]

status: Literal['success', 'fail', 'undone']

class pypeit.state.run_state.TiltsState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

slits: Optional[Dict[int, TiltsSlit]]

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['tilts']

class pypeit.state.run_state.WvCalibSlit(**data)

Bases: BaseModel

_abc_impl = <_abc._abc_data object>

model_config: ClassVar[ConfigDict] = {}

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

rms: Optional[float]

status: Literal['success', 'fail', 'undone']

class pypeit.state.run_state.WvCalibState(**data)

Bases: BaseCalibState

_abc_impl = <_abc._abc_data object>

calib_id: int

det: int | List[int]

input_files: Optional[List[str]]

model_config: ClassVar[ConfigDict] = {}

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

output_file: Optional[str]

qa_files: Optional[List[str]]

required: bool

slits: Optional[Dict[int, WvCalibSlit]]

status: Literal['complete', 'fail', 'undone', 'running', 'success']

step: Literal['wv_calib']

pypeit.state.run_state.same_det(det1, det2)

Compare two detector identifiers for equality, treating a detector mosaic the same whether it is stored as a tuple or a list.

A single detector is an int; a detector mosaic is a tuple of ints (e.g. (1, 5)). Because pydantic coerces the stored det field to a list, a direct == comparison between a stored mosaic ([1, 5]) and an incoming mosaic ((1, 5)) is always False. This helper normalizes both sides before comparing.

Generated by JXP and Claude.

Parameters:

• det1 (int, list, tuple) – First detector identifier (single detector or mosaic).

• det2 (int, list, tuple) – Second detector identifier (single detector or mosaic).

Returns:

True if the two identifiers refer to the same detector or detector mosaic.

Return type:

bool