autods_pet.pipeline#

Programmatic pipeline for Deauville Score computation.

This module contains the DeauvillePipeline class and per-stage helper functions. It has no dependency on Rich or Typer, so it can be used from Python scripts, notebooks, or tests without a terminal.

autods_pet.pipeline.extract_new_targets_only(cfg, patient)[source]#

Extract only target ROIs and merge with cached reference values.

Loads the registered PET, reads cached reference values from the existing SUV_values.csv, extracts only target masks, and computes Deauville Scores. Falls back to a full extract_rois() if cached references are unavailable.

Parameters:

cfg (dict[str, Any]) – Configuration dict.
patient (PatientCase) – Patient case with resolved paths.

Returns:

Keys "scores" and "extract_results" (stats-only dict).

Return type:

dict[str, Any]

autods_pet.pipeline.generate_metadata_template(cfg, patient_ids)[source]#

Generate (or update) a metadata CSV template for patients with gaps.

For each patient, the JSON sidecar is checked. If any required field is missing or null, a row is added to the template with known values pre-filled and missing values left empty. An existing CSV is merged so that user-provided values are preserved.

Returns the path written, or None if every patient already has complete metadata.

Return type:

Path | None

Parameters:

cfg (dict[str, Any])
patient_ids (list[str])

autods_pet.pipeline.convert_images(cfg, patient)[source]#

Convert input images (DICOM, NRRD, or NIfTI) to the standard NIfTI layout.

Parameters:

cfg (dict[str, Any]) – Configuration dict (from autods_pet.config.load_config()).
patient (PatientCase) – Patient case with resolved paths.

Returns:

Keys include ct_path, pet_path, skipped (bool), and optionally metadata (PET DICOM tags).

Return type:

dict[str, Any]

Raises:

FileNotFoundError – If the expected CT or PET input files are missing.

autods_pet.pipeline.normalize_pet(cfg, patient)[source]#

Compute SUV body-weight from a raw PET image.

Parameters:

cfg (dict[str, Any]) – Configuration dict.
patient (PatientCase) – Patient case with resolved paths.

Returns:

Keys: pet_suv_path (Path), skipped (bool).

Return type:

dict[str, Any]

Raises:

ValueError – If no PET metadata is available for the patient.

autods_pet.pipeline.register_pet(cfg, patient)[source]#

Rigidly register PET SUV image onto the CT grid.

Parameters:

cfg (dict[str, Any]) – Configuration dict.
patient (PatientCase) – Patient case with resolved paths.

Returns:

Keys: pet_registered_path (Path), skipped (bool).

Return type:

dict[str, Any]

autods_pet.pipeline.segment_ct(cfg, patient)[source]#

Run TotalSegmentator on the CT image.

Parameters:

cfg (dict[str, Any]) – Configuration dict (reads totalsegmentator.fast and license key).
patient (PatientCase) – Patient case with resolved paths.

Returns:

Keys: seg_multilabel (Path), vb_available (bool), and optionally vert_body_seg.

Return type:

dict[str, Any]

Raises:

RuntimeError – If the main total segmentation task fails.

autods_pet.pipeline.extract_rois(cfg, patient, seg_result=None)[source]#

Extract ROI statistics from PET.

Returns a dict mapping ROI display names to ROIResult instances, plus a "_roi_statuses" key with status tuples.

Return type:

dict[str, Any]

Parameters:

cfg (dict[str, Any])
patient (PatientCase)
seg_result (dict[str, Any] | None)

autods_pet.pipeline.score_deauville(cfg, extract_results)[source]#

Assign Deauville Scores from extracted ROI statistics.

Parameters:

cfg (dict[str, Any]) – Configuration dict. Reads cfg["deauville"]["liver_multiplier"] (default 2.0) for the DS 4 / DS 5 cutoff.
extract_results (dict[str, Any]) – Output of extract_rois(), mapping ROI display names to ROIResult instances.

Returns:

Mapping of score keys (e.g. "FL_DS", "BM_DS", "BLR") to Deauville Scores (int, 1–5) or the Brain-to-Liver Ratio (float). Empty if references are unavailable.

Return type:

dict[str, int | float]

autods_pet.pipeline.write_patient_suv_csv(extract_results, path)[source]#

Write per-patient SUV values to a CSV file (merge mode).

Reads the existing file (if any), merges new entries (update existing, add new), and writes back. One row per (ROI, stat) pair. Columns: ROI, Statistic, Value.

Return type:

None

Parameters:

extract_results (dict[str, Any])
path (Path)

autods_pet.pipeline.write_patient_deauville_csv(scores, path)[source]#

Write per-patient Deauville scores to a CSV file (merge mode).

Reads the existing file (if any), merges new scores (update existing, add new), and writes back. One row per target. Columns: Target, DeauvilleScore.

Return type:

None

Parameters:

scores (dict[str, int | float])
path (Path)

autods_pet.pipeline.save_batch_csv(new_df, path, output_format='csv', int_columns=None)[source]#

Save batch results, merging with an existing file if present.

Existing patients are updated (replaced), new patients are appended.

Parameters:

new_df (Any) – New batch results (must have a patient_id column).
path (Path) – Destination file path.
output_format (str (default: 'csv')) – "csv" or "xlsx".
int_columns (list[str] | None (default: None)) – Columns to cast to nullable Int64 after merge/concat.

Return type:

Path

class autods_pet.pipeline.DeauvillePipeline(cfg, force=False)[source]#

Bases: object

Orchestrates the full DS computation pipeline.

Parameters:

cfg (dict[str, Any]) – Configuration dict (from autods_pet.config.load_config()).
force (bool (default: False)) – When True, re-run every stage even if outputs already exist (sets cfg["pipeline"]["force"] = True). Defaults to False, which skips stages whose output files are already on disk.

Examples

>>> from autods_pet import DeauvillePipeline, load_config
>>> cfg = load_config("config.ini")
>>> pipeline = DeauvillePipeline(cfg)
>>> result = pipeline.run("PATIENT_001")
>>> print(result.scores)

run(patient_id)[source]#

Run the full pipeline for one patient.

Returns:: Contains patient_id, scores, rois, and error.
Return type:: DeauvilleResult
Parameters:: patient_id (str)

convert(patient)[source]#

Convert input images to NIfTI format.

Return type:: dict[str, Any]
Parameters:: patient (PatientCase)

normalize(patient)[source]#

Compute SUVbw from raw PET.

Return type:: dict[str, Any]
Parameters:: patient (PatientCase)

register(patient)[source]#

Rigidly register PET SUV to CT grid.

Return type:: dict[str, Any]
Parameters:: patient (PatientCase)

segment(patient)[source]#

Run TotalSegmentator on CT.

Return type:: dict[str, Any]
Parameters:: patient (PatientCase)

extract(patient, seg_result=None)[source]#

Extract ROI statistics from registered PET.

Return type:

dict[str, Any]

Parameters:

patient (PatientCase)
seg_result (dict[str, Any] | None)

score(extract_results)[source]#

Assign Deauville Scores from extracted ROI stats.

Return type:: dict[str, int | float]
Parameters:: extract_results (dict[str, Any])

update_targets(patient_id)[source]#

Re-extract only target ROIs for a patient, using cached references.

Parameters:: patient_id (str) – Patient identifier.
Returns:: Keys "scores" and "extract_results".
Return type:: dict[str, Any]

run_batch(patients)[source]#

Run the full pipeline for multiple patients.

Errors are captured per-patient - the batch never stops early.

Parameters:: patients (list[str] | str | Path) – Either a list of patient ID strings, or a path to a text file containing one patient ID per line (# comments allowed).
Returns:: One result per patient (check .error for failures).
Return type:: list[DeauvilleResult]

static to_dataframe(results)[source]#

Convert a list of DeauvilleResult into a summary DataFrame.

Columns include patient_id, error, per-ROI statistics (e.g. Liver_median), and DS scores (e.g. DS_FL_DS).

Parameters:: results (list[DeauvilleResult]) – Results from run() or run_batch().
Return type:: DataFrame