Processing Context#

The ProcessingContext dataclass carries all per-subject state (patient ID, config, paths, intermediate results) through the pipeline.

Processing context for the thesis framework.

Provides a context object that carries information about the current processing state, patient data, paths, and configuration through the pipeline.

class thesis.core.context.ProcessingContext[source]#

Bases: object

Context object for medical imaging processing pipelines.

Carries all relevant information about a processing run, including patient ID, data paths, configuration, and intermediate results.

Variables:

  • patient_id – Unique patient identifier

  • config – Configuration for this processing run

  • data_dir – Shared assets base (config paths.assets_dir): templates, atlases, ROIs. Not a parent of input_dir/output_dir.

  • input_dir – Per-patient input data directory (inputs_dir/<patient_id>)

  • output_dir – Per-patient output directory (output_dir/<patient_id>)

  • working_dir – Temporary working/scratch directory

  • metadata – Additional metadata

  • results – Dictionary to store intermediate results

Example

>>> ctx = ProcessingContext(
...     patient_id="DTI_LDF001",
...     config=my_config,
...     data_dir=Path("./data")
... )
>>> ctx.add_result("registration", transform_matrix)
>>> transform = ctx.get_result("registration")
Parameters:
patient_id: str#
config: PipelineConfig#
data_dir: Path#
input_dir: Path | None = None#
output_dir: Path | None = None#
working_dir: Path | None = None#
metadata: Dict[str, Any]#
results: Dict[str, Any]#
add_result(key, value)[source]#

Store a processing result.

Parameters:

  • key (str) – Result identifier

  • value (Any) – Result data

Return type:

None

Example

>>> ctx.add_result("brain_mask", mask_path)
get_result(key, default=None)[source]#

Retrieve a processing result.

Parameters:

  • key (str) – Result identifier

  • default (Any) – Default value if key not found

Return type:

Any

Returns:

Result data or default

Example

>>> mask_path = ctx.get_result("brain_mask")
has_result(key)[source]#

Check if a result exists.

Parameters:

key (str) – Result identifier

Return type:

bool

Returns:

True if result exists, False otherwise

add_metadata(key, value)[source]#

Add metadata to the context.

Parameters:

  • key (str) – Metadata key

  • value (Any) – Metadata value

Return type:

None

Example

>>> ctx.add_metadata("acquisition_date", "2025-01-15")
get_metadata(key, default=None)[source]#

Retrieve metadata.

Parameters:

  • key (str) – Metadata key

  • default (Any) – Default value if key not found

Return type:

Any

Returns:

Metadata value or default

get_input_path(filename)[source]#

Get full path for an input file.

Parameters:

filename (str) – Input filename

Return type:

Path

Returns:

Full path to input file

Raises:

ValueError – If a relative filename escapes input_dir.

Example

>>> t1_path = ctx.get_input_path("DTI_LDF001_T1.nii.gz")
get_output_path(filename, subdir=None)[source]#

Get full path for an output file.

Parameters:

  • filename (str) – Output filename

  • subdir (Optional[str]) – Optional subdirectory within output_dir

Return type:

Path

Returns:

Full path to output file

Raises:

ValueError – If a relative path escapes output_dir.

Example

>>> reg_path = ctx.get_output_path("T1_registered.nii.gz", "registration")
get_working_path(filename)[source]#

Get full path for a temporary working file.

Parameters:

filename (str) – Working filename

Return type:

Path

Returns:

Full path to working file

Example

>>> temp_path = ctx.get_working_path("temp_mask.nii.gz")
cleanup_working_dir()[source]#

Remove all files from the working directory.

Use with caution - this deletes temporary files.

Return type:

None

list_input_files(pattern='*')[source]#

List files in the input directory.

Parameters:

pattern (str) – Glob pattern for matching files

Return type:

list[Path]

Returns:

List of matching file paths

Example

>>> nii_files = ctx.list_input_files("*.nii.gz")
list_output_files(pattern='*')[source]#

List files in the output directory.

Parameters:

pattern (str) – Glob pattern for matching files

Return type:

list[Path]

Returns:

List of matching file paths

to_dict()[source]#

Convert context to dictionary.

Return type:

Dict[str, Any]

Returns:

Dictionary representation of context

__init__(patient_id, config, data_dir, input_dir=None, output_dir=None, working_dir=None, metadata=<factory>, results=<factory>)#
Parameters:
Return type:

None

thesis.core.context.create_context(patient_id, config, data_dir=None, **kwargs)[source]#

Factory function to create a processing context.

Parameters:

  • patient_id (str) – Patient identifier

  • config (PipelineConfig) – Configuration object

  • data_dir (Optional[Path]) – Shared assets base (reads config paths.assets_dir if not provided)

  • **kwargs – Additional arguments passed to ProcessingContext

Return type:

ProcessingContext

Returns:

Initialized ProcessingContext

Example

>>> config = load_config("default")
>>> ctx = create_context("DTI_LDF001", config)