API Reference¶

Auto-generated from NumPy-format docstrings in the pipeline/ package.

Orchestrator¶

pipeline.orchestrator ¶

VIPER Pipeline Orchestrator.

This script orchestrates the end-to-end immunization notice generation pipeline. It executes each step in sequence, handles errors, and provides detailed timing and progress information.

Error Handling Philosophy:

The pipeline distinguishes between critical and optional steps:

Critical Steps (Notice generation, Compilation, PDF validation) implement fail-fast:
- Any error halts the pipeline immediately
- No partial output; users get deterministic results
- Pipeline exits with code 1; user must investigate and retry
Optional Steps (QR codes, Encryption, Bundling) implement per-item recovery:
- Individual item failures (PDF, client, bundle) are logged and skipped
- Remaining items continue processing
- Pipeline completes successfully even if some items failed
- Users are shown summary of successes, skipped, and failed items
Infrastructure Errors (missing files, config errors) always fail-fast:
- Caught and raised immediately; no recovery attempts
- Prevents confusing partial output caused by misconfiguration
- Pipeline exits with code 1

Exit Codes:

0: Pipeline completed successfully
1: Pipeline failed (critical step error or infrastructure error)
2: User cancelled (output preparation step)

parse_args ¶

parse_args()

Parse command-line arguments.

Source code in pipeline/orchestrator.py

def parse_args() -> argparse.Namespace:
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Run the VIPER immunization notice generation pipeline",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s students.xlsx en
  %(prog)s students.xlsx fr
        """,
    )

    parser.add_argument(
        "input_file",
        type=str,
        help="Name of the input file (e.g., students.xlsx)",
    )
    parser.add_argument(
        "language",
        choices=sorted(Language.all_codes()),
        help=f"Language for output ({', '.join(sorted(Language.all_codes()))})",
    )
    parser.add_argument(
        "--input",
        type=Path,
        default=DEFAULT_INPUT_DIR,
        dest="input_dir",
        help=f"Input directory (default: {DEFAULT_INPUT_DIR})",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=DEFAULT_OUTPUT_DIR,
        dest="output_dir",
        help=f"Output directory (default: {DEFAULT_OUTPUT_DIR})",
    )
    parser.add_argument(
        "--config",
        type=Path,
        default=DEFAULT_CONFIG_DIR,
        dest="config_dir",
        help=f"Config directory (default: {DEFAULT_CONFIG_DIR})",
    )
    parser.add_argument(
        "--template",
        type=str,
        default=None,
        dest="template_dir",
        help="PHU template name within phu_templates/ (e.g., 'wdgph'). "
        "If not specified, pipeline is run in testing mode, defaulting to the templates/ directory.",
    )

    return parser.parse_args()

validate_args ¶

validate_args(args)

Validate command-line arguments and raise errors if invalid.

Source code in pipeline/orchestrator.py

def validate_args(args: argparse.Namespace) -> None:
    """Validate command-line arguments and raise errors if invalid."""
    if args.input_file and not (args.input_dir / args.input_file).exists():
        raise FileNotFoundError(
            f"Input file not found: {args.input_dir / args.input_file}"
        )

    # Resolve template directory
    if args.template_dir is None:
        # No custom template specified; use default
        args.template_dir = DEFAULT_TEMPLATES_DIR
    else:
        # Custom PHU template specified; resolve within phu_templates/
        # Validate no path separators (prevent nested directories)
        if "/" in args.template_dir or "\\" in args.template_dir:
            raise ValueError(
                f"Template name cannot contain path separators: {args.template_dir}\n"
                f"Expected a simple name like 'wdgph' or 'my_phu', not a path."
            )

        phu_template_path = DEFAULT_PHU_TEMPLATES_DIR / args.template_dir
        if not phu_template_path.exists():
            raise FileNotFoundError(
                f"PHU template directory not found: {phu_template_path}\n"
                f"Expected location: phu_templates/{args.template_dir}\n"
                f"Ensure the directory exists and contains required template files."
            )
        if not phu_template_path.is_dir():
            raise NotADirectoryError(
                f"PHU template path is not a directory: {phu_template_path}"
            )
        # Update args.template_dir to resolved Path
        args.template_dir = phu_template_path

    # Validate template directory contents
    if not args.template_dir.is_dir():
        raise NotADirectoryError(
            f"Template path is not a directory: {args.template_dir}"
        )

main ¶

main()

Run the pipeline orchestrator.

Source code in pipeline/orchestrator.py

def main() -> int:
    """Run the pipeline orchestrator."""
    try:
        args = parse_args()
        validate_args(args)
    except (ValueError, SystemExit) as exc:
        if isinstance(exc, ValueError):
            print(f"Error: {exc}", file=sys.stderr)
            return 1
        raise

    # Setup paths and load configuration
    output_dir = args.output_dir.resolve()
    config_dir = args.config_dir.resolve()
    log_dir = output_dir / "logs"
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")

    # Load configuration
    try:
        config = load_config(config_dir / "parameters.yaml")
    except FileNotFoundError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    # Extract config settings
    encryption_enabled = config.get("encryption", {}).get("enabled", False)

    print_header(args.input_file)

    total_start = time.time()
    step_times = []
    total_clients = 0

    try:
        # Step 1: Prepare output directory
        step_start = time.time()
        if not run_step_1_prepare_output(output_dir, log_dir, config_dir):
            return 2  # User cancelled
        step_duration = time.time() - step_start
        step_times.append(("Output Preparation", step_duration))
        print_step_complete(1, "Output directory prepared", step_duration)

        # Step 2: Preprocessing
        step_start = time.time()
        total_clients = run_step_2_preprocess(
            args.input_dir,
            args.input_file,
            output_dir,
            args.language,
            run_id,
            config_dir,
        )
        step_duration = time.time() - step_start
        step_times.append(("Preprocessing", step_duration))
        print_step_complete(2, "Preprocessing", step_duration)

        # Step 3: Generating QR Codes (optional)
        step_start = time.time()
        qr_count = run_step_3_generate_qr_codes(
            output_dir,
            run_id,
            config_dir,
        )
        step_duration = time.time() - step_start
        if qr_count > 0:
            step_times.append(("QR Code Generation", step_duration))
            print_step_complete(3, "QR code generation", step_duration)
        else:
            print("QR code generation skipped (disabled or no clients).")

        # Step 4: Generating Notices
        step_start = time.time()
        run_step_4_generate_notices(
            output_dir,
            run_id,
            args.template_dir,
            config_dir,
        )
        step_duration = time.time() - step_start
        step_times.append(("Template Generation", step_duration))
        print_step_complete(4, "Template generation", step_duration)

        # Step 5: Compiling Notices
        step_start = time.time()
        run_step_5_compile_notices(
            output_dir,
            config_dir,
            args.template_dir,
        )
        step_duration = time.time() - step_start
        step_times.append(("Template Compilation", step_duration))
        print_step_complete(5, "Compilation", step_duration)

        # Step 6: Validating PDFs
        step_start = time.time()
        run_step_6_validate_pdfs(output_dir, args.language, run_id, config_dir)
        step_duration = time.time() - step_start
        step_times.append(("PDF Validation", step_duration))
        print_step_complete(6, "PDF validation", step_duration)

        # Step 7: Encrypting PDFs (optional)
        if encryption_enabled:
            step_start = time.time()
            run_step_7_encrypt_pdfs(output_dir, args.language, run_id)
            step_duration = time.time() - step_start
            step_times.append(("PDF Encryption", step_duration))
            print_step_complete(7, "Encryption", step_duration)

        # Step 8: Bundling PDFs (optional, independent of encryption)
        bundling_config = config.get("bundling", {})
        bundle_size = bundling_config.get("bundle_size", 0)

        if bundle_size > 0:
            step_start = time.time()
            run_step_8_bundle_pdfs(
                output_dir,
                args.language,
                run_id,
                config_dir,
            )
            step_duration = time.time() - step_start
            step_times.append(("PDF Bundling", step_duration))
            print_step_complete(8, "Bundling", step_duration)
        else:
            print_step(8, "Bundling")
            print("Bundling skipped (bundle_size set to 0).")

        # Step 9: Cleanup
        run_step_9_cleanup(output_dir, config_dir)

        # Print summary
        total_duration = time.time() - total_start

        print_summary(
            step_times,
            total_duration,
            total_clients,
        )

        return 0

    except Exception as exc:
        print(f"\n❌ Pipeline failed: {exc}", file=sys.stderr)
        traceback.print_exc()
        return 1

Configuration¶

pipeline.config_loader ¶

Configuration loading utilities for the immunization pipeline.

Provides a centralized way to load and validate the parameters.yaml configuration file across all pipeline scripts.

load_config ¶

load_config(config_path=None)

Load and parse the parameters.yaml configuration file.

Automatically validates the configuration after loading. Raises clear exceptions if validation fails, enabling fail-fast behavior for infrastructure errors.

Parameters:

Name	Type	Description	Default
`config_path`	`Path`	Path to the configuration file. If not provided, uses the default location (config/parameters.yaml in the project root).	`None`

Returns:

Type	Description
`Dict[str, Any]`	Parsed and validated YAML configuration as a nested dictionary.

Raises:

Type	Description
`FileNotFoundError`	If the configuration file does not exist.
`YAMLError`	If the configuration file is invalid YAML.
`ValueError`	If the configuration fails validation (see validate_config).

Source code in pipeline/config_loader.py

def load_config(config_path: Optional[Path] = None) -> Dict[str, Any]:
    """Load and parse the parameters.yaml configuration file.

    Automatically validates the configuration after loading. Raises
    clear exceptions if validation fails, enabling fail-fast behavior
    for infrastructure errors.

    Parameters
    ----------
    config_path : Path, optional
        Path to the configuration file. If not provided, uses the default
        location (config/parameters.yaml in the project root).

    Returns
    -------
    Dict[str, Any]
        Parsed and validated YAML configuration as a nested dictionary.

    Raises
    ------
    FileNotFoundError
        If the configuration file does not exist.
    yaml.YAMLError
        If the configuration file is invalid YAML.
    ValueError
        If the configuration fails validation (see validate_config).
    """
    if config_path is None:
        config_path = DEFAULT_CONFIG_PATH

    config_path = Path(config_path)

    if not config_path.exists():
        raise FileNotFoundError(f"Configuration file not found: {config_path}")

    with config_path.open("r", encoding="utf-8") as f:
        config = yaml.safe_load(f) or {}

    validate_config(config)
    return config

validate_template_fields ¶

validate_template_fields(template, context_name, config_key)

Validate that all placeholders in a template are supported fields.

Checks template string placeholders against the centralized whitelist of allowed template fields (TemplateField enum). Raises ValueError if any unsupported placeholders are found.

This validation catches configuration errors early (at config load time) instead of failing silently during QR generation or PDF encryption.

Parameters:

Name	Type	Description	Default
`template`	`str`	Template string with placeholders (e.g., "{client_id}_{invalid_field}")	required
`context_name`	`str`	Human-readable name for error messages (e.g., "QR code payload", "encryption password")	required
`config_key`	`str`	Configuration key path for error messages (e.g., "qr.payload_template")	required

Raises:

Type	Description
`ValueError`	If template contains unsupported placeholder fields with detailed error message listing invalid fields and available options.

Examples:

>>> validate_template_fields(
...     "{client_id}_{invalid_field}",
...     "QR code payload",
...     "qr.payload_template"
... )
ValueError: QR code payload template contains unsupported placeholder(s): ['invalid_field']
Config key: qr.payload_template
Supported fields: client_id, first_name, last_name, ...

Source code in pipeline/config_loader.py

def validate_template_fields(
    template: str,
    context_name: str,
    config_key: str,
) -> None:
    """Validate that all placeholders in a template are supported fields.

    Checks template string placeholders against the centralized whitelist of
    allowed template fields (TemplateField enum). Raises ValueError if any
    unsupported placeholders are found.

    This validation catches configuration errors early (at config load time)
    instead of failing silently during QR generation or PDF encryption.

    Parameters
    ----------
    template : str
        Template string with placeholders (e.g., "{client_id}_{invalid_field}")
    context_name : str
        Human-readable name for error messages (e.g., "QR code payload", "encryption password")
    config_key : str
        Configuration key path for error messages (e.g., "qr.payload_template")

    Raises
    ------
    ValueError
        If template contains unsupported placeholder fields with detailed
        error message listing invalid fields and available options.

    Examples
    --------
    >>> validate_template_fields(
    ...     "{client_id}_{invalid_field}",
    ...     "QR code payload",
    ...     "qr.payload_template"
    ... )
    ValueError: QR code payload template contains unsupported placeholder(s): ['invalid_field']
    Config key: qr.payload_template
    Supported fields: client_id, first_name, last_name, ...
    """
    allowed_fields = TemplateField.all_values()

    try:
        placeholders = extract_template_fields(template)
    except ValueError as exc:
        raise ValueError(
            f"{context_name} template has invalid format syntax: {exc}\n"
            f"Config key: {config_key}"
        ) from exc

    unsupported = placeholders - allowed_fields

    if unsupported:
        raise ValueError(
            f"{context_name} template contains unsupported placeholder(s): {sorted(unsupported)}\n"
            f"Config key: {config_key}\n"
            f"Supported fields: {', '.join(sorted(allowed_fields))}"
        )

validate_config ¶

validate_config(config)

Validate the entire configuration for consistency and required values.

Validates all conditional and required configuration keys across the entire config. Raises clear exceptions if validation fails, allowing the pipeline to fail-fast with actionable error messages.

Parameters:

Name	Type	Description	Default
`config`	`Dict[str, Any]`	Configuration dictionary (result of load_config).	required

Raises:

Type	Description
`ValueError`	If required configuration is missing or invalid.

Notes

Validation checks:

QR Generation: If qr.enabled=true, requires qr.payload_template (non-empty string)
Typst Compilation: If typst.bin is set, must be a string
PDF Bundling: If bundle_size > 0, must be positive integer; group_by must be valid enum
Encryption: If encryption.enabled=true, requires password.template
Cleanup: If delete_unencrypted_pdfs is set, must be boolean

Validation philosophy: - Infrastructure errors (missing config) raise immediately (fail-fast) - All error messages are clear and actionable - Config is validated once at load time, not per-step

Source code in pipeline/config_loader.py

def validate_config(config: Dict[str, Any]) -> None:
    """Validate the entire configuration for consistency and required values.

    Validates all conditional and required configuration keys across the
    entire config. Raises clear exceptions if validation fails, allowing
    the pipeline to fail-fast with actionable error messages.

    Parameters
    ----------
    config : Dict[str, Any]
        Configuration dictionary (result of load_config).

    Raises
    ------
    ValueError
        If required configuration is missing or invalid.

    Notes
    -----
    **Validation checks:**

    - **QR Generation:** If qr.enabled=true, requires qr.payload_template (non-empty string)
    - **Typst Compilation:** If typst.bin is set, must be a string
    - **PDF Bundling:** If bundle_size > 0, must be positive integer; group_by must be valid enum
    - **Encryption:** If encryption.enabled=true, requires password.template
    - **Cleanup:** If delete_unencrypted_pdfs is set, must be boolean

    **Validation philosophy:**
    - Infrastructure errors (missing config) raise immediately (fail-fast)
    - All error messages are clear and actionable
    - Config is validated once at load time, not per-step
    """
    # Validate QR config
    qr_config = config.get("qr", {})
    qr_enabled = qr_config.get("enabled", True)

    if qr_enabled:
        payload_template = qr_config.get("payload_template")
        if not payload_template:
            raise ValueError(
                "QR code generation is enabled but qr.payload_template is not specified. "
                "Please define qr.payload_template in config/parameters.yaml "
                "or set qr.enabled to false."
            )

        if not isinstance(payload_template, str):
            raise ValueError(
                f"qr.payload_template must be a string, got {type(payload_template).__name__}"
            )

        # Validate template placeholders against allowed fields
        validate_template_fields(
            payload_template,
            "QR code payload",
            "qr.payload_template",
        )

    # Validate Typst config
    typst_config = config.get("typst", {})
    typst_bin = typst_config.get("bin", "typst")
    if not isinstance(typst_bin, str):
        raise ValueError(f"typst.bin must be a string, got {type(typst_bin).__name__}")

    # Validate Bundling config
    bundling_config = config.get("bundling", {})
    bundle_size = bundling_config.get("bundle_size", 0)

    # First validate type before comparing values
    if bundle_size != 0:  # Only validate if bundle_size is explicitly set
        if not isinstance(bundle_size, int):
            raise ValueError(
                f"bundling.bundle_size must be an integer, got {type(bundle_size).__name__}"
            )
        if bundle_size <= 0:
            raise ValueError(
                f"bundling.bundle_size must be positive, got {bundle_size}"
            )

        # Validate group_by strategy
        group_by = bundling_config.get("group_by")
        from .enums import BundleStrategy

        try:
            if group_by is not None:
                BundleStrategy.from_string(group_by)
        except ValueError as exc:
            raise ValueError(f"Invalid bundling.group_by strategy: {exc}") from exc

    # Validate Encryption config
    encryption_config = config.get("encryption", {})
    encryption_enabled = encryption_config.get("enabled", False)

    if encryption_enabled:
        password_config = encryption_config.get("password", {})
        password_template = password_config.get("template")
        if not password_template:
            raise ValueError(
                "Encryption is enabled but encryption.password.template is not specified. "
                "Please define encryption.password.template in config/parameters.yaml "
                "or set encryption.enabled to false."
            )

        if not isinstance(password_template, str):
            raise ValueError(
                f"encryption.password.template must be a string, "
                f"got {type(password_template).__name__}"
            )

        # Validate template placeholders against allowed fields
        validate_template_fields(
            password_template,
            "Encryption password",
            "encryption.password.template",
        )

    # Validate Cleanup config
    cleanup_config = config.get("cleanup", {})
    delete_unencrypted = cleanup_config.get("delete_unencrypted_pdfs", False)
    if not isinstance(delete_unencrypted, bool):
        raise ValueError(
            f"cleanup.delete_unencrypted_pdfs must be a boolean, "
            f"got {type(delete_unencrypted).__name__}"
        )

Data Models¶

pipeline.data_models ¶

Unified data models for the immunization pipeline.

This module provides all core dataclasses used throughout the pipeline, ensuring consistency and type safety across processing steps.

ClientRecord `dataclass` ¶

Unified client record across all pipeline steps.

This dataclass represents a single client (student) record passed through the entire pipeline. It contains all necessary information for: - Generating personalized notices - Creating QR codes - Encrypting PDFs - Batching outputs

Fields

sequence : str Zero-padded sequence number for the client (e.g., '00001').

client_id : str Unique client identifier.

language : str ISO 639-1 language code ('en' or 'fr'). Must be a valid Language enum value (see pipeline.enums.Language). Validated using Language.from_string() at entry points (CLI, configuration loading, preprocessing). All functions assume this field contains a valid language code; invalid codes should be caught before ClientRecord instantiation.

person : Dict[str, Any]

- Person details:
    - full_name: Combined first and last name
    - first_name: Given name (optional)
    - last_name: Family name (optional)
    - date_of_birth: Display format (e.g., "Jan 8, 2025")
    - date_of_birth_iso: ISO format (YYYY-MM-DD)
    - date_of_birth_display: Localized display format
    - age: Calculated age in years
    - over_16: Boolean flag for age >= 16

school : Dict[str, Any] School information: name, id, code, type.

board : Dict[str, Any] School board information: name, id, code.

contact : Dict[str, Any] Contact address: street, city, province, postal_code.

vaccines_due : Optional[str] Comma-separated string of vaccines due (display format).

vaccines_due_list : Optional[List[str]] List of vaccine names/codes due.

received : Optional[Sequence[Dict[str, object]]] List of vaccine records already received (structured data).

metadata : Dict[str, object] Custom pipeline metadata (warnings, flags, etc.).

qr : Optional[Dict[str, Any]]

- QR code information (if generated):
    - payload: URL encoded in QR code (same string that becomes QR PNG data)
    - filename: PNG filename (e.g., "qr_code_00001_1234567890.png")
    - path: Relative path to PNG file

Source code in pipeline/data_models.py

@dataclass(frozen=True)
class ClientRecord:
    """Unified client record across all pipeline steps.

    This dataclass represents a single client (student) record passed through
    the entire pipeline. It contains all necessary information for:
    - Generating personalized notices
    - Creating QR codes
    - Encrypting PDFs
    - Batching outputs

    Fields
    ------
    sequence : str
        Zero-padded sequence number for the client (e.g., '00001').

    client_id : str
        Unique client identifier.

    language : str
        ISO 639-1 language code ('en' or 'fr'). Must be a valid Language enum value
        (see pipeline.enums.Language). Validated using Language.from_string() at entry
        points (CLI, configuration loading, preprocessing). All functions assume this
        field contains a valid language code; invalid codes should be caught before
        ClientRecord instantiation.

    person : Dict[str, Any]

        - Person details:
            - full_name: Combined first and last name
            - first_name: Given name (optional)
            - last_name: Family name (optional)
            - date_of_birth: Display format (e.g., "Jan 8, 2025")
            - date_of_birth_iso: ISO format (YYYY-MM-DD)
            - date_of_birth_display: Localized display format
            - age: Calculated age in years
            - over_16: Boolean flag for age >= 16

    school : Dict[str, Any]
        School information: name, id, code, type.

    board : Dict[str, Any]
        School board information: name, id, code.

    contact : Dict[str, Any]
        Contact address: street, city, province, postal_code.

    vaccines_due : Optional[str]
        Comma-separated string of vaccines due (display format).

    vaccines_due_list : Optional[List[str]]
        List of vaccine names/codes due.

    received : Optional[Sequence[Dict[str, object]]]
        List of vaccine records already received (structured data).

    metadata : Dict[str, object]
        Custom pipeline metadata (warnings, flags, etc.).

    qr : Optional[Dict[str, Any]]

        - QR code information (if generated):
            - payload: URL encoded in QR code (same string that becomes QR PNG data)
            - filename: PNG filename (e.g., "qr_code_00001_1234567890.png")
            - path: Relative path to PNG file
    """

    sequence: str
    client_id: str
    language: str
    person: Dict[str, Any]
    school: Dict[str, Any]
    board: Dict[str, Any]
    contact: Dict[str, Any]
    vaccines_due: Optional[str]
    vaccines_due_list: Optional[List[str]]
    received: Optional[Sequence[Dict[str, object]]]
    metadata: Dict[str, object]
    qr: Optional[Dict[str, Any]] = None

PreprocessResult `dataclass` ¶

Result of preprocessing step.

The output of Step 2 (preprocessing) that contains normalized client data and any warnings generated during processing.

Parameters:

Name	Type	Description	Default
`clients`	`List[ClientRecord]`	Processed and validated client records.	required
`warnings`	`List[str]`	Non-fatal warnings encountered during preprocessing (e.g., missing optional fields, unrecognized vaccine codes).	required

Source code in pipeline/data_models.py

@dataclass(frozen=True)
class PreprocessResult:
    """Result of preprocessing step.

    The output of Step 2 (preprocessing) that contains normalized client data
    and any warnings generated during processing.

    Parameters
    ----------
    clients : List[ClientRecord]
        Processed and validated client records.
    warnings : List[str]
        Non-fatal warnings encountered during preprocessing (e.g., missing
        optional fields, unrecognized vaccine codes).
    """

    clients: List[ClientRecord]
    warnings: List[str]

ArtifactPayload `dataclass` ¶

Preprocessed artifact with metadata.

The JSON artifact written by Step 2 (preprocessing) and read by downstream steps. Contains all normalized client data and provenance information.

Parameters:

Name	Type	Description	Default
`run_id`	`str`	Unique pipeline run identifier (timestamp-based).	required
`language`	`str`	ISO 639-1 language code ('en' or 'fr'). Must be a valid Language enum value (see pipeline.enums.Language). All clients in the artifact must have language codes that match this field; validation ensures consistency across all notices generated in a single run.	required
`clients`	`List[ClientRecord]`	All processed client records.	required
`warnings`	`List[str]`	All preprocessing warnings.	required
`created_at`	`str`	ISO 8601 timestamp when artifact was created.	required
`input_file`	`Optional[str]`	Name of the input file processed (for audit trail).	`None`
`total_clients`	`int`	Total number of clients in artifact (convenience field).	`0`

Source code in pipeline/data_models.py

@dataclass(frozen=True)
class ArtifactPayload:
    """Preprocessed artifact with metadata.

    The JSON artifact written by Step 2 (preprocessing) and read by downstream
    steps. Contains all normalized client data and provenance information.

    Parameters
    ----------
    run_id : str
        Unique pipeline run identifier (timestamp-based).
    language : str
        ISO 639-1 language code ('en' or 'fr'). Must be a valid Language enum value
        (see pipeline.enums.Language). All clients in the artifact must have language
        codes that match this field; validation ensures consistency across all
        notices generated in a single run.
    clients : List[ClientRecord]
        All processed client records.
    warnings : List[str]
        All preprocessing warnings.
    created_at : str
        ISO 8601 timestamp when artifact was created.
    input_file : Optional[str]
        Name of the input file processed (for audit trail).
    total_clients : int
        Total number of clients in artifact (convenience field).
    """

    run_id: str
    language: str
    clients: List[ClientRecord]
    warnings: List[str]
    created_at: str
    input_file: Optional[str] = None
    total_clients: int = 0

PdfRecord `dataclass` ¶

Compiled PDF with client metadata.

Represents a single generated PDF notice with its associated client data and page count. Used during batching (Step 8) to group PDFs and generate manifests.

Parameters:

Name	Type	Description	Default
`sequence`	`str`	Zero-padded sequence number matching the PDF filename.	required
`client_id`	`str`	Client identifier matching the PDF filename.	required
`pdf_path`	`Path`	Absolute path to the generated PDF file.	required
`page_count`	`int`	Number of pages in the PDF (usually 2 for immunization notices).	required
`client`	`Dict[str, Any]`	Full client data dict for manifest generation and batching.	required

Source code in pipeline/data_models.py

@dataclass(frozen=True)
class PdfRecord:
    """Compiled PDF with client metadata.

    Represents a single generated PDF notice with its associated client
    data and page count. Used during batching (Step 8) to group PDFs
    and generate manifests.

    Parameters
    ----------
    sequence : str
        Zero-padded sequence number matching the PDF filename.
    client_id : str
        Client identifier matching the PDF filename.
    pdf_path : Path
        Absolute path to the generated PDF file.
    page_count : int
        Number of pages in the PDF (usually 2 for immunization notices).
    client : Dict[str, Any]
        Full client data dict for manifest generation and batching.
    """

    sequence: str
    client_id: str
    pdf_path: Path
    page_count: int
    client: Dict[str, Any]

Enumerations¶

pipeline.enums ¶

Enumerations for the immunization pipeline.

BundleStrategy ¶

Bases: Enum

Bundle grouping strategy.

Source code in pipeline/enums.py

class BundleStrategy(Enum):
    """Bundle grouping strategy."""

    SIZE = "size"
    SCHOOL = "school"
    BOARD = "board"

    @classmethod
    def from_string(cls, value: str | None) -> "BundleStrategy":
        """Convert string to BundleStrategy.

        Parameters
        ----------
        value : str | None
            Bundle strategy name ('size', 'school', 'board'), or None for default.

        Returns
        -------
        BundleStrategy
            Corresponding BundleStrategy enum, defaults to SIZE if value is None.

        Raises
        ------
        ValueError
            If value is not a valid strategy name.
        """
        if value is None:
            return cls.SIZE

        value_lower = value.lower()
        for strategy in cls:
            if strategy.value == value_lower:
                return strategy

        raise ValueError(
            f"Unknown bundle strategy: {value}. "
            f"Valid options: {', '.join(s.value for s in cls)}"
        )

from_string `classmethod` ¶

from_string(value)

Convert string to BundleStrategy.

Parameters:

Name	Type	Description	Default
`value`	`str \| None`	Bundle strategy name ('size', 'school', 'board'), or None for default.	required

Returns:

Type	Description
`BundleStrategy`	Corresponding BundleStrategy enum, defaults to SIZE if value is None.

Raises:

Type	Description
`ValueError`	If value is not a valid strategy name.

Source code in pipeline/enums.py

@classmethod
def from_string(cls, value: str | None) -> "BundleStrategy":
    """Convert string to BundleStrategy.

    Parameters
    ----------
    value : str | None
        Bundle strategy name ('size', 'school', 'board'), or None for default.

    Returns
    -------
    BundleStrategy
        Corresponding BundleStrategy enum, defaults to SIZE if value is None.

    Raises
    ------
    ValueError
        If value is not a valid strategy name.
    """
    if value is None:
        return cls.SIZE

    value_lower = value.lower()
    for strategy in cls:
        if strategy.value == value_lower:
            return strategy

    raise ValueError(
        f"Unknown bundle strategy: {value}. "
        f"Valid options: {', '.join(s.value for s in cls)}"
    )

BundleType ¶

Bases: Enum

Type descriptor for bundle operation.

Source code in pipeline/enums.py

class BundleType(Enum):
    """Type descriptor for bundle operation."""

    SIZE_BASED = "size_based"
    SCHOOL_GROUPED = "school_grouped"
    BOARD_GROUPED = "board_grouped"

Language ¶

Bases: Enum

Supported output languages for immunization notices.

Each language corresponds to:

A template renderer in templates/ (en_template.py, fr_template.py, etc.)
Localization of dates, disease names, and notice formatting
An artifact language code stored in preprocessed data

Currently supports English and French; extensible for future languages.

Attributes:

Name	Type	Description
`ENGLISH`	`str`	English language code ('en'). Templates: templates/en_template.py
`FRENCH`	`str`	French language code ('fr'). Templates: templates/fr_template.py

from_string `classmethod` ¶

from_string(value)

Convert string to Language enum.

Provides safe conversion from user input or configuration strings to Language enum values. Used at CLI entry point and configuration loading to fail fast on invalid language codes.

Parameters:

Name	Type	Description	Default
`value`	`str \| None`	Language code ('en', 'fr'), or None for default (ENGLISH). Case-insensitive (normalizes to lowercase).	required

Returns:

Type	Description
`Language`	Corresponding Language enum value.

Raises:

Type	Description
`ValueError`	If value is not a valid language code. Error message lists all available options.

Examples:

>>> Language.from_string('en')
<Language.ENGLISH: 'en'>

>>> Language.from_string('EN')  # Case-insensitive
<Language.ENGLISH: 'en'>

>>> Language.from_string(None)  # Default to English
<Language.ENGLISH: 'en'>

>>> Language.from_string('es')  # Unsupported
ValueError: Unsupported language: es. Valid options: en, fr

Source code in pipeline/enums.py

@classmethod
def from_string(cls, value: str | None) -> "Language":
    """Convert string to Language enum.

    Provides safe conversion from user input or configuration strings to
    Language enum values. Used at CLI entry point and configuration loading
    to fail fast on invalid language codes.

    Parameters
    ----------
    value : str | None
        Language code ('en', 'fr'), or None for default (ENGLISH).
        Case-insensitive (normalizes to lowercase).

    Returns
    -------
    Language
        Corresponding Language enum value.

    Raises
    ------
    ValueError
        If value is not a valid language code. Error message lists
        all available options.

    Examples
    --------
    >>> Language.from_string('en')
    <Language.ENGLISH: 'en'>

    >>> Language.from_string('EN')  # Case-insensitive
    <Language.ENGLISH: 'en'>

    >>> Language.from_string(None)  # Default to English
    <Language.ENGLISH: 'en'>

    >>> Language.from_string('es')  # Unsupported
    ValueError: Unsupported language: es. Valid options: en, fr
    """
    if value is None:
        return cls.ENGLISH

    value_lower = value.lower()
    for lang in cls:
        if lang.value == value_lower:
            return lang

    raise ValueError(
        f"Unsupported language: {value}. "
        f"Valid options: {', '.join(lang.value for lang in cls)}"
    )

all_codes `classmethod` ¶

all_codes()

Get set of all supported language codes.

Returns:

Type	Description
`set[str]`	Set of all language codes (e.g., {'en', 'fr'}).

Examples:

>>> Language.all_codes()
{'en', 'fr'}

Source code in pipeline/enums.py

@classmethod
def all_codes(cls) -> set[str]:
    """Get set of all supported language codes.

    Returns
    -------
    set[str]
        Set of all language codes (e.g., {'en', 'fr'}).

    Examples
    --------
    >>> Language.all_codes()
    {'en', 'fr'}
    """
    return {lang.value for lang in cls}

TemplateField ¶

Bases: Enum

Available placeholder fields for template rendering (QR codes, PDF passwords).

These fields are dynamically generated from client data by build_client_context() and can be used in configuration templates for:

QR code payloads (qr.payload_template in parameters.yaml)
PDF password generation (encryption.password.template in parameters.yaml)

All fields are validated by validate_and_format_template() to catch config errors early and provide clear error messages.

Fields

CLIENT_ID : str Unique client identifier

FIRST_NAME : str Client's given name.

LAST_NAME : str Client's family name.

NAME : str Full name (first + last combined).

DATE_OF_BIRTH : str Display format (e.g., "Jan 8, 2025" or "8 janvier 2025").

DATE_OF_BIRTH_ISO : str ISO 8601 format: YYYY-MM-DD (e.g., "2015-03-15").

DATE_OF_BIRTH_ISO_COMPACT : str Compact ISO format without hyphens: YYYYMMDD (e.g., "20150315").

SCHOOL : str School name.

BOARD : str School board name.

STREET_ADDRESS : str Full street address.

CITY : str City/municipality.

PROVINCE : str Province/territory.

POSTAL_CODE : str Postal/ZIP code.

LANGUAGE_CODE : str ISO 639-1 language code: 'en' or 'fr'.

all_values `classmethod` ¶

all_values()

Get set of all available field names for use as allowed_fields whitelist.

Returns:

Type	Description
`set[str]`	Set of all field values (e.g., {'client_id', 'first_name', ...}).

Examples:

>>> TemplateField.all_values()
{'client_id', 'first_name', 'last_name', 'name', ...}

Source code in pipeline/enums.py

@classmethod
def all_values(cls) -> set[str]:
    """Get set of all available field names for use as allowed_fields whitelist.

    Returns
    -------
    set[str]
        Set of all field values (e.g., {'client_id', 'first_name', ...}).

    Examples
    --------
    >>> TemplateField.all_values()
    {'client_id', 'first_name', 'last_name', 'name', ...}
    """
    return {field.value for field in cls}

Step 1 — Output Preparation¶

pipeline.prepare_output ¶

Utility to prepare the pipeline output directory.

This script ensures the output directory exists, optionally removes any existing contents (while preserving the logs directory), and creates the log directory if needed.

Note: This module is called exclusively from orchestrator.py. The internal functions handle all logic; CLI support has been removed in favor of explicit function calls from the orchestrator.

Input Contract:

Receives output directory path and auto_remove_output flag from config
Assumes configuration has been validated by load_config() at orchestrator startup

Output Contract:

Creates output directory structure if it doesn't exist
Optionally removes existing output while preserving logs
Ensures log and artifact subdirectories are ready for pipeline output

Error Handling:

File system permission errors raise immediately (infrastructure error)
Missing directories are created automatically (no error)
Fails fast on unrecoverable I/O errors

Validation Contract:

What this module validates:

Output directory can be created if missing
File system permissions allow write/delete operations
Log directory can be preserved during cleanup

What this module assumes (validated upstream):

Config keys (pipeline.auto_remove_output) have been validated by load_config()
Output path is a valid directory path (basic format validation)

Note: This is a utility/setup step. Runs before the main pipeline; failures halt everything (fail-fast) since output directory is prerequisite for all steps.

is_log_directory ¶

is_log_directory(candidate, log_dir)

Check if a path is the log directory or one of its ancestors.

Module-internal helper for purge_output_directory(). The pipeline stores logs under a dedicated directory (output/logs). When cleaning the output directory we must preserve the log directory and its contents. This check accounts for potential symlinks by resolving both paths.

Parameters:

Name	Type	Description	Default
`candidate`	`Path`	Path to check.	required
`log_dir`	`Path`	Reference log directory path.	required

Returns:

Type	Description
`bool`	True if candidate is the log directory or an ancestor, False otherwise.

Source code in pipeline/prepare_output.py

def is_log_directory(candidate: Path, log_dir: Path) -> bool:
    """Check if a path is the log directory or one of its ancestors.

    Module-internal helper for purge_output_directory(). The pipeline stores logs
    under a dedicated directory (``output/logs``). When cleaning the output directory
    we must preserve the log directory and its contents. This check accounts for
    potential symlinks by resolving both paths.

    Parameters
    ----------
    candidate : Path
        Path to check.
    log_dir : Path
        Reference log directory path.

    Returns
    -------
    bool
        True if candidate is the log directory or an ancestor, False otherwise.
    """

    try:
        candidate_resolved = candidate.resolve()
    except FileNotFoundError:
        # If the child disappears while scanning, treat it as non-log.
        return False

    try:
        log_resolved = log_dir.resolve()
    except FileNotFoundError:
        # If the log directory does not exist yet we should not attempt to skip
        # siblings – the caller will create it afterwards.
        return False

    return candidate_resolved == log_resolved

purge_output_directory ¶

purge_output_directory(output_dir, log_dir)

Remove everything inside output_dir except the logs directory.

Module-internal helper for prepare_output_directory(). Recursively deletes all files and subdirectories except the log directory, which is preserved for audit trails.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Output directory to clean.	required
`log_dir`	`Path`	Log directory to preserve.	required

Source code in pipeline/prepare_output.py

def purge_output_directory(output_dir: Path, log_dir: Path) -> None:
    """Remove everything inside output_dir except the logs directory.

    Module-internal helper for prepare_output_directory(). Recursively deletes
    all files and subdirectories except the log directory, which is preserved
    for audit trails.

    Parameters
    ----------
    output_dir : Path
        Output directory to clean.
    log_dir : Path
        Log directory to preserve.
    """

    for child in output_dir.iterdir():
        if is_log_directory(child, log_dir):
            continue
        if child.is_dir():
            shutil.rmtree(child)
        else:
            child.unlink(missing_ok=True)

default_prompt ¶

default_prompt(output_dir)

Prompt user for confirmation to delete output directory contents.

Module-internal helper for prepare_output_directory(). Interactive prompt to prevent accidental data loss when auto_remove is False.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Directory path being queried.	required

Returns:

Type	Description
`bool`	True if user confirms (y/yes), False otherwise.

Source code in pipeline/prepare_output.py

def default_prompt(output_dir: Path) -> bool:
    """Prompt user for confirmation to delete output directory contents.

    Module-internal helper for prepare_output_directory(). Interactive prompt
    to prevent accidental data loss when auto_remove is False.

    Parameters
    ----------
    output_dir : Path
        Directory path being queried.

    Returns
    -------
    bool
        True if user confirms (y/yes), False otherwise.
    """
    print("")
    print(f"⚠️  Output directory already exists: {output_dir}")
    response = input("Delete contents (except logs) and proceed? [y/N] ")
    return response.strip().lower() in {"y", "yes"}

prepare_output_directory ¶

prepare_output_directory(output_dir, log_dir, auto_remove, prompt=None)

Prepare the output directory for a new pipeline run.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root directory for pipeline outputs.	required
`log_dir`	`Path`	Directory where pipeline logs are stored. Typically a subdirectory of `output_dir`.	required
`auto_remove`	`bool`	When `True` the directory is emptied without prompting the user.	required
`prompt`	`Optional[Callable[[Path], bool]]`	Optional callable used to prompt the user for confirmation. A return value of `True` proceeds with cleanup, while `False` aborts.	`None`

Returns:

Type	Description
`bool`	`True` when preparation succeeded, `False` when the user aborted the operation.

Source code in pipeline/prepare_output.py

def prepare_output_directory(
    output_dir: Path,
    log_dir: Path,
    auto_remove: bool,
    prompt: Optional[Callable[[Path], bool]] = None,
) -> bool:
    """Prepare the output directory for a new pipeline run.

    Parameters
    ----------
    output_dir:
        Root directory for pipeline outputs.
    log_dir:
        Directory where pipeline logs are stored. Typically a subdirectory of
        ``output_dir``.
    auto_remove:
        When ``True`` the directory is emptied without prompting the user.
    prompt:
        Optional callable used to prompt the user for confirmation. A return
        value of ``True`` proceeds with cleanup, while ``False`` aborts.

    Returns
    -------
    bool
        ``True`` when preparation succeeded, ``False`` when the user aborted the
        operation.
    """

    prompt_callable = prompt or default_prompt

    if output_dir.exists():
        if not auto_remove and not prompt_callable(output_dir):
            print("❌ Pipeline cancelled. No changes made.")
            return False
        purge_output_directory(output_dir, log_dir)
    else:
        output_dir.mkdir(parents=True, exist_ok=True)

    log_dir.mkdir(parents=True, exist_ok=True)
    return True

Step 2 — Preprocessing¶

pipeline.preprocess ¶

Preprocessing pipeline for immunization-charts.

Normalizes and structures input data into a single JSON artifact for downstream pipeline steps. Handles data validation, client sorting, and vaccine processing. QR code generation is handled by a separate step after preprocessing.

Input Contract:

Reads raw client data from CSV or Excel file (.xlsx, .xls, .csv)
Validates file type and encoding (tries multiple encodings for CSV)
Validates all required columns are present

Output Contract:

Writes preprocessed artifact JSON to output/artifacts/preprocessed_clients_*.json
Artifact contains all valid client records with normalized data types
Artifact includes metadata (run_id, language, created_at, warnings)
Downstream steps assume artifact is valid; preprocessing is the sole validation step

Error Handling:

File I/O errors (missing file, unsupported format) raise immediately (infrastructure)
Missing required columns raise immediately (data error in required step)
Invalid data (missing DOB, unparseable date) logged as warnings; processing continues
Fail-fast for structural issues; warn-and-continue for data quality issues

Validation Contract:

What this module validates:

Input file exists and is readable
Input file is supported format (.xlsx, .xls, .csv)
File encoding (tries UTF-8, Latin-1, etc. for CSV)
All required columns are present in input data
Client data normalization (DOB parsing, vaccine processing)
Language code is valid (from CLI argument)

What this module assumes (validated upstream):

Language code from CLI is valid (validated by Language.from_string() at orchestrator)
Disease and vaccine reference data are valid JSON (validated by config loading)

Note: This is the primary validation step. Downstream steps trust preprocessing output.

convert_date_string ¶

convert_date_string(date_str, locale='en')

Convert a date to display format with locale-aware formatting.

Uses Babel for locale-aware date formatting. Generates format like "May 8, 2025" (en) or "8 mai 2025" (fr) depending on locale.

Parameters:

Name	Type	Description	Default
`date_str`	`str \| datetime \| Timestamp`	Date string in YYYY-MM-DD format or datetime-like object.	required
`locale`	`str`	Locale code for date formatting (default: "en"). Examples: "en" for English, "fr" for French.	`'en'`

Returns:

Type	Description
`str \| None`	Date in locale-specific format, or None if input is null.

Raises:

Type	Description
`ValueError`	If date_str is a string in unrecognized format.

Source code in pipeline/preprocess.py

def convert_date_string(
    date_str: str | datetime | pd.Timestamp, locale: str = "en"
) -> str | None:
    """Convert a date to display format with locale-aware formatting.

    Uses Babel for locale-aware date formatting. Generates format like
    "May 8, 2025" (en) or "8 mai 2025" (fr) depending on locale.

    Parameters
    ----------
    date_str : str | datetime | pd.Timestamp
        Date string in YYYY-MM-DD format or datetime-like object.
    locale : str, optional
        Locale code for date formatting (default: "en").
        Examples: "en" for English, "fr" for French.

    Returns
    -------
    str | None
        Date in locale-specific format, or None if input is null.

    Raises
    ------
    ValueError
        If date_str is a string in unrecognized format.
    """
    if pd.isna(date_str):
        return None

    # If it's already a datetime or Timestamp, use it directly
    if isinstance(date_str, (pd.Timestamp, datetime)):
        date_obj = date_str
    else:
        # Parse string input
        try:
            date_obj = datetime.strptime(str(date_str).strip(), "%Y-%m-%d")
        except ValueError:
            raise ValueError(f"Unrecognized date format: {date_str}")

    return format_date(date_obj, format="long", locale=locale)

format_iso_date_for_language ¶

format_iso_date_for_language(iso_date, language)

Format an ISO date string with locale-aware formatting for the given language.

Converts a date from ISO format (YYYY-MM-DD) to a long, locale-specific display format using Babel. This function handles language-specific date formatting for templates.

Parameters:

Name	Type	Description	Default
`iso_date`	`str`	Date in ISO format (YYYY-MM-DD), e.g., "2025-08-31".	required
`language`	`str`	ISO 639-1 language code ("en", "fr", etc.).	required

Returns:

Type	Description
`str`	Formatted date in the specified language, e.g., "August 31, 2025" (en) or "31 août 2025" (fr).

Raises:

Type	Description
`ValueError`	If iso_date is not in YYYY-MM-DD format.

Source code in pipeline/preprocess.py

def format_iso_date_for_language(iso_date: str, language: str) -> str:
    """Format an ISO date string with locale-aware formatting for the given language.

    Converts a date from ISO format (YYYY-MM-DD) to a long, locale-specific
    display format using Babel. This function handles language-specific date
    formatting for templates.

    Parameters
    ----------
    iso_date : str
        Date in ISO format (YYYY-MM-DD), e.g., "2025-08-31".
    language : str
        ISO 639-1 language code ("en", "fr", etc.).

    Returns
    -------
    str
        Formatted date in the specified language, e.g.,
        "August 31, 2025" (en) or "31 août 2025" (fr).

    Raises
    ------
    ValueError
        If iso_date is not in YYYY-MM-DD format.
    """
    locale_map = {"en": "en_US", "fr": "fr_FR"}
    locale = locale_map.get(language, language)

    try:
        date_obj = datetime.strptime(iso_date.strip(), "%Y-%m-%d")
    except ValueError:
        raise ValueError(f"Invalid ISO date format: {iso_date}. Expected YYYY-MM-DD.")

    return format_date(date_obj, format="long", locale=locale)

check_addresses_complete ¶

check_addresses_complete(df)

Check if address fields are complete in the DataFrame.

Adds a boolean 'address_complete' column based on presence of street address, city, province, and postal code.

Source code in pipeline/preprocess.py

def check_addresses_complete(df: pd.DataFrame) -> pd.DataFrame:
    """
    Check if address fields are complete in the DataFrame.

    Adds a boolean 'address_complete' column based on presence of
    street address, city, province, and postal code.
    """

    df = df.copy()

    # Normalize text fields: convert to string, strip whitespace, convert "" to NA
    address_cols = [
        "STREET_ADDRESS_LINE_1",
        "STREET_ADDRESS_LINE_2",
        "CITY",
        "PROVINCE",
        "POSTAL_CODE",
    ]

    for col in address_cols:
        df[col] = df[col].astype(str).str.strip().replace({"": pd.NA, "nan": pd.NA})

    # Build combined address line
    df["ADDRESS"] = (
        df["STREET_ADDRESS_LINE_1"].fillna("")
        + " "
        + df["STREET_ADDRESS_LINE_2"].fillna("")
    ).str.strip()

    df["ADDRESS"] = df["ADDRESS"].replace({"": pd.NA})

    # Check completeness
    df["address_complete"] = (
        df["ADDRESS"].notna()
        & df["CITY"].notna()
        & df["PROVINCE"].notna()
        & df["POSTAL_CODE"].notna()
    )

    if not df["address_complete"].all():
        incomplete_count = (~df["address_complete"]).sum()
        LOG.warning(
            "There are %d records with incomplete address information.",
            incomplete_count,
        )

        incomplete_records = df.loc[~df["address_complete"]]

        incomplete_path = Path("output/incomplete_addresses.csv")
        incomplete_records.to_csv(incomplete_path, index=False)
        LOG.info("Incomplete address records written to %s", incomplete_path)

    # Return only rows with complete addresses
    return df.loc[df["address_complete"]].drop(columns=["address_complete"])

convert_date_iso ¶

convert_date_iso(date_str)

Convert a date from English display format to ISO format.

Reverses the formatting from convert_date_string(). Expects input in "Mon DD, YYYY" format (e.g., "May 8, 2025").

Parameters:

Name	Type	Description	Default
`date_str`	`str`	Date in English display format (e.g., "May 8, 2025").	required

Returns:

Type	Description
`str`	Date in ISO format (YYYY-MM-DD).

Source code in pipeline/preprocess.py

def convert_date_iso(date_str: str) -> str:
    """Convert a date from English display format to ISO format.

    Reverses the formatting from convert_date_string(). Expects input
    in "Mon DD, YYYY" format (e.g., "May 8, 2025").

    Parameters
    ----------
    date_str : str
        Date in English display format (e.g., "May 8, 2025").

    Returns
    -------
    str
        Date in ISO format (YYYY-MM-DD).
    """
    date_obj = datetime.strptime(date_str, "%b %d, %Y")
    return date_obj.strftime("%Y-%m-%d")

over_16_check ¶

over_16_check(date_of_birth, date_notice_delivery)

Check if a client is over 16 years old on notice delivery date.

Parameters:

Name	Type	Description	Default
`date_of_birth`	`str`	Date of birth in YYYY-MM-DD format.	required
`date_notice_delivery`	`str`	Notice delivery date in YYYY-MM-DD format.	required

Returns:

Type	Description
`bool`	True if the client is over 16 years old on date_notice_delivery, False otherwise.

Source code in pipeline/preprocess.py

def over_16_check(date_of_birth, date_notice_delivery):
    """Check if a client is over 16 years old on notice delivery date.

    Parameters
    ----------
    date_of_birth : str
        Date of birth in YYYY-MM-DD format.
    date_notice_delivery : str
        Notice delivery date in YYYY-MM-DD format.

    Returns
    -------
    bool
        True if the client is over 16 years old on date_notice_delivery, False otherwise.
    """

    birth_datetime = datetime.strptime(date_of_birth, "%Y-%m-%d")
    delivery_datetime = datetime.strptime(date_notice_delivery, "%Y-%m-%d")

    age = delivery_datetime.year - birth_datetime.year

    # Adjust if birthday hasn't occurred yet in the DOV month
    if (delivery_datetime.month < birth_datetime.month) or (
        delivery_datetime.month == birth_datetime.month
        and delivery_datetime.day < birth_datetime.day
    ):
        age -= 1

    return age >= 16

configure_logging ¶

configure_logging(output_dir, run_id)

Configure file logging for the preprocessing step.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root output directory where logs subdirectory will be created.	required
`run_id`	`str`	Unique run identifier used in log filename.	required

Returns:

Type	Description
`Path`	Path to the created log file.

Source code in pipeline/preprocess.py

def configure_logging(output_dir: Path, run_id: str) -> Path:
    """Configure file logging for the preprocessing step.

    Parameters
    ----------
    output_dir : Path
        Root output directory where logs subdirectory will be created.
    run_id : str
        Unique run identifier used in log filename.

    Returns
    -------
    Path
        Path to the created log file.
    """
    log_dir = output_dir / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)
    log_path = log_dir / f"preprocess_{run_id}.log"

    handler = logging.FileHandler(log_path, encoding="utf-8")
    formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
    handler.setFormatter(formatter)

    root_logger = logging.getLogger()
    root_logger.handlers.clear()
    root_logger.setLevel(logging.INFO)
    root_logger.addHandler(handler)

    return log_path

detect_file_type ¶

detect_file_type(file_path)

Detect file type by extension.

Parameters:

Name	Type	Description	Default
`file_path`	`Path`	Path to the file to detect.	required

Returns:

Type	Description
`str`	File extension in lowercase (e.g., '.xlsx', '.csv').

Raises:

Type	Description
`FileNotFoundError`	If the file does not exist.

Source code in pipeline/preprocess.py

def detect_file_type(file_path: Path) -> str:
    """Detect file type by extension.

    Parameters
    ----------
    file_path : Path
        Path to the file to detect.

    Returns
    -------
    str
        File extension in lowercase (e.g., '.xlsx', '.csv').

    Raises
    ------
    FileNotFoundError
        If the file does not exist.
    """
    if not file_path.exists():
        raise FileNotFoundError(f"Input file not found: {file_path}")
    return file_path.suffix.lower()

read_input ¶

read_input(file_path)

Read CSV or Excel input file into a pandas DataFrame.

Supports .xlsx, .xls, and .csv formats with robust encoding and delimiter detection. This is a critical preprocessing step that loads raw client data.

Parameters:

Name	Type	Description	Default
`file_path`	`Path`	Path to the input file (CSV, XLSX, or XLS).	required

Returns:

Type	Description
`DataFrame`	DataFrame with raw client data loaded from the file.

Raises:

Type	Description
`ValueError`	If file type is unsupported or CSV cannot be decoded with common encodings.
`Exception`	If file reading fails for any reason (logged to preprocessing logs).

Source code in pipeline/preprocess.py

def read_input(file_path: Path) -> pd.DataFrame:
    """Read CSV or Excel input file into a pandas DataFrame.

    Supports .xlsx, .xls, and .csv formats with robust encoding and delimiter
    detection. This is a critical preprocessing step that loads raw client data.

    Parameters
    ----------
    file_path : Path
        Path to the input file (CSV, XLSX, or XLS).

    Returns
    -------
    pd.DataFrame
        DataFrame with raw client data loaded from the file.

    Raises
    ------
    ValueError
        If file type is unsupported or CSV cannot be decoded with common encodings.
    Exception
        If file reading fails for any reason (logged to preprocessing logs).
    """
    ext = detect_file_type(file_path)

    try:
        if ext in [".xlsx", ".xls"]:
            df = pd.read_excel(file_path, engine="openpyxl", dtype={"CLIENT ID": str})
        elif ext == ".csv":
            # Try common encodings
            for enc in ["utf-8-sig", "latin-1", "cp1252"]:
                try:
                    # Let pandas sniff the delimiter
                    df = pd.read_csv(file_path, sep=None, encoding=enc, engine="python")
                    break
                except (UnicodeDecodeError, pd.errors.ParserError):
                    continue
            else:
                raise ValueError(
                    "Could not decode CSV with common encodings or delimiters"
                )
        else:
            raise ValueError(f"Unsupported file type: {ext}")

        LOG.info("Loaded %s rows from %s", len(df), file_path)
        return df

    except Exception as exc:  # pragma: no cover - logging branch
        LOG.error("Failed to read %s: %s", file_path, exc)
        raise

normalize ¶

normalize(col)

Normalize formatting prior to matching.

Source code in pipeline/preprocess.py

def normalize(col: str) -> str:
    """Normalize formatting prior to matching."""

    # Trim whitespaces
    col_normalized = col.lower().strip().replace("_", " ").replace("-", " ")

    # Check to see if double whitespace
    col_normalized = re.sub(r"\s+", " ", col_normalized)

    return col_normalized

split_vaccine_due_entry ¶

split_vaccine_due_entry(item)

Separate a disease name from its optional dose field.

Source code in pipeline/preprocess.py

def split_vaccine_due_entry(item: str) -> tuple[str, str | None]:
    """Separate a disease name from its optional dose field."""

    disease, separator, dose = item.strip().rpartition(" -")
    if not separator:
        return item.strip(), None
    return disease.strip(), dose.strip()

hide_vaccine_due_doses ¶

hide_vaccine_due_doses(vaccine_due_list)

Remove supplied dose fields while preserving ordinary disease labels.

Source code in pipeline/preprocess.py

def hide_vaccine_due_doses(vaccine_due_list: list[str]) -> list[str]:
    """Remove supplied dose fields while preserving ordinary disease labels."""

    return [split_vaccine_due_entry(item)[0] for item in vaccine_due_list]

format_vaccine_due_list ¶

format_vaccine_due_list(vaccine_due_list)

Format dose-bearing overdue entries for display.

Source code in pipeline/preprocess.py

def format_vaccine_due_list(vaccine_due_list: list[str]) -> list[str]:
    """Format dose-bearing overdue entries for display."""

    formatted: list[str] = []
    for item in vaccine_due_list:
        disease, dose = split_vaccine_due_entry(item)
        if dose is None:
            raise ValueError(
                "preprocess.include_dose requires overdue entries using the "
                "'<disease> - <dose>' schema"
            )

        if not dose:
            formatted.append(disease)
            continue

        if dose[-1] == "1":
            formatted.append(f"{disease} ({dose}st dose)")
        elif dose[-1] == "2":
            formatted.append(f"{disease} ({dose}nd dose)")
        elif dose[-1] == "3":
            formatted.append(f"{disease} ({dose}rd dose)")
        else:
            # Ensure separated section is an integer within vaccine series range
            try:
                if int(dose) < 10:
                    formatted.append(f"{disease} ({dose}th dose)")
                else:
                    LOG.warning(
                        f"Vaccine with ' - ' separator but invalid dose number: {item}"
                    )
                    formatted.append(item.strip())

            except (ValueError, TypeError):
                formatted.append(item.strip())

    return formatted

map_columns ¶

map_columns(df, required_columns=REQUIRED_COLUMNS)

Map dataframe columns to a set of required column names using fuzzy matching.

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	Input dataframe whose columns will be matched and optionally renamed.	required
`required_columns`	`Sequence[str]`	Sequence of expected/required column names to match against. Defaults to REQUIRED_COLUMNS.	`REQUIRED_COLUMNS`

Returns:

Type	Description
`tuple[DataFrame, dict]`	A tuple (renamed_df, col_map) where `renamed_df` is `df` with columns renamed according to successful matches, and `col_map` is a dict mapping original column names (keys) to matched required column names (values).

Behavior

Normalizes input column names and required column names using normalize(...) before matching.
For each normalized input column, finds the best fuzzy match among normalized required_columns using process.extractOne(..., scorer=fuzz.partial_ratio).
If the best match score is >= 80 (threshold in the implementation), the original input column name is mapped to the corresponding required column name; the mapping is recorded and the dataframe is renamed accordingly.
A debug line is printed for each accepted match: "Matching '' to '' with score ".
Columns with a best match score < 80 are ignored (not included in col_map); matches with score 0 are effectively dropped.
The code resolves the original column name by locating the first column whose normalized form equals the normalized input. If multiple original columns normalize to the same value, the first encountered is used.

Notes

The function depends on external helpers: normalize, process.extractOne, and fuzz.partial_ratio.
The match threshold (80) is adjustable; lowering it makes matching more permissive, raising it makes it stricter.
A StopIteration may occur if a normalized input column cannot be resolved back to an original column name.
required_columns should be an iterable of strings; df.columns are expected to be convertible to strings.

Examples:

Example usage (illustrative only):¶

renamed_df, mapping = map_columns(df, required_columns=['state', 'date', 'count'])¶

Source code in pipeline/preprocess.py

def map_columns(df: pd.DataFrame, required_columns=REQUIRED_COLUMNS):
    """
    Map dataframe columns to a set of required column names using fuzzy matching.
    Parameters
    ----------
    df : pandas.DataFrame
        Input dataframe whose columns will be matched and optionally renamed.
    required_columns : Sequence[str], optional
        Sequence of expected/required column names to match against. Defaults to REQUIRED_COLUMNS.
    Returns
    -------
    tuple[pandas.DataFrame, dict]
        A tuple (renamed_df, col_map) where `renamed_df` is `df` with columns renamed according to successful matches,
        and `col_map` is a dict mapping original column names (keys) to matched required column names (values).
    Behavior
    --------
    - Normalizes input column names and required column names using `normalize(...)` before matching.
    - For each normalized input column, finds the best fuzzy match among normalized `required_columns` using
      `process.extractOne(..., scorer=fuzz.partial_ratio)`.
    - If the best match score is >= 80 (threshold in the implementation), the original input column name is mapped to the
      corresponding required column name; the mapping is recorded and the dataframe is renamed accordingly.
    - A debug line is printed for each accepted match: "Matching '<normalized_input>' to '<best_match>' with score <score>".
    - Columns with a best match score < 80 are ignored (not included in `col_map`); matches with score 0 are effectively dropped.
    - The code resolves the original column name by locating the first column whose normalized form equals the normalized input.
      If multiple original columns normalize to the same value, the first encountered is used.
    Notes
    -----
    - The function depends on external helpers: `normalize`, `process.extractOne`, and `fuzz.partial_ratio`.
    - The match threshold (80) is adjustable; lowering it makes matching more permissive, raising it makes it stricter.
    - A `StopIteration` may occur if a normalized input column cannot be resolved back to an original column name.
    - `required_columns` should be an iterable of strings; `df.columns` are expected to be convertible to strings.
    Examples
    --------
    # Example usage (illustrative only):
    # renamed_df, mapping = map_columns(df, required_columns=['state', 'date', 'count'])
    """
    input_cols = df.columns
    col_map = {}
    normalized_required = [normalize(req) for req in required_columns]
    best_matches = {}

    # Check each input column against required columns
    for actual_in_col in input_cols:
        input_col = normalize(actual_in_col)
        result = process.extractOne(
            query=input_col,
            choices=normalized_required,
            scorer=fuzz.partial_ratio,
        )
        if result is None:
            continue

        _, score, index = result
        if score < THRESHOLD:
            continue

        best_match = required_columns[index]
        print(f"Matching '{input_col}' to '{best_match}' with score {score}")

        prior = best_matches.get(best_match)
        if prior is None:
            print(
                f"The value {best_match} does not exist in the dictionary - adding value."
            )
            col_map[actual_in_col] = best_match
            best_matches[best_match] = {"actual_in_col": actual_in_col, "score": score}
        elif score > prior["score"]:
            print(
                f"{input_col} has a higher score than current best match in the dictionary - replacing value."
            )
            col_map.pop(prior["actual_in_col"], None)
            col_map[actual_in_col] = best_match
            best_matches[best_match] = {"actual_in_col": actual_in_col, "score": score}

    return df.rename(columns=col_map), col_map

filter_columns ¶

filter_columns(df: DataFrame, required_columns: list[str] = REQUIRED_COLUMNS) -> pd.DataFrame

filter_columns(df: None, required_columns: list[str] = REQUIRED_COLUMNS) -> None

filter_columns(df, required_columns=REQUIRED_COLUMNS)

Filter dataframe to only include required columns.

Source code in pipeline/preprocess.py

def filter_columns(
    df: pd.DataFrame | None, required_columns: list[str] = REQUIRED_COLUMNS
) -> pd.DataFrame | None:
    """Filter dataframe to only include required columns."""
    if df is None or df.empty:
        return df

    return df[[col for col in df.columns if col in required_columns]]

normalize_dataframe ¶

normalize_dataframe(df)

Validate, rename, and normalize the input DataFrame.

Combines column validation with type normalization. Standardizes column names to uppercase with underscores, validates all required columns are present, then normalizes data types and fills missing values.

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	Input DataFrame with raw client data (column names may have mixed case/spacing).	required

Returns:

Type	Description
`DataFrame`	Copy of DataFrame with validated, renamed, and normalized columns.

Raises:

Type	Description
`ValueError`	If any required columns are missing from the DataFrame.

Source code in pipeline/preprocess.py

def normalize_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    """Validate, rename, and normalize the input DataFrame.

    Combines column validation with type normalization. Standardizes column
    names to uppercase with underscores, validates all required columns are
    present, then normalizes data types and fills missing values.

    Parameters
    ----------
    df : pd.DataFrame
        Input DataFrame with raw client data (column names may have mixed
        case/spacing).

    Returns
    -------
    pd.DataFrame
        Copy of DataFrame with validated, renamed, and normalized columns.

    Raises
    ------
    ValueError
        If any required columns are missing from the DataFrame.
    """
    working = df.copy()

    # Normalize and validate column names in a single pass
    def _normalize_col(col: str) -> str:
        col = col.strip().upper().replace(" ", "_")
        return "PROVINCE" if col == "PROVINCE/TERRITORY" else col

    working.columns = [_normalize_col(col) for col in working.columns]

    _required_normalized = {_normalize_col(col) for col in REQUIRED_COLUMNS}
    missing = [col for col in _required_normalized if col not in working.columns]
    if missing:
        raise ValueError(
            f"Missing required columns: {missing} \n Found columns: {list(working.columns)} "
        )

    # Normalize data types and fill missing values.
    # Required string columns are derived from REQUIRED_COLUMNS; optional
    # supplemental columns are listed separately.
    _non_string_required = {"CLIENT_ID", "DATE_OF_BIRTH", "OVERDUE_DISEASE", "IMMS_GIVEN"}
    required_string_cols = [col for col in _required_normalized if col not in _non_string_required]
    optional_string_cols = ["SCHOOL_TYPE", "BOARD_NAME", "BOARD_ID", "SCHOOL_ID", "UNIQUE_ID"]

    for column in required_string_cols + optional_string_cols:
        if column not in working.columns:
            working[column] = ""
        working[column] = working[column].fillna(" ").astype(str).str.strip()

    working["DATE_OF_BIRTH"] = pd.to_datetime(working["DATE_OF_BIRTH"], errors="coerce")
    if "AGE" in working.columns:
        working["AGE"] = pd.to_numeric(working["AGE"], errors="coerce")
    else:
        working["AGE"] = pd.NA

    return working

synthesize_identifier ¶

synthesize_identifier(existing, source, prefix)

Generate a deterministic identifier if one is not provided.

Source code in pipeline/preprocess.py

def synthesize_identifier(existing: str, source: str, prefix: str) -> str:
    """Generate a deterministic identifier if one is not provided."""
    existing = (existing or "").strip()
    if existing:
        return existing

    base = (source or "").strip().lower() or "unknown"
    digest = sha1(base.encode("utf-8")).hexdigest()[:10]
    return f"{prefix}_{digest}"

process_vaccines_due ¶

process_vaccines_due(vaccines_due, language)

Map overdue diseases to canonical disease names.

Normalizes raw input disease strings to canonical disease names using config/disease_normalization.json. Returns a comma-separated string of canonical disease names.

Parameters:

Name	Type	Description	Default
`vaccines_due`	`Any`	Raw string of comma-separated disease names from input.	required
`language`	`str`	Language code (e.g., "en", "fr"). Used for logging.	required

Returns:

Type	Description
`str`	Comma-separated string of canonical disease names (English). Empty string if input is empty or invalid.

Source code in pipeline/preprocess.py

def process_vaccines_due(vaccines_due: Any, language: str) -> str:
    """Map overdue diseases to canonical disease names.

    Normalizes raw input disease strings to canonical disease names using
    config/disease_normalization.json. Returns a comma-separated string of
    canonical disease names.

    Parameters
    ----------
    vaccines_due : Any
        Raw string of comma-separated disease names from input.
    language : str
        Language code (e.g., "en", "fr"). Used for logging.

    Returns
    -------
    str
        Comma-separated string of canonical disease names (English).
        Empty string if input is empty or invalid.
    """
    if not isinstance(vaccines_due, str) or not vaccines_due.strip():
        return ""

    items: List[str] = []
    for token in vaccines_due.split(";"):
        # Normalize: raw input -> canonical disease name
        normalized = normalize_disease(token.strip())
        items.append(normalized)

    # Filter empty items and clean quotes
    return ", ".join(
        item.replace("'", "").replace('"', "") for item in items if item.strip()
    )

normalize_validity_status ¶

normalize_validity_status(raw_status)

Normalize a raw validity token to one of three canonical statuses.

Only the exact casings "valid"/"Valid" and "invalid"/"Invalid" are accepted as known statuses; everything else — typos, alternate casings, empty strings, None, NaN — maps to "unknown". This strict gate prevents ambiguous data from silently influencing validity markers on notices.

Parameters:

Name	Type	Description	Default
`raw_status`	`Any`	Raw value extracted from an IMMS_GIVEN segment, or any value that needs to be coerced to a canonical status. Typically a str, but accepts Any so callers need not guard against None or NaN.	required

Returns:

Type	Description
`str`	One of `"valid"`, `"invalid"`, or `"unknown"`.

Examples:

>>> normalize_validity_status("Valid")
'valid'
>>> normalize_validity_status("")
'unknown'
>>> normalize_validity_status(None)
'unknown'

Source code in pipeline/preprocess.py

def normalize_validity_status(raw_status: Any) -> str:
    """Normalize a raw validity token to one of three canonical statuses.

    Only the exact casings "valid"/"Valid" and "invalid"/"Invalid" are
    accepted as known statuses; everything else — typos, alternate casings,
    empty strings, None, NaN — maps to "unknown". This strict gate prevents
    ambiguous data from silently influencing validity markers on notices.

    Parameters
    ----------
    raw_status : Any
        Raw value extracted from an IMMS_GIVEN segment, or any value that
        needs to be coerced to a canonical status. Typically a str, but
        accepts Any so callers need not guard against None or NaN.

    Returns
    -------
    str
        One of ``"valid"``, ``"invalid"``, or ``"unknown"``.

    Examples
    --------
    >>> normalize_validity_status("Valid")
    'valid'
    >>> normalize_validity_status("")
    'unknown'
    >>> normalize_validity_status(None)
    'unknown'
    """
    status = str(raw_status).strip()
    if status in {"valid", "Valid"}:
        return "valid"
    if status in {"invalid", "Invalid"}:
        return "invalid"
    return "unknown"

collapse_validity_statuses ¶

collapse_validity_statuses(statuses)

Collapse multiple validity statuses using strict precedence.

Precedence:

mixed (if both valid and invalid are present and no unknown)
valid (if at least one valid is present and no unknown)
invalid (if invalid is present and no unknown)
unknown (otherwise)

Source code in pipeline/preprocess.py

def collapse_validity_statuses(statuses: List[Any]) -> str:
    """Collapse multiple validity statuses using strict precedence.

    Precedence:

    1. mixed (if both valid and invalid are present and no unknown)
    2. valid (if at least one valid is present and no unknown)
    3. invalid (if invalid is present and no unknown)
    4. unknown (otherwise)
    """
    normalized = [normalize_validity_status(s) for s in statuses]

    has_valid = "valid" in normalized
    has_invalid = "invalid" in normalized
    has_unknown = "unknown" in normalized

    if has_valid and has_invalid and not has_unknown:
        return "mixed"

    # "All valid" / "all invalid" only when no unknowns are present
    if has_valid and not has_unknown:
        return "valid"
    if has_invalid and not has_unknown:
        return "invalid"

    # anything involving unknown (or empty) stays unknown
    return "unknown"

classify_dataset_validity ¶

classify_dataset_validity(imms_given_series)

Scan all IMMS_GIVEN values and classify dataset-level validity coverage.

Performs a pre-pass over the full dataset before the per-client loop so that a single, accurate dataset-level decision can be made about whether to show warnings, raise errors, or proceed normally. This avoids the unreliable alternative of accumulating per-record "unknown" counts, which can mask a structurally inconsistent dataset.

Called once by build_preprocess_result immediately before the client loop. Its return value determines which branch of the warning/error logic fires; see the behaviour table in the module docstring.

Parameters:

Name	Type	Description	Default
`imms_given_series`	`Series`	The `IMMS_GIVEN` column of the normalized working DataFrame. Each element is a semicolon-delimited string of dose segments such as `"May 1, 2020 - DTaP - Valid; Jun 15, 2021 - MMR"`. NaN values and empty strings are silently skipped.	required

Returns:

Type Description

Literal['all_present', 'all_absent', 'mixed']

"all_present" Every dose segment in the dataset that contains a recognisable date entry also carries a - Valid or - Invalid suffix.

"all_absent" No dose segment carries a validity suffix, or the series contains no recognisable dose entries at all.

"mixed" At least one segment has a suffix and at least one does not. This state causes a ValueError when show_validity_markers is True.

Notes

Pure scan — no side effects, no logging.
O(n × d) where n is the number of rows and d is the average number of dose segments per row; short-circuits as soon as "mixed" is confirmed.

Source code in pipeline/preprocess.py

def classify_dataset_validity(
    imms_given_series: pd.Series,
) -> Literal["all_present", "all_absent", "mixed"]:
    """Scan all IMMS_GIVEN values and classify dataset-level validity coverage.

    Performs a pre-pass over the full dataset before the per-client loop so
    that a single, accurate dataset-level decision can be made about whether
    to show warnings, raise errors, or proceed normally. This avoids the
    unreliable alternative of accumulating per-record ``"unknown"`` counts,
    which can mask a structurally inconsistent dataset.

    Called once by ``build_preprocess_result`` immediately before the client
    loop. Its return value determines which branch of the warning/error logic
    fires; see the behaviour table in the module docstring.

    Parameters
    ----------
    imms_given_series : pd.Series
        The ``IMMS_GIVEN`` column of the normalized working DataFrame.
        Each element is a semicolon-delimited string of dose segments such as
        ``"May 1, 2020 - DTaP - Valid; Jun 15, 2021 - MMR"``.
        NaN values and empty strings are silently skipped.

    Returns
    -------
    Literal["all_present", "all_absent", "mixed"]
        ``"all_present"``
            Every dose segment in the dataset that contains a recognisable
            date entry also carries a ``- Valid`` or ``- Invalid`` suffix.

        ``"all_absent"``
            No dose segment carries a validity suffix, or the series
            contains no recognisable dose entries at all.

        ``"mixed"``
            At least one segment has a suffix and at least one does not.
            This state causes a ``ValueError`` when
            ``show_validity_markers`` is ``True``.

    Notes
    -----
    - Pure scan — no side effects, no logging.
    - O(n × d) where n is the number of rows and d is the average number
      of dose segments per row; short-circuits as soon as ``"mixed"`` is
      confirmed.
    """
    with_validity = re.compile(r" - (?:[Vv]alid|[Ii]nvalid)(?=;|$)")
    dose_entry = re.compile(r"\w{3} \d{1,2}, \d{4}")

    has_with = False
    has_without = False

    for raw in imms_given_series:
        if not isinstance(raw, str) or not raw.strip():
            continue
        for segment in raw.split(";"):
            segment = segment.strip()
            if not dose_entry.search(segment):
                continue
            if with_validity.search(segment):
                has_with = True
            else:
                has_without = True
            if has_with and has_without:
                return "mixed"

    return "all_present" if has_with else "all_absent"

parse_dose_segments ¶

parse_dose_segments(received_agents, replace_unspecified)

Parse an IMMS_GIVEN string into a flat sorted list of individual dose entries.

Extracts individual dose entries from a semicolon-delimited string, normalizes dates to ISO format, normalizes validity to one of "valid" / "invalid" / "unknown", and filters unwanted vaccine names. Unlike the former process_received_agents, this function does not group by date — grouping and disease-expansion are handled by build_received_rows.

Parameters:

Name	Type	Description	Default
`received_agents`	`Any`	Raw IMMS_GIVEN cell value. Must be a non-empty `str` to be parsed; any other type returns `[]`. Expected format per segment: `"MMM D, YYYY - VaccineName"` or `"MMM D, YYYY - VaccineName - Valid\|Invalid"`.	required
`replace_unspecified`	`List[str]`	Vaccine names to silently drop (e.g. `["Not Specified"]`).	required

Returns:

Type	Description
`List[Dict[str, str]]`	Flat list of `{"date_given": str, "vaccine": str, "validity": str}` dicts, sorted ascending by date. Returns `[]` if `received_agents` is not a parseable string or contains no recognisable dose segments after filtering.

Source code in pipeline/preprocess.py

def parse_dose_segments(
    received_agents: Any, replace_unspecified: List[str]
) -> List[Dict[str, str]]:
    """Parse an IMMS_GIVEN string into a flat sorted list of individual dose entries.

    Extracts individual dose entries from a semicolon-delimited string,
    normalizes dates to ISO format, normalizes validity to one of
    ``"valid"`` / ``"invalid"`` / ``"unknown"``, and filters unwanted
    vaccine names.  Unlike the former ``process_received_agents``, this
    function does *not* group by date — grouping and disease-expansion are
    handled by ``build_received_rows``.

    Parameters
    ----------
    received_agents : Any
        Raw IMMS_GIVEN cell value.  Must be a non-empty ``str`` to be
        parsed; any other type returns ``[]``.
        Expected format per segment: ``"MMM D, YYYY - VaccineName"`` or
        ``"MMM D, YYYY - VaccineName - Valid|Invalid"``.
    replace_unspecified : List[str]
        Vaccine names to silently drop (e.g. ``["Not Specified"]``).

    Returns
    -------
    List[Dict[str, str]]
        Flat list of ``{"date_given": str, "vaccine": str, "validity": str}``
        dicts, sorted ascending by date.  Returns ``[]`` if
        ``received_agents`` is not a parseable string or contains no
        recognisable dose segments after filtering.
    """
    if not isinstance(received_agents, str) or not received_agents.strip():
        return []

    pattern = re.compile(
        r"(\w{3} \d{1,2}, \d{4}) - (.*?)(?:\s*-\s*([Vv]alid|[Ii]nvalid))?(?=;|$)"
    )

    rows: List[Dict[str, str]] = []
    for date_str, vaccine, raw_valid in pattern.findall(received_agents):
        vaccine = vaccine.strip()
        vaccine = vaccine.replace("-unspecified", "*").replace(" unspecified", "*")
        if vaccine in replace_unspecified:
            continue
        rows.append({
            "date_given": convert_date_iso(date_str.strip()),
            "vaccine": vaccine,
            "validity": normalize_validity_status(raw_valid),
        })

    rows.sort(key=lambda item: item["date_given"])
    return rows

compute_column_statuses ¶

compute_column_statuses(vaccines, chart_diseases_header)

Compute per-column validity status for a set of vaccine entries.

For each named disease in the header, collects the validity of all vaccines that contribute to that disease and collapses using unknown > mixed > valid > invalid precedence. The "Other" column captures any vaccine that contributes at least one disease not found in the named-disease set.

"mixed" is produced when both "valid" and "invalid" contribute to a column with no "unknown" — meaning different vaccines have conflicting validity for that column on this date.

Parameters:

Name	Type	Description	Default
`vaccines`	`List[Dict[str, Any]]`	Vaccine entries from `_deduplicate_vaccines_for_date`, each with `{"vaccine": str, "diseases": List[str], "validity": str}`.	required
`chart_diseases_header`	`List[str]`	Ordered disease column headers. `"Other"` (if present) acts as a catch-all for unmapped diseases.	required

Returns:

Type	Description
`Dict[str, str]`	Column name → one of `"valid"`, `"invalid"`, `"unknown"`, or `"mixed"`. Only columns with at least one contributing vaccine are included.

Source code in pipeline/preprocess.py

def compute_column_statuses(
    vaccines: List[Dict[str, Any]],
    chart_diseases_header: List[str],
) -> Dict[str, str]:
    """Compute per-column validity status for a set of vaccine entries.

    For each named disease in the header, collects the validity of all
    vaccines that contribute to that disease and collapses using
    ``unknown > mixed > valid > invalid`` precedence.  The ``"Other"``
    column captures any vaccine that contributes at least one disease
    not found in the named-disease set.

    ``"mixed"`` is produced when both ``"valid"`` and ``"invalid"``
    contribute to a column with no ``"unknown"`` — meaning different
    vaccines have conflicting validity for that column on this date.

    Parameters
    ----------
    vaccines : List[Dict[str, Any]]
        Vaccine entries from ``_deduplicate_vaccines_for_date``, each
        with ``{"vaccine": str, "diseases": List[str], "validity": str}``.
    chart_diseases_header : List[str]
        Ordered disease column headers.  ``"Other"`` (if present) acts
        as a catch-all for unmapped diseases.

    Returns
    -------
    Dict[str, str]
        Column name → one of ``"valid"``, ``"invalid"``, ``"unknown"``,
        or ``"mixed"``.  Only columns with at least one contributing
        vaccine are included.
    """
    named = {d for d in chart_diseases_header if d != "Other"}
    has_other_col = "Other" in chart_diseases_header

    column_statuses: Dict[str, List[str]] = {}

    for vax in vaccines:
        for disease in vax["diseases"]:
            if disease in named:
                column_statuses.setdefault(disease, []).append(vax["validity"])
        if has_other_col and any(d not in named for d in vax["diseases"]):
            column_statuses.setdefault("Other", []).append(vax["validity"])

    result: Dict[str, str] = {}
    for col, statuses in column_statuses.items():
        has_unknown = "unknown" in statuses
        has_valid = "valid" in statuses
        has_invalid = "invalid" in statuses
        if has_unknown:
            result[col] = "unknown"
        elif has_valid and has_invalid:
            result[col] = "mixed"
        elif has_valid:
            result[col] = "valid"
        else:
            result[col] = "invalid"

    return result

build_received_rows ¶

build_received_rows(received_agents, replace_unspecified, vaccine_reference, chart_diseases_header)

Parse IMMS_GIVEN into display rows with pre-computed per-column validity.

Orchestrates parse_dose_segments → _deduplicate_vaccines_for_date → _split_into_rows for each administration date. Dates whose vaccines would produce a "mixed" column status are split into separate rows: valid vaccines on the first row, others on subsequent rows. The date_rowspan field carries the row-merge count so that Typst can render a single merged date cell spanning all rows of a date.

Parameters:

Name	Type	Description	Default
`received_agents`	`Any`	Raw IMMS_GIVEN cell value.	required
`replace_unspecified`	`List[str]`	Vaccine names to suppress.	required
`vaccine_reference`	`Dict[str, Any]`	Vaccine-to-disease mapping.	required
`chart_diseases_header`	`List[str]`	Ordered disease column headers (used for column assignment and split ordering).	required

Returns:

Type Description

List[Dict[str, Any]]

Flat list of display rows, each with::

{
    "date_given":   str,            # ISO date
    "date_rowspan": int,            # N on first row, 0 on continuations
    "vaccines":     List[str],      # display vaccine names for this row
    "columns":      Dict[str, str], # column name → validity status
}

Source code in pipeline/preprocess.py

def build_received_rows(
    received_agents: Any,
    replace_unspecified: List[str],
    vaccine_reference: Dict[str, Any],
    chart_diseases_header: List[str],
) -> List[Dict[str, Any]]:
    """Parse IMMS_GIVEN into display rows with pre-computed per-column validity.

    Orchestrates ``parse_dose_segments`` → ``_deduplicate_vaccines_for_date``
    → ``_split_into_rows`` for each administration date.  Dates whose
    vaccines would produce a ``"mixed"`` column status are split into
    separate rows: valid vaccines on the first row, others on subsequent
    rows.  The ``date_rowspan`` field carries the row-merge count so that
    Typst can render a single merged date cell spanning all rows of a date.

    Parameters
    ----------
    received_agents : Any
        Raw IMMS_GIVEN cell value.
    replace_unspecified : List[str]
        Vaccine names to suppress.
    vaccine_reference : Dict[str, Any]
        Vaccine-to-disease mapping.
    chart_diseases_header : List[str]
        Ordered disease column headers (used for column assignment and
        split ordering).

    Returns
    -------
    List[Dict[str, Any]]
        Flat list of display rows, each with::

            {
                "date_given":   str,            # ISO date
                "date_rowspan": int,            # N on first row, 0 on continuations
                "vaccines":     List[str],      # display vaccine names for this row
                "columns":      Dict[str, str], # column name → validity status
            }
    """
    flat = parse_dose_segments(received_agents, replace_unspecified)
    if not flat:
        return []

    by_date: Dict[str, List[Dict[str, str]]] = {}
    for dose in flat:
        by_date.setdefault(dose["date_given"], []).append(
            {"vaccine": dose["vaccine"], "validity": dose["validity"]}
        )

    rows: List[Dict[str, Any]] = []
    for date, doses in by_date.items():
        vaccines = _deduplicate_vaccines_for_date(doses, vaccine_reference)
        date_rows = _split_into_rows(vaccines, chart_diseases_header)
        n = len(date_rows)
        for i, row in enumerate(date_rows):
            rows.append({
                "date_given": date,
                "date_rowspan": n if i == 0 else 0,
                "vaccines": [v["vaccine"] for v in row["vaccines"]],
                "columns": row["columns"],
            })

    return rows

build_preprocess_result ¶

build_preprocess_result(df, language, vaccine_reference, replace_unspecified, config_path=None)

Normalize client data and produce the structured preprocessing artifact.

Orchestrates all per-dataset and per-client normalization: column mapping, sorting, sequence assignment, dataset-level validity classification, age calculation, vaccine history parsing, and disease enrichment. The resulting PreprocessResult is the sole artifact consumed by all downstream pipeline steps.

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	Raw input DataFrame, typically loaded from an Excel or CSV file. Must contain all columns listed in `REQUIRED_COLUMNS` (column names are fuzzy-matched, so spacing and case variants are accepted).	required
`language`	`str`	Language code for this batch (`"en"` or `"fr"`). Stored on every `ClientRecord` and used to format display dates.	required
`vaccine_reference`	`Dict[str, Any]`	Maps vaccine codes to disease names. Passed through to `enrich_grouped_records`.	required
`replace_unspecified`	`List[str]`	Vaccine names to suppress from immunization history. Passed through to `process_received_agents`.	required
`config_path`	`Path`	Path to `parameters.yaml`. Defaults to the repository configuration.	`None`

Returns:

Type	Description
`PreprocessResult`	Contains a list of `ClientRecord` objects (one per input row, sorted by school → last name → first name → client ID) and a list of warning strings for data-quality issues discovered during processing.

Raises:

Type	Description
`ValueError`	If `classify_dataset_validity` returns `"mixed"` and `show_validity_markers` is `True` in `parameters.yaml`. This indicates the source data is structurally inconsistent and cannot be displayed reliably.
`ValueError`	If any required columns are missing from `df` (raised by the underlying `normalize_dataframe` call).

Notes

Reads config_path on every call to obtain date_notice_delivery, chart_diseases_header, preprocess.include_dose, and preprocess.show_validity_markers. If no path is provided, PARAMETERS_PATH (config/parameters.yaml) is used. A missing file is treated as empty config (all defaults apply).
Warns (does not raise) for: missing board name, missing date of birth, duplicate client IDs, all_absent validity when show_validity_markers is True, mixed validity when show_validity_markers is False.

Source code in pipeline/preprocess.py

def build_preprocess_result(
    df: pd.DataFrame,
    language: str,
    vaccine_reference: Dict[str, Any],
    replace_unspecified: List[str],
    config_path: Path | None = None,
) -> PreprocessResult:
    """Normalize client data and produce the structured preprocessing artifact.

    Orchestrates all per-dataset and per-client normalization: column
    mapping, sorting, sequence assignment, dataset-level validity
    classification, age calculation, vaccine history parsing, and disease
    enrichment. The resulting ``PreprocessResult`` is the sole artifact
    consumed by all downstream pipeline steps.

    Parameters
    ----------
    df : pd.DataFrame
        Raw input DataFrame, typically loaded from an Excel or CSV file.
        Must contain all columns listed in ``REQUIRED_COLUMNS`` (column
        names are fuzzy-matched, so spacing and case variants are accepted).
    language : str
        Language code for this batch (``"en"`` or ``"fr"``). Stored on
        every ``ClientRecord`` and used to format display dates.
    vaccine_reference : Dict[str, Any]
        Maps vaccine codes to disease names. Passed through to
        ``enrich_grouped_records``.
    replace_unspecified : List[str]
        Vaccine names to suppress from immunization history. Passed
        through to ``process_received_agents``.
    config_path : Path, optional
        Path to ``parameters.yaml``. Defaults to the repository configuration.

    Returns
    -------
    PreprocessResult
        Contains a list of ``ClientRecord`` objects (one per input row,
        sorted by school → last name → first name → client ID) and a
        list of warning strings for data-quality issues discovered during
        processing.

    Raises
    ------
    ValueError
        If ``classify_dataset_validity`` returns ``"mixed"`` and
        ``show_validity_markers`` is ``True`` in ``parameters.yaml``.
        This indicates the source data is structurally inconsistent and
        cannot be displayed reliably.
    ValueError
        If any required columns are missing from ``df`` (raised by the
        underlying ``normalize_dataframe`` call).

    Notes
    -----
    - Reads ``config_path`` on every call to obtain ``date_notice_delivery``,
      ``chart_diseases_header``, ``preprocess.include_dose``, and
      ``preprocess.show_validity_markers``. If no path is provided,
      ``PARAMETERS_PATH`` (``config/parameters.yaml``) is used. A missing file
      is treated as empty config (all defaults apply).
    - Warns (does not raise) for: missing board name, missing date of
      birth, duplicate client IDs, ``all_absent`` validity when
      ``show_validity_markers`` is ``True``, ``mixed`` validity when
      ``show_validity_markers`` is ``False``.
    """
    warnings: set[str] = set()
    working = normalize_dataframe(df)

    # Load parameters for date_notice_delivery and chart_diseases_header
    parameters_path = config_path if config_path is not None else PARAMETERS_PATH
    params = {}
    if parameters_path.exists():
        params = yaml.safe_load(parameters_path.read_text(encoding="utf-8")) or {}
    date_notice_delivery: Optional[str] = params.get("date_notice_delivery")
    chart_diseases_header: List[str] = params.get("chart_diseases_header", [])
    preprocess_cfg: Dict[str, Any] = params.get("preprocess", {})
    include_dose: bool = preprocess_cfg.get("include_dose", False)
    show_validity_markers: bool = preprocess_cfg.get("show_validity_markers", False)

    working["SCHOOL_ID"] = working.apply(
        lambda row: synthesize_identifier(
            row.get("SCHOOL_ID", ""), row["SCHOOL_NAME"], "sch"
        ),
        axis=1,
    )
    working["BOARD_ID"] = working.apply(
        lambda row: synthesize_identifier(
            row.get("BOARD_ID", ""), row.get("BOARD_NAME", ""), "brd"
        ),
        axis=1,
    )

    if (working["BOARD_NAME"] == "").any():
        affected = (
            working.loc[working["BOARD_NAME"] == "", "SCHOOL_NAME"].unique().tolist()
        )
        warnings.add(
            "Missing board name for: " + ", ".join(sorted(filter(None, affected)))
            if affected
            else "Missing board name for one or more schools."
        )

    sorted_df = working.sort_values(
        by=["SCHOOL_NAME", "LAST_NAME", "FIRST_NAME", "CLIENT_ID"],
        kind="stable",
    ).reset_index(drop=True)
    sorted_df["SEQUENCE"] = [f"{idx + 1:05d}" for idx in range(len(sorted_df))]

    validity_coverage = classify_dataset_validity(sorted_df["IMMS_GIVEN"])
    if validity_coverage == "mixed":
        if show_validity_markers:
            raise ValueError(
                "Dataset contains a mix of records with and without validity indicators. "
                "Cannot display validity markers reliably. "
                "Either fix the source data or set show_validity_markers: false."
            )
        warnings.add(
            "Dataset contains records both with and without validity indicators. "
            "Validity markers are disabled; output is unaffected."
        )
    elif validity_coverage == "all_absent" and show_validity_markers:
        warnings.add(
            "show_validity_markers is enabled but no validity data was detected in the dataset. "
            "Default indicators will be used."
        )

    clients: List[ClientRecord] = []
    for row in sorted_df.itertuples(index=False):
        client_id = str(row.CLIENT_ID)  # type: ignore[attr-defined]
        sequence = row.SEQUENCE  # type: ignore[attr-defined]
        dob_iso = (
            row.DATE_OF_BIRTH.strftime("%Y-%m-%d")  # type: ignore[attr-defined]
            if pd.notna(row.DATE_OF_BIRTH)  # type: ignore[attr-defined]
            else None
        )
        if dob_iso is None:
            warnings.add(f"Missing date of birth for client {client_id}")

        language_enum = Language.from_string(language)
        formatted_dob = (
            convert_date_string(dob_iso, locale="fr")
            if language_enum == Language.FRENCH and dob_iso
            else (convert_date_string(dob_iso, locale="en") if dob_iso else None)
        )
        vaccines_due = process_vaccines_due(row.OVERDUE_DISEASE, language)  # type: ignore[attr-defined]
        vaccines_due_list = [
            item.strip() for item in vaccines_due.split(",") if item.strip()
        ]
        for item in vaccines_due_list:
            disease, dose = split_vaccine_due_entry(item)
            if dose == "":
                warnings.add(
                    f"Blank overdue dose number for client {client_id}: {disease}. "
                    "Displaying disease without a dose number."
                )
        if include_dose:
            vaccines_due_list = format_vaccine_due_list(vaccines_due_list)
        else:
            vaccines_due_list = hide_vaccine_due_doses(vaccines_due_list)
        received = build_received_rows(
            row.IMMS_GIVEN,  # type: ignore[attr-defined]
            replace_unspecified,
            vaccine_reference,
            chart_diseases_header,
        )
        postal_code = row.POSTAL_CODE if row.POSTAL_CODE else "Not provided"  # type: ignore[attr-defined]
        address_line = " ".join(
            filter(None, [row.STREET_ADDRESS_LINE_1, row.STREET_ADDRESS_LINE_2])  # type: ignore[attr-defined]
        ).strip()

        if not pd.isna(row.AGE):  # type: ignore[attr-defined]
            over_16 = bool(row.AGE >= 16)  # type: ignore[attr-defined]
        elif dob_iso and date_notice_delivery:
            over_16 = over_16_check(dob_iso, date_notice_delivery)
        else:
            over_16 = False

        person = {
            "first_name": row.FIRST_NAME or "",  # type: ignore[attr-defined]
            "last_name": row.LAST_NAME or "",  # type: ignore[attr-defined]
            "date_of_birth": dob_iso or "",
            "date_of_birth_display": formatted_dob or "",
            "date_of_birth_iso": dob_iso or "",
            "age": str(row.AGE) if not pd.isna(row.AGE) else "",  # type: ignore[attr-defined]
            "over_16": over_16,
        }

        school = {
            "name": row.SCHOOL_NAME,  # type: ignore[attr-defined]
            "id": row.SCHOOL_ID,  # type: ignore[attr-defined]
        }

        board = {
            "name": row.BOARD_NAME or "",  # type: ignore[attr-defined]
            "id": row.BOARD_ID,  # type: ignore[attr-defined]
        }

        contact = {
            "street": address_line,
            "city": row.CITY,  # type: ignore[attr-defined]
            "province": row.PROVINCE,  # type: ignore[attr-defined]
            "postal_code": postal_code,
        }

        client = ClientRecord(
            sequence=sequence,
            client_id=client_id,
            language=language,
            person=person,
            school=school,
            board=board,
            contact=contact,
            vaccines_due=vaccines_due if vaccines_due else None,
            vaccines_due_list=vaccines_due_list if vaccines_due_list else None,
            received=received if received else None,
            metadata={
                "unique_id": row.UNIQUE_ID or None,  # type: ignore[attr-defined]
            },
        )

        clients.append(client)

    # Detect and warn about duplicate client IDs
    client_id_counts: dict[str, int] = {}
    for client in clients:
        client_id_counts[client.client_id] = (
            client_id_counts.get(client.client_id, 0) + 1
        )

    duplicates = {cid: count for cid, count in client_id_counts.items() if count > 1}
    if duplicates:
        for cid in sorted(duplicates.keys()):
            warnings.add(
                f"Duplicate client ID '{cid}' found {duplicates[cid]} times. "
                "Later records will overwrite earlier ones in generated notices."
            )

    return PreprocessResult(
        clients=clients,
        warnings=list(warnings),
    )

run_phix_validation ¶

run_phix_validation(df, output_dir)

Validate school names against the PHIX mapping file.

Reads phix_validation config from parameters.yaml. Returns the DataFrame unchanged (with no warnings) when validation is disabled or mapping_file is not set.

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	Normalized DataFrame (output of `check_addresses_complete`).	required
`output_dir`	`Path`	Directory where result CSVs (`phix_exact.csv`, etc.) are written.	required

Returns:

Type	Description
`tuple[DataFrame, list[str]]`	Enriched DataFrame and a (possibly empty) list of warning strings.

Source code in pipeline/preprocess.py

def run_phix_validation(
    df: pd.DataFrame,
    output_dir: Path,
) -> tuple[pd.DataFrame, list[str]]:
    """Validate school names against the PHIX mapping file.

    Reads ``phix_validation`` config from ``parameters.yaml``. Returns the
    DataFrame unchanged (with no warnings) when validation is disabled or
    ``mapping_file`` is not set.

    Parameters
    ----------
    df:
        Normalized DataFrame (output of ``check_addresses_complete``).
    output_dir:
        Directory where result CSVs (``phix_exact.csv``, etc.) are written.

    Returns
    -------
    tuple[DataFrame, list[str]]
        Enriched DataFrame and a (possibly empty) list of warning strings.
    """
    config = yaml.safe_load(PARAMETERS_PATH.read_text(encoding="utf-8")) or {}
    phix_config = config.get("phix_validation", {})

    if not phix_config.get("enabled", False):
        return df, []

    mapping_file = phix_config.get("mapping_file", "")
    if not mapping_file:
        LOG.warning("phix_validation.enabled is true but mapping_file is not set.")
        return df, []

    target_phu = phix_config.get("target_phu", "")
    if not target_phu:
        LOG.warning("phix_validation.target_phu is not set — skipping PHIX validation.")
        return df, []

    from . import validate_phix  # local import avoids circular dependency at module load

    mapping_path = Path(mapping_file)
    if not mapping_path.is_absolute():
        mapping_path = (SCRIPT_DIR.parent / mapping_file).resolve()

    return validate_phix.validate_schools(
        df=df,
        mapping_path=mapping_path,
        target_phu=target_phu,
        output_dir=output_dir,
        unmatched_behavior=phix_config.get("unmatched_behavior", "warn"),
        column_prefix=phix_config.get("column_prefix", "PHIX_"),
    )

write_artifact ¶

write_artifact(output_dir, language, run_id, result)

Write preprocessed result to JSON artifact file.

Source code in pipeline/preprocess.py

def write_artifact(
    output_dir: Path, language: str, run_id: str, result: PreprocessResult
) -> Path:
    """Write preprocessed result to JSON artifact file."""
    output_dir.mkdir(parents=True, exist_ok=True)

    # Create ArtifactPayload with rich metadata
    artifact_payload = ArtifactPayload(
        run_id=run_id,
        language=language,
        clients=result.clients,
        warnings=result.warnings,
        created_at=datetime.now(timezone.utc).isoformat(),
        total_clients=len(result.clients),
    )

    # Serialize to JSON (clients are dataclasses, so convert to dict)
    payload_dict = {
        "run_id": artifact_payload.run_id,
        "language": artifact_payload.language,
        "created_at": artifact_payload.created_at,
        "total_clients": artifact_payload.total_clients,
        "warnings": artifact_payload.warnings,
        "clients": [
            {
                "sequence": client.sequence,
                "client_id": client.client_id,
                "language": client.language,
                "person": {
                    "first_name": client.person["first_name"],
                    "last_name": client.person["last_name"],
                    "date_of_birth": client.person["date_of_birth"],
                    "date_of_birth_display": client.person["date_of_birth_display"],
                    "date_of_birth_iso": client.person["date_of_birth_iso"],
                    "age": client.person["age"],
                    "over_16": client.person["over_16"],
                },
                "school": {
                    "name": client.school["name"],
                    "id": client.school["id"],
                },
                "board": {
                    "name": client.board["name"],
                    "id": client.board["id"],
                },
                "contact": {
                    "street": client.contact["street"],
                    "city": client.contact["city"],
                    "province": client.contact["province"],
                    "postal_code": client.contact["postal_code"],
                },
                "vaccines_due": client.vaccines_due,
                "vaccines_due_list": client.vaccines_due_list or [],
                "received": client.received or [],
                "metadata": client.metadata,
            }
            for client in artifact_payload.clients
        ],
    }

    artifact_path = output_dir / f"preprocessed_clients_{run_id}.json"
    artifact_path.write_text(json.dumps(payload_dict, indent=2), encoding="utf-8")
    LOG.info("Wrote normalized artifact to %s", artifact_path)
    return artifact_path

Step 3 — QR Code Generation¶

pipeline.generate_qr_codes ¶

Generate QR code PNG files from preprocessed client artifact.

This module creates QR code images for each client in the preprocessed artifact. QR payloads are generated from template strings defined in parameters.yaml and rendered as PNG files in the output artifacts directory.

The QR code generation step is optional and can be skipped via the qr.enabled configuration setting.

Input Contract:

Reads preprocessed artifact JSON (created by preprocess step)
Assumes artifact contains valid client records with required fields
Assumes qr.enabled=true and qr.payload_template defined in config (if QR generation requested)

Output Contract:

Writes QR code PNG files to output/artifacts/qr_codes/
Returns list of successfully generated QR file paths
Per-client errors are logged and skipped (optional feature; doesn't halt pipeline)

Error Handling:

Configuration errors (missing template) raise immediately (infrastructure error)
Per-client failures (invalid data) log warning and continue (data error in optional feature)
This strategy allows partial success; some clients may not have QR codes

Validation Contract:

What this module validates:

Artifact file exists and is valid JSON (validation in read_preprocessed_artifact())
QR code generation is enabled in config (qr.enabled=true)
Payload template is defined if QR generation is enabled
Payload template format is valid (has valid placeholders)
QR code can be rendered as PNG (infrastructure check)

What this module assumes (validated upstream):

Artifact JSON structure is valid (validated by preprocessing step)
Client records have all required fields (validated by preprocessing step)
Output directory can be created (general I/O)

Per-client failures (invalid client data, template rendering errors) are logged and skipped (intentional for optional feature). Some clients may lack QR codes.

encode_qr_payload_url ¶

encode_qr_payload_url(payload_url)

URL-encode parameter values in a QR payload URL.

This function preserves the URL structure (scheme, domain, path, parameter delimiters) while properly encoding parameter values to handle special characters like spaces, accents, and special symbols.

Parameters:

Name	Type	Description	Default
`payload_url`	`str`	A URL string like "https://example.com/surveys/?s=ABC&qrfn=First Name&qrln=Last Name"	required

Returns:

Type	Description
`str`	The same URL with parameter values properly percent-encoded (e.g., spaces → %20)

Source code in pipeline/generate_qr_codes.py

def encode_qr_payload_url(payload_url: str) -> str:
    """URL-encode parameter values in a QR payload URL.

    This function preserves the URL structure (scheme, domain, path, parameter
    delimiters) while properly encoding parameter values to handle special
    characters like spaces, accents, and special symbols.

    Parameters
    ----------
    payload_url : str
        A URL string like "https://example.com/surveys/?s=ABC&qrfn=First Name&qrln=Last Name"

    Returns
    -------
    str
        The same URL with parameter values properly percent-encoded (e.g., spaces → %20)
    """
    # Split URL into scheme+domain+path and query string
    if "?" not in payload_url:
        return payload_url

    base_url, query_string = payload_url.split("?", 1)

    # Parse and re-encode each parameter
    encoded_params = []
    for param in query_string.split("&"):
        if "=" in param:
            key, value = param.split("=", 1)
            # URL-encode the value (preserve unreserved characters)
            encoded_value = quote(value, safe="")
            encoded_params.append(f"{key}={encoded_value}")
        else:
            # No value, just preserve the key
            encoded_params.append(param)

    # Reconstruct the URL
    return base_url + "?" + "&".join(encoded_params)

generate_qr_code ¶

generate_qr_code(data, output_dir, *, filename=None)

Generate a monochrome QR code PNG and return the saved path.

Parameters:

Name	Type	Description	Default
`data`	`str`	The string payload to encode inside the QR code.	required
`output_dir`	`Path`	Directory where the QR image should be saved. The directory is created if it does not already exist.	required
`filename`	`Optional[str]`	Optional file name (including extension) for the resulting PNG. When omitted a deterministic name derived from the payload hash is used.	`None`

Returns:

Type	Description
`Path`	Absolute path to the generated PNG file.

Source code in pipeline/generate_qr_codes.py

def generate_qr_code(
    data: str,
    output_dir: Path,
    *,
    filename: Optional[str] = None,
) -> Path:
    """Generate a monochrome QR code PNG and return the saved path.

    Parameters
    ----------
    data:
        The string payload to encode inside the QR code.
    output_dir:
        Directory where the QR image should be saved. The directory is created
        if it does not already exist.
    filename:
        Optional file name (including extension) for the resulting PNG. When
        omitted a deterministic name derived from the payload hash is used.

    Returns
    -------
    Path
        Absolute path to the generated PNG file.
    """

    if qrcode is None or Image is None:  # pragma: no cover - exercised in optional envs
        raise RuntimeError(
            "QR code generation requires the 'qrcode' and 'pillow' packages. "
            "Install them via 'uv sync' before enabling QR payloads."
        )

    output_dir.mkdir(parents=True, exist_ok=True)

    qr = qrcode.QRCode(
        version=1,
        error_correction=qrcode_constants.ERROR_CORRECT_L,
        box_size=10,
        border=4,
    )
    qr.add_data(data)
    qr.make(fit=True)

    image = qr.make_image(fill_color="black", back_color="white")
    pil_image = getattr(image, "get_image", lambda: image)()

    # Convert to 1-bit black/white without dithering to keep crisp edges.
    # NONE (0) means no dithering
    pil_bitmap = pil_image.convert("1", dither=0)

    if not filename:
        digest = hashlib.sha1(data.encode("utf-8")).hexdigest()[:12]
        filename = f"qr_{digest}.png"

    target_path = output_dir / filename
    pil_bitmap.save(target_path, format="PNG", bits=1)
    return target_path

read_preprocessed_artifact ¶

read_preprocessed_artifact(path)

Read preprocessed client artifact from JSON.

Input Contract: Assumes artifact was created by preprocessing step and exists on disk. Does not validate artifact schema; assumes preprocessing has already validated client data structure.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the preprocessed JSON artifact file.	required

Returns:

Type	Description
`Dict[str, Any]`	Parsed artifact dict with clients and metadata.

Raises:

Type	Description
`FileNotFoundError`	If artifact file does not exist.
`JSONDecodeError`	If artifact is not valid JSON.

Source code in pipeline/generate_qr_codes.py

def read_preprocessed_artifact(path: Path) -> Dict[str, Any]:
    """Read preprocessed client artifact from JSON.

    **Input Contract:** Assumes artifact was created by preprocessing step and
    exists on disk. Does not validate artifact schema; assumes preprocessing
    has already validated client data structure.

    Parameters
    ----------
    path : Path
        Path to the preprocessed JSON artifact file.

    Returns
    -------
    Dict[str, Any]
        Parsed artifact dict with clients and metadata.

    Raises
    ------
    FileNotFoundError
        If artifact file does not exist.
    json.JSONDecodeError
        If artifact is not valid JSON.
    """
    if not path.exists():
        raise FileNotFoundError(
            f"Preprocessed artifact not found: {path}. "
            "Ensure preprocessing step has completed."
        )
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        return payload
    except json.JSONDecodeError as exc:
        raise ValueError(f"Preprocessed artifact is not valid JSON: {path}") from exc

load_qr_settings ¶

load_qr_settings(config_path=None)

Load QR payload template from parameters.yaml file.

Raises ValueError if qr.payload_template is not specified in the configuration.

Returns: QR payload template string

Source code in pipeline/generate_qr_codes.py

def load_qr_settings(config_path: Path | None = None) -> str:
    """Load QR payload template from parameters.yaml file.

    Raises ValueError if qr.payload_template is not specified in the configuration.

    Returns:
        QR payload template string
    """
    if config_path is None:
        config_path = PARAMETERS_PATH

    if not config_path.exists():
        raise FileNotFoundError(
            f"QR code generation enabled but configuration file not found: {config_path}"
        )

    params = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
    config_data = params.get("qr", {})

    template_config = config_data.get("payload_template")
    if not template_config:
        raise ValueError(
            "QR code generation is enabled but qr.payload_template is not specified in config. "
            "Please define qr.payload_template in parameters.yaml or set qr.enabled to false."
        )

    if not isinstance(template_config, str):
        raise ValueError(
            f"qr.payload_template must be a string, got {type(template_config).__name__}"
        )

    payload_template = template_config

    return payload_template

generate_qr_codes ¶

generate_qr_codes(artifact_path, output_dir, config_path=None)

Generate QR code PNG files from preprocessed artifact.

Parameters:

Name	Type	Description	Default
`artifact_path`	`Path`	Path to the preprocessed JSON artifact.	required
`output_dir`	`Path`	Directory to write QR code PNG files.	required
`config_path`	`Path`	Path to parameters.yaml. If not provided, uses default location.	`None`

Returns:

Type	Description
`List[Path]`	List of generated QR code PNG file paths.

Source code in pipeline/generate_qr_codes.py

def generate_qr_codes(
    artifact_path: Path,
    output_dir: Path,
    config_path: Path | None = None,
) -> List[Path]:
    """Generate QR code PNG files from preprocessed artifact.

    Parameters
    ----------
    artifact_path : Path
        Path to the preprocessed JSON artifact.
    output_dir : Path
        Directory to write QR code PNG files.
    config_path : Path, optional
        Path to parameters.yaml. If not provided, uses default location.

    Returns
    -------
    List[Path]
        List of generated QR code PNG file paths.
    """
    if config_path is None:
        config_path = PARAMETERS_PATH

    # Load QR configuration
    config = load_config(config_path)
    qr_config = config.get("qr", {})
    qr_enabled = qr_config.get("enabled", True)

    if not qr_enabled:
        LOG.info("QR code generation disabled in configuration")
        return []

    # Read artifact
    artifact = read_preprocessed_artifact(artifact_path)
    clients = artifact.get("clients", [])

    if not clients:
        LOG.info("No clients in artifact")
        return []

    # Load QR settings (will raise ValueError if template not specified)
    try:
        payload_template = load_qr_settings(config_path)
    except (FileNotFoundError, ValueError) as exc:
        raise RuntimeError(f"Cannot generate QR codes: {exc}") from exc

    # Ensure output directory exists
    qr_output_dir = output_dir / "qr_codes"
    qr_output_dir.mkdir(parents=True, exist_ok=True)

    generated_files: List[Path] = []

    # Generate QR code for each client
    for client in clients:
        client_id = client.get("client_id")
        # Build context directly from client data using shared helper
        qr_context = build_client_context(client)

        # Generate payload (template is now required)
        try:
            qr_payload = validate_and_format_template(
                payload_template,
                qr_context,
                allowed_fields=SUPPORTED_QR_TEMPLATE_FIELDS,
            )
            # Properly URL-encode the payload for QR code
            qr_payload = encode_qr_payload_url(qr_payload)
        except (KeyError, ValueError) as exc:
            LOG.warning(
                "Could not format QR payload for client %s: %s",
                client_id,
                exc,
            )
            continue

        # Generate PNG
        try:
            sequence = client.get("sequence")
            qr_path = generate_qr_code(
                qr_payload,
                qr_output_dir,
                filename=f"qr_code_{sequence}_{client_id}.png",
            )
            generated_files.append(qr_path)

            # Store the QR payload in the client record
            try:
                rel_path = qr_path.relative_to(ROOT_DIR)
            except ValueError:
                # For testing: if path is not under ROOT_DIR, use as-is
                rel_path = qr_path
            client["qr"] = {
                "payload": qr_payload,
                "filename": f"qr_code_{sequence}_{client_id}.png",
                "path": str(rel_path),
            }

            LOG.info("Generated QR code for client %s: %s", client_id, qr_path)
        except RuntimeError as exc:
            LOG.warning(
                "Could not generate QR code for client %s: %s",
                client_id,
                exc,
            )

    # Write updated artifact back to disk with qr fields
    if generated_files:
        artifact_path.write_text(
            json.dumps(artifact, indent=2, ensure_ascii=False), encoding="utf-8"
        )
        LOG.info("Updated artifact with QR codes: %s", artifact_path)

    return generated_files

main ¶

main(artifact_path, output_dir, config_path=None)

Main entry point for QR code generation.

Parameters:

Name	Type	Description	Default
`artifact_path`	`Path`	Path to the preprocessed JSON artifact.	required
`output_dir`	`Path`	Directory to write QR code PNG files.	required
`config_path`	`Path`	Path to parameters.yaml configuration file.	`None`

Returns:

Type	Description
`int`	Number of QR codes generated.

Source code in pipeline/generate_qr_codes.py

def main(
    artifact_path: Path,
    output_dir: Path,
    config_path: Path | None = None,
) -> int:
    """Main entry point for QR code generation.

    Parameters
    ----------
    artifact_path : Path
        Path to the preprocessed JSON artifact.
    output_dir : Path
        Directory to write QR code PNG files.
    config_path : Path, optional
        Path to parameters.yaml configuration file.

    Returns
    -------
    int
        Number of QR codes generated.
    """
    generated = generate_qr_codes(artifact_path, output_dir, config_path)
    if generated:
        print(
            f"Generated {len(generated)} QR code PNG file(s) in {output_dir}/qr_codes/"
        )
    return len(generated)

Step 4 — Notice Generation¶

pipeline.generate_notices ¶

Generate per-client Typst notices from the normalized preprocessing artifact.

This module consumes the JSON artifact emitted by preprocess.py and generates per-client Typst templates for notice rendering.

Input Contract:

Reads preprocessed artifact JSON (created by preprocess step)
Assumes artifact contains valid client records with all required fields
Assumes language validation already occurred at CLI entry point

Output Contract:

Writes per-client Typst template files to output/artifacts/typst/
Returns list of successfully generated .typ file paths
All clients must succeed; fails immediately on first error (critical feature)

Error Handling:

Client data errors raise immediately (cannot produce incomplete output)
Infrastructure errors (missing paths) raise immediately
Invalid language enum raises immediately (should never occur if upstream validates)
No per-client recovery; fail-fast approach ensures deterministic output

Validation Contract:

What this module validates:

Artifact language matches all client languages (fail-fast if mismatch)

What this module assumes (validated upstream):

Artifact file exists and is valid JSON (validated by read_artifact())
Language code is valid (validated at CLI by argparse choices)
Client records have all required fields (validated by preprocessing step)
File paths exist (output_dir, logo_path, signature_path)

Functions with special validation notes:

render_notice(): Calls Language.from_string() on client.language to convert string to enum; this adds a second validation layer (redundant but safe)
get_language_renderer(): Assumes language enum is valid; no defensive check (language validated upstream via CLI choices + Language.from_string())

load_template_module ¶

load_template_module(template_dir, language_code)

Dynamically load a template module from specified directory.

Loads a language-specific template module at runtime from a custom directory. This enables PHU-specific template customization without code changes.

Validation Contract:

Template file must exist at {template_dir}/{language_code}_template.py
Module must define render_notice() function
Raises immediately if file missing or render_notice() not found

Parameters:

Name	Type	Description	Default
`template_dir`	`Path`	Directory containing template modules (e.g., templates/ or phu_templates/my_phu/)	required
`language_code`	`str`	Two-character ISO language code (e.g., "en", "fr")	required

Returns:

Type	Description
`module`	Loaded Python module with render_notice() function

Raises:

Type	Description
`FileNotFoundError`	If template file doesn't exist at expected path
`ImportError`	If module cannot be loaded
`AttributeError`	If module doesn't define render_notice() function

Examples:

>>> module = load_template_module(Path("templates"), "en")
>>> module.render_notice(context, logo_path="/logo.png", signature_path="/sig.png")

Source code in pipeline/generate_notices.py

def load_template_module(template_dir: Path, language_code: str):
    """Dynamically load a template module from specified directory.

    Loads a language-specific template module at runtime from a custom directory.
    This enables PHU-specific template customization without code changes.

    **Validation Contract:**

    - Template file must exist at {template_dir}/{language_code}_template.py
    - Module must define render_notice() function
    - Raises immediately if file missing or render_notice() not found

    Parameters
    ----------
    template_dir : Path
        Directory containing template modules (e.g., templates/ or phu_templates/my_phu/)
    language_code : str
        Two-character ISO language code (e.g., "en", "fr")

    Returns
    -------
    module
        Loaded Python module with render_notice() function

    Raises
    ------
    FileNotFoundError
        If template file doesn't exist at expected path
    ImportError
        If module cannot be loaded
    AttributeError
        If module doesn't define render_notice() function

    Examples
    --------
    >>> module = load_template_module(Path("templates"), "en")
    >>> module.render_notice(context, logo_path="/logo.png", signature_path="/sig.png")
    """
    module_name = f"{language_code}_template"
    module_path = template_dir / f"{module_name}.py"

    # Validate file exists
    if not module_path.exists():
        raise FileNotFoundError(
            f"Template module not found: {module_path}. "
            f"Expected {module_name}.py in {template_dir}"
        )

    # Load module dynamically
    spec = importlib.util.spec_from_file_location(module_name, module_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Could not load template module: {module_path}")

    module = importlib.util.module_from_spec(spec)

    # Register in sys.modules to prevent duplicate loads
    sys.modules[f"_dynamic_{module_name}"] = module
    spec.loader.exec_module(module)

    # Validate render_notice() exists
    if not hasattr(module, "render_notice"):
        raise AttributeError(
            f"Template module {module_name} must define render_notice() function. "
            f"Check {module_path} and ensure it implements the required interface."
        )

    return module

build_language_renderers ¶

build_language_renderers(template_dir)

Build renderer dictionary from templates in specified directory.

Discovers and loads all available language template modules from the given directory, building a mapping of language codes to their render_notice functions.

Validation Contract:

Only languages with corresponding template files are included
Each available template must have valid render_notice() function
Raises immediately if any template file exists but is invalid
Does NOT require all Language enum values to be present
Later validation ensures requested language is available when needed

Parameters:

Name	Type	Description	Default
`template_dir`	`Path`	Directory containing language template modules	required

Returns:

Type	Description
`dict`	Mapping of language codes (str) to render_notice functions (callable) Format: {"en": , "fr": , ...} May contain subset of all languages; only includes available templates

Raises:

Type	Description
`AttributeError`	If a template file exists but doesn't define render_notice()

Examples:

>>> renderers = build_language_renderers(Path("templates"))
>>> renderers["en"](context, logo_path="/logo.png", signature_path="/sig.png")

>>> # PHU providing only English
>>> renderers = build_language_renderers(Path("phu_templates/my_phu"))
>>> renderers  # May only contain {"en": <function>}

Source code in pipeline/generate_notices.py

def build_language_renderers(template_dir: Path) -> dict:
    """Build renderer dictionary from templates in specified directory.

    Discovers and loads all available language template modules from the given directory,
    building a mapping of language codes to their render_notice functions.

    **Validation Contract:**

    - Only languages with corresponding template files are included
    - Each available template must have valid render_notice() function
    - Raises immediately if any template file exists but is invalid
    - Does NOT require all Language enum values to be present
    - Later validation ensures requested language is available when needed

    Parameters
    ----------
    template_dir : Path
        Directory containing language template modules

    Returns
    -------
    dict
        Mapping of language codes (str) to render_notice functions (callable)
        Format: {"en": <function>, "fr": <function>, ...}
        May contain subset of all languages; only includes available templates

    Raises
    ------
    AttributeError
        If a template file exists but doesn't define render_notice()

    Examples
    --------
    >>> renderers = build_language_renderers(Path("templates"))
    >>> renderers["en"](context, logo_path="/logo.png", signature_path="/sig.png")

    >>> # PHU providing only English
    >>> renderers = build_language_renderers(Path("phu_templates/my_phu"))
    >>> renderers  # May only contain {"en": <function>}
    """
    renderers = {}
    for lang in Language:
        module_path = template_dir / f"{lang.value}_template.py"
        # Only load if template file exists
        if module_path.exists():
            module = load_template_module(template_dir, lang.value)
            renderers[lang.value] = module.render_notice
    return renderers

get_language_renderer ¶

get_language_renderer(language, renderers)

Get template renderer for given language from provided renderer dict.

Maps Language enum values to their corresponding template rendering functions from a dynamically-built renderer dictionary. This provides a single dispatch point for template selection with runtime-configurable template sources.

Validation Contract: Assumes language is a valid Language enum (validated upstream at CLI entry point via argparse choices, and again by Language.from_string() before calling this function). Checks that language is available in renderers dict; raises with helpful error if template for requested language is not available.

Parameters:

Name	Type	Description	Default
`language`	`Language`	Language enum value (guaranteed to be valid from Language enum).	required
`renderers`	`dict`	Mapping of language codes to render_notice functions, built by build_language_renderers(). May only contain subset of all languages.	required

Returns:

Type	Description
`callable`	Template rendering function for the language.

Raises:

Type	Description
`FileNotFoundError`	If requested language template is not available in renderers dict. Provides helpful message listing available languages.

Examples:

>>> renderers = build_language_renderers(Path("templates"))
>>> renderer = get_language_renderer(Language.ENGLISH, renderers)
>>> # renderer is now the render_notice function from en_template

Source code in pipeline/generate_notices.py

def get_language_renderer(language: Language, renderers: dict):
    """Get template renderer for given language from provided renderer dict.

    Maps Language enum values to their corresponding template rendering functions
    from a dynamically-built renderer dictionary. This provides a single dispatch
    point for template selection with runtime-configurable template sources.

    **Validation Contract:** Assumes language is a valid Language enum (validated
    upstream at CLI entry point via argparse choices, and again by Language.from_string()
    before calling this function). Checks that language is available in renderers dict;
    raises with helpful error if template for requested language is not available.

    Parameters
    ----------
    language : Language
        Language enum value (guaranteed to be valid from Language enum).
    renderers : dict
        Mapping of language codes to render_notice functions, built by
        build_language_renderers(). May only contain subset of all languages.

    Returns
    -------
    callable
        Template rendering function for the language.

    Raises
    ------
    FileNotFoundError
        If requested language template is not available in renderers dict.
        Provides helpful message listing available languages.

    Examples
    --------
    >>> renderers = build_language_renderers(Path("templates"))
    >>> renderer = get_language_renderer(Language.ENGLISH, renderers)
    >>> # renderer is now the render_notice function from en_template
    """
    if language.value not in renderers:
        available = ", ".join(sorted(renderers.keys())) if renderers else "none"
        raise FileNotFoundError(
            f"Template not available for language: {language.value}\n"
            f"Available languages: {available}\n"
            f"Ensure your template directory contains {language.value}_template.py"
        )
    return renderers[language.value]

read_artifact ¶

read_artifact(path)

Read and deserialize the preprocessed artifact JSON.

Input Contract: Assumes artifact was created by preprocessing step and contains valid client records. Does not validate client schema; relies on preprocessing to have ensured data quality.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Path to the preprocessed artifact JSON file.	required

Returns:

Type	Description
`ArtifactPayload`	Parsed artifact with clients and metadata.

Raises:

Type	Description
`FileNotFoundError`	If artifact file does not exist.
`JSONDecodeError`	If artifact is not valid JSON.
`KeyError`	If artifact is missing required fields.

Source code in pipeline/generate_notices.py

def read_artifact(path: Path) -> ArtifactPayload:
    """Read and deserialize the preprocessed artifact JSON.

    **Input Contract:** Assumes artifact was created by preprocessing step and
    contains valid client records. Does not validate client schema; relies on
    preprocessing to have ensured data quality.

    Parameters
    ----------
    path : Path
        Path to the preprocessed artifact JSON file.

    Returns
    -------
    ArtifactPayload
        Parsed artifact with clients and metadata.

    Raises
    ------
    FileNotFoundError
        If artifact file does not exist.
    json.JSONDecodeError
        If artifact is not valid JSON.
    KeyError
        If artifact is missing required fields.
    """
    if not path.exists():
        raise FileNotFoundError(
            f"Preprocessed artifact not found: {path}. "
            "Ensure preprocessing step has completed."
        )

    try:
        payload_dict = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ValueError(f"Preprocessed artifact is not valid JSON: {path}") from exc

    clients = []

    for client_dict in payload_dict["clients"]:
        client = deserialize_client_record(client_dict)
        clients.append(client)

    return ArtifactPayload(
        run_id=payload_dict["run_id"],
        language=payload_dict["language"],
        clients=clients,
        warnings=payload_dict.get("warnings", []),
        created_at=payload_dict.get("created_at", ""),
        total_clients=payload_dict.get("total_clients", len(clients)),
    )

escape_string ¶

escape_string(value)

Escape special characters in a string for Typst template output.

Module-internal helper for to_typ_value(). Escapes backslashes, quotes, and newlines to ensure the string can be safely embedded in a Typst template.

Parameters:

Name	Type	Description	Default
`value`	`str`	String to escape.	required

Returns:

Type	Description
`str`	Escaped string safe for Typst embedding.

Source code in pipeline/generate_notices.py

def escape_string(value: str) -> str:
    """Escape special characters in a string for Typst template output.

    Module-internal helper for to_typ_value(). Escapes backslashes, quotes,
    and newlines to ensure the string can be safely embedded in a Typst template.

    Parameters
    ----------
    value : str
        String to escape.

    Returns
    -------
    str
        Escaped string safe for Typst embedding.
    """
    return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")

to_typ_value ¶

to_typ_value(value)

Convert a Python value to its Typst template representation.

Module-internal helper for building template contexts. Handles strings (with escaping), booleans, None, numbers, sequences (tuples), and mappings (dicts) by converting them to Typst syntax.

Parameters:

Name	Type	Description	Default
`value`	`Any`	Python value to convert.	required

Returns:

Type	Description
`str`	Typst-compatible representation of the value.

Raises:

Type	Description
`TypeError`	If value type is not supported.

Examples:

>>> to_typ_value("hello")
'"hello"'
>>> to_typ_value(True)
'true'
>>> to_typ_value([1, 2, 3])
'(1, 2, 3)'

Source code in pipeline/generate_notices.py

def to_typ_value(value) -> str:
    """Convert a Python value to its Typst template representation.

    Module-internal helper for building template contexts. Handles strings
    (with escaping), booleans, None, numbers, sequences (tuples), and mappings
    (dicts) by converting them to Typst syntax.

    Parameters
    ----------
    value : Any
        Python value to convert.

    Returns
    -------
    str
        Typst-compatible representation of the value.

    Raises
    ------
    TypeError
        If value type is not supported.

    Examples
    --------
    >>> to_typ_value("hello")
    '"hello"'
    >>> to_typ_value(True)
    'true'
    >>> to_typ_value([1, 2, 3])
    '(1, 2, 3)'
    """
    if isinstance(value, str):
        return f'"{escape_string(value)}"'
    if isinstance(value, bool):
        return "true" if value else "false"
    if value is None:
        return "none"
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
        items = [to_typ_value(item) for item in value]
        if len(items) == 1:
            inner = f"{items[0]},"
        else:
            inner = ", ".join(items)
        return f"({inner})"
    if isinstance(value, Mapping):
        items = ", ".join(f"{key}: {to_typ_value(val)}" for key, val in value.items())
        return f"({items})"
    raise TypeError(f"Unsupported value type for Typst conversion: {type(value)!r}")

load_and_translate_chart_diseases ¶

load_and_translate_chart_diseases(language, config_path=None)

Load and translate the chart disease list from configuration.

Loads chart_diseases_header from config/parameters.yaml and translates each disease name to the target language using the diseases_chart translation domain. This ensures chart column headers match the configured set of diseases and are properly localized.

Parameters:

Name	Type	Description	Default
`language`	`str`	Language code (e.g., "en", "fr").	required
`config_path`	`Path`	Path to `parameters.yaml`. Defaults to the repository configuration.	`None`

Returns:

Type	Description
`List[str]`	List of translated disease names in order.

Source code in pipeline/generate_notices.py

def load_and_translate_chart_diseases(
    language: str, config_path: Path | None = None
) -> List[str]:
    """Load and translate the chart disease list from configuration.

    Loads chart_diseases_header from config/parameters.yaml and translates each
    disease name to the target language using the diseases_chart translation domain.
    This ensures chart column headers match the configured set of diseases and are
    properly localized.

    Parameters
    ----------
    language : str
        Language code (e.g., "en", "fr").
    config_path : Path, optional
        Path to ``parameters.yaml``. Defaults to the repository configuration.

    Returns
    -------
    List[str]
        List of translated disease names in order.
    """
    config = load_config(config_path)
    chart_diseases_header = config.get("chart_diseases_header", [])

    translated_diseases: List[str] = []
    for disease in chart_diseases_header:
        label = display_label("diseases_chart", disease, language, strict=False)
        translated_diseases.append(label)

    return translated_diseases

build_template_context ¶

build_template_context(client, qr_output_dir=None, config_path=None)

Build template context from client data.

Translates disease names in vaccines_due_list and received records to localized display strings using the configured translation files. Also loads and translates the chart disease header list from configuration. Formats the notice date_data_cutoff with locale-aware formatting using Babel.

Parameters:

Name	Type	Description	Default
`client`	`ClientRecord`	Client record with all required fields.	required
`qr_output_dir`	`Path`	Directory containing QR code PNG files.	`None`
`config_path`	`Path`	Path to `parameters.yaml`. Defaults to the repository configuration.	`None`

Returns:

Type	Description
`Dict[str, str]`	Template context with translated disease names and formatted date.

Source code in pipeline/generate_notices.py

def build_template_context(
    client: ClientRecord,
    qr_output_dir: Path | None = None,
    config_path: Path | None = None,
) -> Dict[str, str]:
    """Build template context from client data.

    Translates disease names in vaccines_due_list and received records to
    localized display strings using the configured translation files.
    Also loads and translates the chart disease header list from configuration.
    Formats the notice date_data_cutoff with locale-aware formatting using Babel.

    Parameters
    ----------
    client : ClientRecord
        Client record with all required fields.
    qr_output_dir : Path, optional
        Directory containing QR code PNG files.
    config_path : Path, optional
        Path to ``parameters.yaml``. Defaults to the repository configuration.

    Returns
    -------
    Dict[str, str]
        Template context with translated disease names and formatted date.
    """
    config = load_config(config_path)
    preprocess_cfg: Dict[str, object] = config.get("preprocess", {})
    show_validity_markers = bool(
        preprocess_cfg.get("show_validity_markers", False)
    )

    # Load and format date_data_cutoff for the client's language
    date_data_cutoff_iso = config.get("date_data_cutoff")
    if date_data_cutoff_iso:
        date_data_cutoff_formatted = format_iso_date_for_language(
            date_data_cutoff_iso, client.language
        )
    else:
        date_data_cutoff_formatted = ""

    client_data = {
        "name": " ".join(
            filter(None, [client.person["first_name"], client.person["last_name"]])
        ).strip(),
        "address": client.contact["street"],
        "city": client.contact["city"],
        "postal_code": client.contact["postal_code"],
        "date_of_birth": client.person["date_of_birth_display"],
        "school": client.school["name"],
        "date_data_cutoff": date_data_cutoff_formatted,
        "over_16": client.person["over_16"]
    }

    # Check if QR code PNG exists from prior generation step
    if qr_output_dir:
        qr_filename = f"qr_code_{client.sequence}_{client.client_id}.png"
        qr_path = qr_output_dir / qr_filename
        if qr_path.exists():
            client_data["qr_img"] = to_root_relative(qr_path)

            # Also include QR URL (payload) if available
            if client.qr and client.qr.get("payload"):
                client_data["qr_url"] = client.qr["payload"]

    # If qr payload is present but no qr_output_dir, still include it
    # (may occur if QR generation is disabled but qr payload exists in artifact)
    if client.qr and client.qr.get("payload") and "qr_url" not in client_data:
        client_data["qr_url"] = client.qr["payload"]

    # Load and translate chart disease header
    chart_diseases_translated = load_and_translate_chart_diseases(
        client.language, config_path
    )

    # Translate vaccines_due_list to display labels
    vaccines_due_array_translated: List[str] = []
    if client.vaccines_due_list:
        for disease in client.vaccines_due_list:
            label = _localize_vaccine_due_label(disease, client.language)
            vaccines_due_array_translated.append(label)

    # Translate vaccines_due string
    vaccines_due_str_translated = (
        ", ".join(vaccines_due_array_translated)
        if vaccines_due_array_translated
        else ""
    )

    # Translate received records' column keys
    received_translated: List[Dict[str, object]] = []
    if client.received:
        for record in client.received:
            translated_record = dict(record)
            if "columns" in translated_record and isinstance(
                translated_record["columns"], dict
            ):
                translated_record["columns"] = {
                    display_label(
                        "diseases_chart", disease, client.language, strict=False
                    ): status
                    for disease, status in translated_record["columns"].items()
                }
            received_translated.append(translated_record)

    return {
        "client_row": to_typ_value([client.client_id]),
        "client_data": to_typ_value(client_data),
        "vaccines_due_str": to_typ_value(vaccines_due_str_translated),
        "vaccines_due_array": to_typ_value(vaccines_due_array_translated),
        "received": to_typ_value(received_translated),
        "num_rows": str(len(received_translated)),
        "chart_diseases_translated": to_typ_value(chart_diseases_translated),
        "show_validity_markers": to_typ_value(show_validity_markers),
    }

to_root_relative ¶

to_root_relative(path)

Convert absolute path to project-root-relative Typst path reference.

Module-internal helper for template rendering. Converts absolute file paths to paths relative to the project root, formatted for Typst's import resolution. If path is outside project root (e.g., custom assets), returns absolute path.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Absolute path to convert.	required

Returns:

Type	Description
`str`	Path string like "/artifacts/qr_codes/code.png" (relative to project root) or absolute path if outside project root.

Raises:

Type	Description
`ValueError`	If path cannot be resolved (defensive guard, should not occur in practice).

Source code in pipeline/generate_notices.py

def to_root_relative(path: Path) -> str:
    """Convert absolute path to project-root-relative Typst path reference.

    Module-internal helper for template rendering. Converts absolute file paths
    to paths relative to the project root, formatted for Typst's import resolution.
    If path is outside project root (e.g., custom assets), returns absolute path.

    Parameters
    ----------
    path : Path
        Absolute path to convert.

    Returns
    -------
    str
        Path string like "/artifacts/qr_codes/code.png" (relative to project root)
        or absolute path if outside project root.

    Raises
    ------
    ValueError
        If path cannot be resolved (defensive guard, should not occur in practice).
    """
    absolute = path.resolve()
    try:
        relative = absolute.relative_to(ROOT_DIR)
        return "/" + relative.as_posix()
    except ValueError:
        # Path is outside project root (e.g., custom template assets)
        # Return as absolute path string for Typst
        return str(absolute)

render_notice ¶

render_notice(client, *, output_dir, logo, signature, renderers, qr_output_dir=None, config_path=None)

Render a Typst notice for a single client using provided renderers.

Parameters:

Name	Type	Description	Default
`client`	`ClientRecord`	Client record with all required fields	required
`output_dir`	`Path`	Output directory (used for path resolution)	required
`logo`	`Path`	Path to logo image file	required
`signature`	`Path`	Path to signature image file	required
`renderers`	`dict`	Language code to render_notice function mapping from build_language_renderers()	required
`qr_output_dir`	`Path`	Directory containing QR code PNG files	`None`
`config_path`	`Path`	Path to `parameters.yaml`. Defaults to the repository configuration.	`None`

Returns:

Type	Description
`str`	Rendered Typst template content

Source code in pipeline/generate_notices.py

def render_notice(
    client: ClientRecord,
    *,
    output_dir: Path,
    logo: Path,
    signature: Path,
    renderers: dict,
    qr_output_dir: Path | None = None,
    config_path: Path | None = None,
) -> str:
    """Render a Typst notice for a single client using provided renderers.

    Parameters
    ----------
    client : ClientRecord
        Client record with all required fields
    output_dir : Path
        Output directory (used for path resolution)
    logo : Path
        Path to logo image file
    signature : Path
        Path to signature image file
    renderers : dict
        Language code to render_notice function mapping from build_language_renderers()
    qr_output_dir : Path, optional
        Directory containing QR code PNG files
    config_path : Path, optional
        Path to ``parameters.yaml``. Defaults to the repository configuration.

    Returns
    -------
    str
        Rendered Typst template content
    """
    language = Language.from_string(client.language)
    renderer = get_language_renderer(language, renderers)
    context = build_template_context(client, qr_output_dir, config_path)
    return renderer(
        context,
        logo_path=to_root_relative(logo),
        signature_path=to_root_relative(signature),
    )

generate_typst_files ¶

generate_typst_files(payload, output_dir, logo_path, signature_path, template_dir, config_path=None)

Generate Typst template files for all clients in payload.

Parameters:

Name	Type	Description	Default
`payload`	`ArtifactPayload`	Preprocessed client data with metadata	required
`output_dir`	`Path`	Directory to write Typst files	required
`logo_path`	`Path`	Path to logo image	required
`signature_path`	`Path`	Path to signature image	required
`template_dir`	`Path`	Directory containing language template modules	required
`config_path`	`Path`	Path to `parameters.yaml`. Defaults to the repository configuration.	`None`

Returns:

Type	Description
`List[Path]`	List of generated .typ file paths

Source code in pipeline/generate_notices.py

def generate_typst_files(
    payload: ArtifactPayload,
    output_dir: Path,
    logo_path: Path,
    signature_path: Path,
    template_dir: Path,
    config_path: Path | None = None,
) -> List[Path]:
    """Generate Typst template files for all clients in payload.

    Parameters
    ----------
    payload : ArtifactPayload
        Preprocessed client data with metadata
    output_dir : Path
        Directory to write Typst files
    logo_path : Path
        Path to logo image
    signature_path : Path
        Path to signature image
    template_dir : Path
        Directory containing language template modules
    config_path : Path, optional
        Path to ``parameters.yaml``. Defaults to the repository configuration.

    Returns
    -------
    List[Path]
        List of generated .typ file paths
    """
    # Build renderers from specified template directory
    renderers = build_language_renderers(template_dir)

    output_dir.mkdir(parents=True, exist_ok=True)
    qr_output_dir = output_dir / "qr_codes"
    typst_output_dir = output_dir / "typst"
    typst_output_dir.mkdir(parents=True, exist_ok=True)
    files: List[Path] = []
    language = payload.language
    for client in payload.clients:
        if client.language != language:
            raise ValueError(
                f"Client {client.client_id} language {client.language!r} does not match artifact language {language!r}."
            )
        typst_content = render_notice(
            client,
            output_dir=output_dir,
            logo=logo_path,
            signature=signature_path,
            renderers=renderers,
            qr_output_dir=qr_output_dir,
            config_path=config_path,
        )
        filename = f"{language}_notice_{client.sequence}_{client.client_id}.typ"
        file_path = typst_output_dir / filename
        file_path.write_text(typst_content, encoding="utf-8")
        files.append(file_path)
        LOG.info("Wrote %s", file_path)
    return files

main ¶

main(artifact_path, output_dir, logo_path, signature_path, template_dir, config_path=None)

Main entry point for Typst notice generation.

Parameters:

Name	Type	Description	Default
`artifact_path`	`Path`	Path to the preprocessed JSON artifact.	required
`output_dir`	`Path`	Directory to write Typst files.	required
`logo_path`	`Path`	Path to the logo image.	required
`signature_path`	`Path`	Path to the signature image.	required
`template_dir`	`Path`	Directory containing language template modules.	required
`config_path`	`Path`	Path to `parameters.yaml`. Defaults to the repository configuration.	`None`

Returns:

Type	Description
`List[Path]`	List of generated Typst file paths.

Source code in pipeline/generate_notices.py

def main(
    artifact_path: Path,
    output_dir: Path,
    logo_path: Path,
    signature_path: Path,
    template_dir: Path,
    config_path: Path | None = None,
) -> List[Path]:
    """Main entry point for Typst notice generation.

    Parameters
    ----------
    artifact_path : Path
        Path to the preprocessed JSON artifact.
    output_dir : Path
        Directory to write Typst files.
    logo_path : Path
        Path to the logo image.
    signature_path : Path
        Path to the signature image.
    template_dir : Path
        Directory containing language template modules.
    config_path : Path, optional
        Path to ``parameters.yaml``. Defaults to the repository configuration.

    Returns
    -------
    List[Path]
        List of generated Typst file paths.
    """
    payload = read_artifact(artifact_path)
    generated = generate_typst_files(
        payload,
        output_dir,
        logo_path,
        signature_path,
        template_dir,
        config_path,
    )
    print(
        f"Generated {len(generated)} Typst files in {output_dir} for language {payload.language}"
    )
    return generated

Step 5 — PDF Compilation¶

pipeline.compile_notices ¶

Compile per-client Typst notices into PDFs sequentially.

This lightweight helper keeps the compilation step in Python so future enhancements (parallel workers, structured logging) can be layered on in a follow-up. For now it mirrors the behaviour of the original shell script.

Input Contract:

Reads Typst template files from output/artifacts/typst/
Assumes .typ files are valid Typst templates (generated by generate_notices step)
Assumes typst compiler binary is available (configured in parameters.yaml)

Output Contract:

Writes compiled PDF files to output/pdf_individual/
All .typ files must compile successfully (critical step; fail-fast)
Filenames match input .typ files with .pdf extension

Error Handling:

Typst compilation errors raise immediately (subprocess check=True)
Missing .typ files raise immediately (fail-fast)
No per-file recovery; all-or-nothing output (critical feature)

Validation Contract:

What this module validates:

All .typ files in artifact/typst/ can be discovered
Typst compiler exists at configured path (or default 'typst')
Typst compiler exits with success (exit code 0)
Font paths are accessible (if configured)

What this module assumes (validated upstream):

.typ files are valid Typst templates (validated by generate_notices step)
Output directory can be created (general I/O)
typst.bin and typst.font_path config keys are valid (from load_config)

Note: This is a critical step. Compilation failure halts pipeline (fail-fast).

discover_typst_files ¶

discover_typst_files(artifact_dir)

Discover all Typst template files in the artifact directory.

Parameters:

Name	Type	Description	Default
`artifact_dir`	`Path`	Directory containing pipeline artifacts (should have a 'typst' subdirectory).	required

Returns:

Type	Description
`list[Path]`	Sorted list of Typst (.typ) file paths, or empty list if directory doesn't exist.

Source code in pipeline/compile_notices.py

def discover_typst_files(artifact_dir: Path) -> list[Path]:
    """Discover all Typst template files in the artifact directory.

    Parameters
    ----------
    artifact_dir : Path
        Directory containing pipeline artifacts (should have a 'typst' subdirectory).

    Returns
    -------
    list[Path]
        Sorted list of Typst (.typ) file paths, or empty list if directory doesn't exist.
    """
    typst_dir = artifact_dir / "typst"
    if not typst_dir.exists():
        return []
    return sorted(typst_dir.glob("*.typ"))

compile_file ¶

compile_file(typ_path, pdf_dir, *, typst_bin, font_path, root_dir, verbose)

Compile a single Typst template file to PDF.

Parameters:

Name	Type	Description	Default
`typ_path`	`Path`	Path to the .typ Typst template file to compile.	required
`pdf_dir`	`Path`	Directory where the compiled PDF should be written.	required
`typst_bin`	`str`	Path or name of the typst binary to use for compilation.	required
`font_path`	`Path \| None`	Optional path to directory containing custom fonts.	required
`root_dir`	`Path`	Root directory for relative path resolution in Typst compilation. This should be the template directory containing conf.typ and assets/.	required
`verbose`	`bool`	If True, print compilation status message.	required

Source code in pipeline/compile_notices.py

def compile_file(
    typ_path: Path,
    pdf_dir: Path,
    *,
    typst_bin: str,
    font_path: Path | None,
    root_dir: Path,
    verbose: bool,
) -> None:
    """Compile a single Typst template file to PDF.

    Parameters
    ----------
    typ_path : Path
        Path to the .typ Typst template file to compile.
    pdf_dir : Path
        Directory where the compiled PDF should be written.
    typst_bin : str
        Path or name of the typst binary to use for compilation.
    font_path : Path | None
        Optional path to directory containing custom fonts.
    root_dir : Path
        Root directory for relative path resolution in Typst compilation.
        This should be the template directory containing conf.typ and assets/.
    verbose : bool
        If True, print compilation status message.
    """
    pdf_path = pdf_dir / f"{typ_path.stem}.pdf"
    command = [typst_bin, "compile"]
    if font_path:
        command.extend(["--font-path", str(font_path)])
    command.extend(["--root", str(root_dir), str(typ_path), str(pdf_path)])
    subprocess.run(command, check=True)
    if verbose:
        print(f"Compiled {typ_path.name} -> {pdf_path.name}")

compile_typst_files ¶

compile_typst_files(artifact_dir, pdf_dir, *, typst_bin, font_path, root_dir, verbose)

Compile all discovered Typst template files sequentially to PDFs.

Parameters:

Name	Type	Description	Default
`artifact_dir`	`Path`	Directory containing Typst artifacts.	required
`pdf_dir`	`Path`	Output directory for compiled PDFs.	required
`typst_bin`	`str`	Path or name of the typst binary.	required
`font_path`	`Path \| None`	Optional custom fonts directory.	required
`root_dir`	`Path`	Root directory for relative path resolution. Should be the template directory containing conf.typ and assets/.	required
`verbose`	`bool`	If True, print per-file compilation status.	required

Returns:

Type	Description
`int`	Number of files successfully compiled.

Source code in pipeline/compile_notices.py

def compile_typst_files(
    artifact_dir: Path,
    pdf_dir: Path,
    *,
    typst_bin: str,
    font_path: Path | None,
    root_dir: Path,
    verbose: bool,
) -> int:
    """Compile all discovered Typst template files sequentially to PDFs.

    Parameters
    ----------
    artifact_dir : Path
        Directory containing Typst artifacts.
    pdf_dir : Path
        Output directory for compiled PDFs.
    typst_bin : str
        Path or name of the typst binary.
    font_path : Path | None
        Optional custom fonts directory.
    root_dir : Path
        Root directory for relative path resolution.
        Should be the template directory containing conf.typ and assets/.
    verbose : bool
        If True, print per-file compilation status.

    Returns
    -------
    int
        Number of files successfully compiled.
    """
    pdf_dir.mkdir(parents=True, exist_ok=True)
    typ_files = discover_typst_files(artifact_dir)
    if not typ_files:
        print(f"No Typst artifacts found in {artifact_dir}.")
        return 0

    for typ_path in typ_files:
        compile_file(
            typ_path,
            pdf_dir,
            typst_bin=typst_bin,
            font_path=font_path,
            root_dir=root_dir,
            verbose=verbose,
        )
    return len(typ_files)

compile_with_config ¶

compile_with_config(artifact_dir, output_dir, config_path=None, template_dir=None)

Compile Typst files using configuration from parameters.yaml.

Reads typst configuration (binary path, font path) from parameters.yaml and compiles all Typst files in the artifact directory.

Parameters:

Name	Type	Description	Default
`artifact_dir`	`Path`	Directory containing Typst template files	required
`output_dir`	`Path`	Directory where compiled PDFs will be written	required
`config_path`	`Path`	Path to parameters.yaml. If not provided, uses default location.	`None`
`template_dir`	`Path`	Template directory for dynamic template loading. Typst compilation always uses PROJECT_ROOT as --root to find both templates and output artifacts.	`None`

Returns:

Type	Description
`int`	Number of files compiled.

Source code in pipeline/compile_notices.py

def compile_with_config(
    artifact_dir: Path,
    output_dir: Path,
    config_path: Path | None = None,
    template_dir: Path | None = None,
) -> int:
    """Compile Typst files using configuration from parameters.yaml.

    Reads typst configuration (binary path, font path) from parameters.yaml
    and compiles all Typst files in the artifact directory.

    Parameters
    ----------
    artifact_dir : Path
        Directory containing Typst template files
    output_dir : Path
        Directory where compiled PDFs will be written
    config_path : Path, optional
        Path to parameters.yaml. If not provided, uses default location.
    template_dir : Path, optional
        Template directory for dynamic template loading. Typst compilation always
        uses PROJECT_ROOT as --root to find both templates and output artifacts.

    Returns
    -------
    int
        Number of files compiled.
    """
    config = load_config(config_path)

    typst_config = config.get("typst", {})
    font_path_str = typst_config.get("font_path", "/usr/share/fonts/truetype/freefont/")
    typst_bin = typst_config.get("bin", "typst")

    # Allow TYPST_BIN environment variable to override config
    typst_bin = os.environ.get("TYPST_BIN", typst_bin)

    font_path = Path(font_path_str) if font_path_str else None

    # Typst compilation always uses PROJECT_ROOT to find both templates and output artifacts.
    # template_dir parameter is for dynamic template loading in generate_notices, not for Typst --root.
    root_dir = ROOT_DIR

    return compile_typst_files(
        artifact_dir,
        output_dir,
        typst_bin=typst_bin,
        font_path=font_path,
        root_dir=root_dir,
        verbose=False,
    )

main ¶

main(artifact_dir, output_dir, config_path=None, template_dir=None)

Main entry point for Typst compilation.

Parameters:

Name	Type	Description	Default
`artifact_dir`	`Path`	Directory containing Typst artifacts.	required
`output_dir`	`Path`	Directory for output PDFs.	required
`config_path`	`Path`	Path to parameters.yaml configuration file.	`None`
`template_dir`	`Path`	Template directory containing conf.typ and assets/. Used as Typst --root. If not provided, defaults to project root.	`None`

Returns:

Type	Description
`int`	Number of files compiled.

Source code in pipeline/compile_notices.py

def main(
    artifact_dir: Path,
    output_dir: Path,
    config_path: Path | None = None,
    template_dir: Path | None = None,
) -> int:
    """Main entry point for Typst compilation.

    Parameters
    ----------
    artifact_dir : Path
        Directory containing Typst artifacts.
    output_dir : Path
        Directory for output PDFs.
    config_path : Path, optional
        Path to parameters.yaml configuration file.
    template_dir : Path, optional
        Template directory containing conf.typ and assets/. Used as Typst --root.
        If not provided, defaults to project root.

    Returns
    -------
    int
        Number of files compiled.
    """
    compiled = compile_with_config(artifact_dir, output_dir, config_path, template_dir)
    if compiled:
        print(f"Compiled {compiled} Typst file(s) to PDFs in {output_dir}.")
    return compiled

Step 6 — PDF Validation¶

pipeline.validate_pdfs ¶

Validate compiled PDFs for layout, structure, and quality issues.

Performs comprehensive validation of compiled PDF files including page counts, layout checks (signature placement), and structural integrity. Outputs validation results to JSON metadata for downstream processing and optional console warnings.

Input Contract:

Reads PDF files from output/pdf_individual/ directory
Assumes PDFs are valid (created by compilation step)
Assumes each PDF corresponds to one client notice

Output Contract:

Writes validation results to JSON: output/metadata/{language}validation.json
Records per-PDF validations: page counts, layout warnings, structural issues
Aggregate statistics: total PDFs, warnings by type, pass/fail counts
Optional console output (controlled by config: pdf_validation.print_warnings)

Error Handling:

Invalid/corrupt PDFs raise immediately (fail-fast; quality validation step)
Missing PDF files raise immediately (infrastructure error)
Layout warnings are non-fatal (logged but don't halt pipeline)
All PDFs must be readable; validation results may contain warnings (quality step)

Validation Contract:

What this module validates:

PDF files are readable and structurally valid (uses PdfReader)
Page count statistics and distribution
Layout markers (signature block placement using MARK_END_SIGNATURE_BLOCK)
Expected vs actual page counts (configurable tolerance)

What this module assumes (validated upstream):

PDF files exist and are complete (created by compile step)
PDF filenames match expected pattern (from notice generation)
Output metadata directory can be created (general I/O)

Note: This is a validation/QA step. Structural PDF errors halt pipeline (fail-fast), but layout warnings are non-fatal and logged for review.

ValidationResult `dataclass` ¶

Result of validating a single PDF file.

Attributes:

Name	Type	Description
`filename`	`str`	Name of the PDF file
`warnings`	`List[str]`	List of validation warnings (layout issues, unexpected page counts, etc.)
`passed`	`bool`	True if no warnings, False otherwise
`measurements`	`dict[str, int \| float \| str]`	Actual measurements extracted from PDF with proper types: page_count (int): Number of pages signature_page (int): Page where signature block ends contact_height_inches (float): Contact table height in inches client_id_found_page (int): Page where client ID was found client_id_found_value (str): The actual client ID found

Source code in pipeline/validate_pdfs.py

@dataclass
class ValidationResult:
    """Result of validating a single PDF file.

    Attributes
    ----------
    filename : str
        Name of the PDF file
    warnings : List[str]
        List of validation warnings (layout issues, unexpected page counts, etc.)
    passed : bool
        True if no warnings, False otherwise
    measurements : dict[str, int | float | str]
        Actual measurements extracted from PDF with proper types:

        - page_count (int): Number of pages
        - signature_page (int): Page where signature block ends
        - contact_height_inches (float): Contact table height in inches
        - client_id_found_page (int): Page where client ID was found
        - client_id_found_value (str): The actual client ID found
    """

    filename: str
    warnings: List[str]
    passed: bool
    measurements: dict[str, int | float | str]

RuleResult `dataclass` ¶

Result of a single validation rule across all PDFs.

Attributes:

Name	Type	Description
`rule_name`	`str`	Name of the validation rule
`severity`	`str`	Rule severity: "disabled", "warn", or "error"
`passed_count`	`int`	Number of PDFs that passed this rule
`failed_count`	`int`	Number of PDFs that failed this rule

Source code in pipeline/validate_pdfs.py

@dataclass
class RuleResult:
    """Result of a single validation rule across all PDFs.

    Attributes
    ----------
    rule_name : str
        Name of the validation rule
    severity : str
        Rule severity: "disabled", "warn", or "error"
    passed_count : int
        Number of PDFs that passed this rule
    failed_count : int
        Number of PDFs that failed this rule
    """

    rule_name: str
    severity: str
    passed_count: int
    failed_count: int

ValidationSummary `dataclass` ¶

Aggregate validation results for all PDFs.

Attributes:

Name	Type	Description
`language`	`str \| None`	Language code if filtered (e.g., 'en' or 'fr')
`total_pdfs`	`int`	Total number of PDFs validated
`passed_count`	`int`	Number of PDFs with no warnings
`warning_count`	`int`	Number of PDFs with warnings
`page_count_distribution`	`dict[int, int]`	Distribution of page counts (pages -> count)
`warning_types`	`dict[str, int]`	Count of warnings by type/category
`rule_results`	`List[RuleResult]`	Per-rule validation statistics
`results`	`List[ValidationResult]`	Per-file validation results

Source code in pipeline/validate_pdfs.py

@dataclass
class ValidationSummary:
    """Aggregate validation results for all PDFs.

    Attributes
    ----------
    language : str | None
        Language code if filtered (e.g., 'en' or 'fr')
    total_pdfs : int
        Total number of PDFs validated
    passed_count : int
        Number of PDFs with no warnings
    warning_count : int
        Number of PDFs with warnings
    page_count_distribution : dict[int, int]
        Distribution of page counts (pages -> count)
    warning_types : dict[str, int]
        Count of warnings by type/category
    rule_results : List[RuleResult]
        Per-rule validation statistics
    results : List[ValidationResult]
        Per-file validation results
    """

    language: str | None
    total_pdfs: int
    passed_count: int
    warning_count: int
    page_count_distribution: dict[int, int]
    warning_types: dict[str, int]
    rule_results: List[RuleResult]
    results: List[ValidationResult]

discover_pdfs ¶

discover_pdfs(target)

Discover all PDF files at the given target path.

Parameters:

Name	Type	Description	Default
`target`	`Path`	Either a directory containing PDFs or a single PDF file.	required

Returns:

Type	Description
`List[Path]`	Sorted list of PDF file paths.

Raises:

Type	Description
`FileNotFoundError`	If target is neither a PDF file nor a directory containing PDFs.

Source code in pipeline/validate_pdfs.py

def discover_pdfs(target: Path) -> List[Path]:
    """Discover all PDF files at the given target path.

    Parameters
    ----------
    target : Path
        Either a directory containing PDFs or a single PDF file.

    Returns
    -------
    List[Path]
        Sorted list of PDF file paths.

    Raises
    ------
    FileNotFoundError
        If target is neither a PDF file nor a directory containing PDFs.
    """
    if target.is_dir():
        return sorted(target.glob("*.pdf"))
    if target.is_file() and target.suffix.lower() == ".pdf":
        return [target]
    raise FileNotFoundError(f"No PDF(s) found at {target}")

filter_by_language ¶

filter_by_language(files, language)

Filter PDF files by language prefix in filename.

Parameters:

Name	Type	Description	Default
`files`	`List[Path]`	PDF file paths to filter.	required
`language`	`str \| None`	Language code to filter by (e.g., 'en' or 'fr'). If None, returns all files.	required

Returns:

Type	Description
`List[Path]`	Filtered list of PDF paths, or all files if language is None.

Source code in pipeline/validate_pdfs.py

def filter_by_language(files: List[Path], language: str | None) -> List[Path]:
    """Filter PDF files by language prefix in filename.

    Parameters
    ----------
    files : List[Path]
        PDF file paths to filter.
    language : str | None
        Language code to filter by (e.g., 'en' or 'fr'). If None, returns all files.

    Returns
    -------
    List[Path]
        Filtered list of PDF paths, or all files if language is None.
    """
    if not language:
        return list(files)
    prefix = f"{language}_"
    return [path for path in files if path.name.startswith(prefix)]

find_client_id_in_text ¶

find_client_id_in_text(page_text)

Find a 10-digit client ID in extracted PDF page text.

Searches for any 10-digit number; assumes the first match is the client ID. May be preceded by "Client ID: " or "Identifiant du client: " (optional).

Parameters:

Name	Type	Description	Default
`page_text`	`str`	Extracted text from a PDF page.	required

Returns:

Type	Description
`str \| None`	10-digit client ID if found, None otherwise.

Source code in pipeline/validate_pdfs.py

def find_client_id_in_text(page_text: str) -> str | None:
    """Find a 10-digit client ID in extracted PDF page text.

    Searches for any 10-digit number; assumes the first match is the client ID.
    May be preceded by "Client ID: " or "Identifiant du client: " (optional).

    Parameters
    ----------
    page_text : str
        Extracted text from a PDF page.

    Returns
    -------
    str | None
        10-digit client ID if found, None otherwise.
    """
    import re

    # Search for any 10-digit number (word boundary on both sides to avoid false matches)
    match = re.search(r"\b(\d{10})\b", page_text)
    if match:
        return match.group(1)
    return None

extract_measurements_from_markers ¶

extract_measurements_from_markers(page_text)

Extract dimension measurements from invisible text markers.

Typst templates embed invisible markers with measurements like: MEASURE_CONTACT_HEIGHT:123.45

Parameters:

Name	Type	Description	Default
`page_text`	`str`	Extracted text from a PDF page.	required

Returns:

Type	Description
`dict[str, float]`	Dictionary mapping dimension names to values in points. Example: {"measure_contact_height": 123.45}

Source code in pipeline/validate_pdfs.py

def extract_measurements_from_markers(page_text: str) -> dict[str, float]:
    """Extract dimension measurements from invisible text markers.

    Typst templates embed invisible markers with measurements like:
    MEASURE_CONTACT_HEIGHT:123.45

    Parameters
    ----------
    page_text : str
        Extracted text from a PDF page.

    Returns
    -------
    dict[str, float]
        Dictionary mapping dimension names to values in points.
        Example: {"measure_contact_height": 123.45}
    """
    import re

    measurements = {}

    # Pattern to match our invisible marker format: MEASURE_NAME:123.45
    pattern = r"MEASURE_(\w+):([\d.]+)"

    for match in re.finditer(pattern, page_text):
        key = "measure_" + match.group(1).lower()  # normalize to lowercase
        value = float(match.group(2))
        measurements[key] = value

    return measurements

validate_pdf_layout ¶

validate_pdf_layout(pdf_path, reader, enabled_rules, client_id_map=None)

Check PDF for layout issues using invisible markers and metadata.

Parameters:

Name	Type	Description	Default
`pdf_path`	`Path`	Path to the PDF file being validated.	required
`reader`	`PdfReader`	Opened PDF reader instance.	required
`enabled_rules`	`dict[str, str]`	Validation rules configuration (rule_name -> "disabled"/"warn"/"error").	required
`client_id_map`	`dict[str, str]`	Mapping of PDF filename (without path) to expected client ID. If provided, client_id_presence validation uses this as source of truth.	`None`

Returns:

Type	Description
`tuple[List[str], dict[str, float]]`	Tuple of (warning messages, actual measurements). Measurements include signature_page, contact_height_inches, etc.

Source code in pipeline/validate_pdfs.py

def validate_pdf_layout(
    pdf_path: Path,
    reader: PdfReader,
    enabled_rules: dict[str, str],
    client_id_map: dict[str, str] | None = None,
) -> tuple[List[str], dict[str, float]]:
    """Check PDF for layout issues using invisible markers and metadata.

    Parameters
    ----------
    pdf_path : Path
        Path to the PDF file being validated.
    reader : PdfReader
        Opened PDF reader instance.
    enabled_rules : dict[str, str]
        Validation rules configuration (rule_name -> "disabled"/"warn"/"error").
    client_id_map : dict[str, str], optional
        Mapping of PDF filename (without path) to expected client ID.
        If provided, client_id_presence validation uses this as source of truth.

    Returns
    -------
    tuple[List[str], dict[str, float]]
        Tuple of (warning messages, actual measurements).
        Measurements include signature_page, contact_height_inches, etc.
    """
    warnings = []
    measurements = {}

    # Check signature block marker placement
    rule_setting = enabled_rules.get("signature_overflow", "warn")
    if rule_setting != "disabled":
        for page_num, page in enumerate(reader.pages, start=1):
            try:
                page_text = page.extract_text()
                if "MARK_END_SIGNATURE_BLOCK" in page_text:
                    measurements["signature_page"] = page_num
                    if page_num != 1:
                        warnings.append(
                            f"signature_overflow: Signature block ends on page {page_num} "
                            f"(expected page 1)"
                        )
                    break
            except Exception:
                # If text extraction fails, skip this check
                pass

    # Check contact table dimensions (envelope window validation)
    envelope_rule = enabled_rules.get("envelope_window_1_125", "disabled")
    if envelope_rule != "disabled":
        # Envelope window constraint: 1.125 inches max height
        max_height_inches = 1.125

        # Look for contact table measurements in page 1
        try:
            page_text = reader.pages[0].extract_text()
            extracted_measurements = extract_measurements_from_markers(page_text)

            contact_height_pt = extracted_measurements.get("measure_contact_height")
            if contact_height_pt:
                # Convert from points to inches (72 points = 1 inch)
                height_inches = contact_height_pt / 72.0
                measurements["contact_height_inches"] = height_inches

                if height_inches > max_height_inches:
                    warnings.append(
                        f"envelope_window_1_125: Contact table height {height_inches:.2f}in "
                        f"exceeds envelope window (max {max_height_inches}in)"
                    )
        except Exception:
            # If measurement extraction fails, skip this check
            pass

    # Check client ID presence (markerless: search for 10-digit number in text)
    client_id_rule = enabled_rules.get("client_id_presence", "disabled")
    if client_id_rule != "disabled" and client_id_map:
        try:
            # Get expected client ID from the mapping (source of truth: preprocessed_clients.json)
            expected_client_id = client_id_map.get(pdf_path.name)
            if expected_client_id:
                # Search all pages for the client ID
                found_client_id = None
                for page_num, page in enumerate(reader.pages, start=1):
                    page_text = page.extract_text()
                    found_id = find_client_id_in_text(page_text)
                    if found_id:
                        found_client_id = found_id
                        measurements["client_id_found_page"] = page_num
                        break

                # Warn if ID not found or doesn't match
                if found_client_id is None:
                    warnings.append(
                        f"client_id_presence: Client ID {expected_client_id} not found in PDF"
                    )
                elif found_client_id != expected_client_id:
                    warnings.append(
                        f"client_id_presence: Found ID {found_client_id}, expected {expected_client_id}"
                    )
                else:
                    # Store the found ID for debugging
                    measurements["client_id_found_value"] = found_client_id
        except Exception:
            # If client ID check fails, skip silently (parsing error)
            pass

    return warnings, measurements

validate_pdf_structure ¶

validate_pdf_structure(pdf_path, enabled_rules=None, client_id_map=None)

Validate a single PDF file for structure and layout.

Parameters:

Name	Type	Description	Default
`pdf_path`	`Path`	Path to the PDF file to validate.	required
`enabled_rules`	`dict[str, str]`	Validation rules configuration (rule_name -> "disabled"/"warn"/"error").	`None`
`client_id_map`	`dict[str, str]`	Mapping of PDF filename to expected client ID (from preprocessed_clients.json).	`None`

Returns:

Type	Description
`ValidationResult`	Validation result with measurements, warnings, and pass/fail status.

Raises:

Type	Description
`Exception`	If PDF cannot be read (structural corruption).

Source code in pipeline/validate_pdfs.py

def validate_pdf_structure(
    pdf_path: Path,
    enabled_rules: dict[str, str] | None = None,
    client_id_map: dict[str, str] | None = None,
) -> ValidationResult:
    """Validate a single PDF file for structure and layout.

    Parameters
    ----------
    pdf_path : Path
        Path to the PDF file to validate.
    enabled_rules : dict[str, str], optional
        Validation rules configuration (rule_name -> "disabled"/"warn"/"error").
    client_id_map : dict[str, str], optional
        Mapping of PDF filename to expected client ID (from preprocessed_clients.json).

    Returns
    -------
    ValidationResult
        Validation result with measurements, warnings, and pass/fail status.

    Raises
    ------
    Exception
        If PDF cannot be read (structural corruption).
    """
    warnings = []
    measurements = {}
    if enabled_rules is None:
        enabled_rules = {}

    # Read PDF and count pages
    reader = PdfReader(str(pdf_path))
    page_count = len(reader.pages)
    measurements["page_count"] = page_count

    # Check for exactly 2 pages (standard notice format)
    rule_setting = enabled_rules.get("exactly_two_pages", "warn")
    if rule_setting != "disabled":
        if page_count != 2:
            warnings.append(f"exactly_two_pages: has {page_count} pages (expected 2)")

    # Validate layout using markers
    layout_warnings, layout_measurements = validate_pdf_layout(
        pdf_path, reader, enabled_rules, client_id_map=client_id_map
    )
    warnings.extend(layout_warnings)
    measurements.update(layout_measurements)

    return ValidationResult(
        filename=pdf_path.name,
        warnings=warnings,
        passed=len(warnings) == 0,
        measurements=measurements,
    )

compute_rule_results ¶

compute_rule_results(results, enabled_rules)

Compute per-rule pass/fail statistics.

Parameters:

Name	Type	Description	Default
`results`	`List[ValidationResult]`	Validation results for all PDFs.	required
`enabled_rules`	`dict[str, str]`	Validation rules configuration (rule_name -> "disabled"/"warn"/"error").	required

Returns:

Type	Description
`List[RuleResult]`	Per-rule statistics with pass/fail counts.

Source code in pipeline/validate_pdfs.py

def compute_rule_results(
    results: List[ValidationResult], enabled_rules: dict[str, str]
) -> List[RuleResult]:
    """Compute per-rule pass/fail statistics.

    Parameters
    ----------
    results : List[ValidationResult]
        Validation results for all PDFs.
    enabled_rules : dict[str, str]
        Validation rules configuration (rule_name -> "disabled"/"warn"/"error").

    Returns
    -------
    List[RuleResult]
        Per-rule statistics with pass/fail counts.
    """
    # Count failures per rule
    rule_failures: Counter = Counter()
    for result in results:
        for warning in result.warnings:
            rule_name = warning.split(":")[0] if ":" in warning else "other"
            rule_failures[rule_name] += 1

    # Build rule results for all configured rules
    rule_results = []
    for rule_name, severity in enabled_rules.items():
        failed_count = rule_failures.get(rule_name, 0)
        passed_count = len(results) - failed_count

        rule_results.append(
            RuleResult(
                rule_name=rule_name,
                severity=severity,
                passed_count=passed_count,
                failed_count=failed_count,
            )
        )

    return rule_results

validate_pdfs ¶

validate_pdfs(files, enabled_rules=None, client_id_map=None)

Validate all PDF files and generate summary.

Parameters:

Name	Type	Description	Default
`files`	`List[Path]`	PDF file paths to validate.	required
`enabled_rules`	`dict[str, str]`	Validation rules configuration (rule_name -> "disabled"/"warn"/"error").	`None`
`client_id_map`	`dict[str, str]`	Mapping of PDF filename to expected client ID (from preprocessed_clients.json).	`None`

Returns:

Type	Description
`ValidationSummary`	Aggregate validation results with statistics and per-file details.

Source code in pipeline/validate_pdfs.py

def validate_pdfs(
    files: List[Path],
    enabled_rules: dict[str, str] | None = None,
    client_id_map: dict[str, str] | None = None,
) -> ValidationSummary:
    """Validate all PDF files and generate summary.

    Parameters
    ----------
    files : List[Path]
        PDF file paths to validate.
    enabled_rules : dict[str, str], optional
        Validation rules configuration (rule_name -> "disabled"/"warn"/"error").
    client_id_map : dict[str, str], optional
        Mapping of PDF filename to expected client ID (from preprocessed_clients.json).

    Returns
    -------
    ValidationSummary
        Aggregate validation results with statistics and per-file details.
    """
    if enabled_rules is None:
        enabled_rules = {}
    if client_id_map is None:
        client_id_map = {}

    results: List[ValidationResult] = []
    page_buckets: Counter = Counter()
    warning_type_counts: Counter = Counter()

    for pdf_path in files:
        result = validate_pdf_structure(
            pdf_path, enabled_rules=enabled_rules, client_id_map=client_id_map
        )
        results.append(result)
        page_count = int(result.measurements.get("page_count", 0))
        page_buckets[page_count] += 1

        # Count warning types
        for warning in result.warnings:
            warning_type = warning.split(":")[0] if ":" in warning else "other"
            warning_type_counts[warning_type] += 1

    passed_count = sum(1 for r in results if r.passed)
    warning_count = len(results) - passed_count

    # Compute per-rule statistics
    rule_results = compute_rule_results(results, enabled_rules)

    return ValidationSummary(
        language=None,  # Set by caller
        total_pdfs=len(results),
        passed_count=passed_count,
        warning_count=warning_count,
        page_count_distribution=dict(sorted(page_buckets.items())),
        warning_types=dict(warning_type_counts),
        rule_results=rule_results,
        results=results,
    )

print_validation_summary ¶

print_validation_summary(summary, *, validation_json_path=None)

Print human-readable validation summary to console.

Parameters:

Name	Type	Description	Default
`summary`	`ValidationSummary`	Validation summary to print.	required
`validation_json_path`	`Path`	Path to validation JSON for reference in output.	`None`

Source code in pipeline/validate_pdfs.py

def print_validation_summary(
    summary: ValidationSummary,
    *,
    validation_json_path: Path | None = None,
) -> None:
    """Print human-readable validation summary to console.

    Parameters
    ----------
    summary : ValidationSummary
        Validation summary to print.
    validation_json_path : Path, optional
        Path to validation JSON for reference in output.
    """
    # Per-rule summary (all rules, including disabled)
    print("Validation rules:")
    for rule in summary.rule_results:
        status_str = f"- {rule.rule_name} [{rule.severity}]"
        count_str = f"✓ {rule.passed_count} passed"

        if rule.failed_count > 0:
            fail_label = "PDF" if rule.failed_count == 1 else "PDFs"
            count_str += f", ✗ {rule.failed_count} {fail_label} failed"

        print(f"  {status_str}: {count_str}")

    # Reference to detailed log
    if validation_json_path:
        try:
            relative_path = validation_json_path.relative_to(Path.cwd())
            print(f"\nDetailed validation results: {relative_path}")
        except ValueError:
            # If path is not relative to cwd (e.g., in temp dir), use absolute
            print(f"\nDetailed validation results: {validation_json_path}")

write_validation_json ¶

write_validation_json(summary, output_path)

Write validation summary to JSON file.

Parameters:

Name	Type	Description	Default
`summary`	`ValidationSummary`	Validation summary to serialize.	required
`output_path`	`Path`	Path to output JSON file.	required

Source code in pipeline/validate_pdfs.py

def write_validation_json(summary: ValidationSummary, output_path: Path) -> None:
    """Write validation summary to JSON file.

    Parameters
    ----------
    summary : ValidationSummary
        Validation summary to serialize.
    output_path : Path
        Path to output JSON file.
    """
    output_path.parent.mkdir(parents=True, exist_ok=True)

    # Convert to dict and serialize
    payload = asdict(summary)
    output_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")

check_for_errors ¶

check_for_errors(summary, enabled_rules)

Check if any validation rules are set to 'error' and have failures.

Parameters:

Name	Type	Description	Default
`summary`	`ValidationSummary`	Validation summary with warning counts by type.	required
`enabled_rules`	`dict[str, str]`	Validation rules configuration (rule_name -> "disabled"/"warn"/"error").	required

Returns:

Type	Description
`List[str]`	List of error messages for rules that failed with severity 'error'.

Source code in pipeline/validate_pdfs.py

def check_for_errors(
    summary: ValidationSummary, enabled_rules: dict[str, str]
) -> List[str]:
    """Check if any validation rules are set to 'error' and have failures.

    Parameters
    ----------
    summary : ValidationSummary
        Validation summary with warning counts by type.
    enabled_rules : dict[str, str]
        Validation rules configuration (rule_name -> "disabled"/"warn"/"error").

    Returns
    -------
    List[str]
        List of error messages for rules that failed with severity 'error'.
    """
    errors = []
    for rule_name, severity in enabled_rules.items():
        if severity == "error" and rule_name in summary.warning_types:
            count = summary.warning_types[rule_name]
            label = "PDF" if count == 1 else "PDFs"
            errors.append(f"{rule_name}: {count} {label} failed validation")
    return errors

main ¶

main(target, language=None, enabled_rules=None, json_output=None, client_id_map=None, config_dir=None)

Main entry point for PDF validation.

Parameters:

Name	Type	Description	Default
`target`	`Path`	PDF file or directory containing PDFs.	required
`language`	`str`	Optional language prefix to filter PDF filenames (e.g., 'en').	`None`
`enabled_rules`	`dict[str, str]`	Validation rules configuration (rule_name -> "disabled"/"warn"/"error"). If not provided and config_dir is given, loads from config_dir/parameters.yaml.	`None`
`json_output`	`Path`	Optional path to write validation summary as JSON.	`None`
`client_id_map`	`dict[str, str]`	Mapping of PDF filename to expected client ID (from preprocessed_clients.json).	`None`
`config_dir`	`Path`	Path to config directory containing parameters.yaml. Used to load enabled_rules if not explicitly provided. If not provided, uses default location (config/parameters.yaml in project root).	`None`

Returns:

Type	Description
`ValidationSummary`	Validation summary with all results and statistics.

Raises:

Type	Description
`RuntimeError`	If any validation rule with severity 'error' fails.

Source code in pipeline/validate_pdfs.py

def main(
    target: Path,
    language: str | None = None,
    enabled_rules: dict[str, str] | None = None,
    json_output: Path | None = None,
    client_id_map: dict[str, str] | None = None,
    config_dir: Path | None = None,
) -> ValidationSummary:
    """Main entry point for PDF validation.

    Parameters
    ----------
    target : Path
        PDF file or directory containing PDFs.
    language : str, optional
        Optional language prefix to filter PDF filenames (e.g., 'en').
    enabled_rules : dict[str, str], optional
        Validation rules configuration (rule_name -> "disabled"/"warn"/"error").
        If not provided and config_dir is given, loads from config_dir/parameters.yaml.
    json_output : Path, optional
        Optional path to write validation summary as JSON.
    client_id_map : dict[str, str], optional
        Mapping of PDF filename to expected client ID (from preprocessed_clients.json).
    config_dir : Path, optional
        Path to config directory containing parameters.yaml.
        Used to load enabled_rules if not explicitly provided.
        If not provided, uses default location (config/parameters.yaml in project root).

    Returns
    -------
    ValidationSummary
        Validation summary with all results and statistics.

    Raises
    ------
    RuntimeError
        If any validation rule with severity 'error' fails.
    """
    # Load enabled_rules from config if not provided
    if enabled_rules is None:
        config_path = None if config_dir is None else config_dir / "parameters.yaml"
        config = load_config(config_path)
        validation_config = config.get("pdf_validation", {})
        enabled_rules = validation_config.get("rules", {})

    if client_id_map is None:
        client_id_map = {}

    files = discover_pdfs(target)
    filtered = filter_by_language(files, language)
    summary = validate_pdfs(
        filtered, enabled_rules=enabled_rules, client_id_map=client_id_map
    )
    summary.language = language

    if json_output:
        write_validation_json(summary, json_output)

    # Always print summary
    print_validation_summary(summary, validation_json_path=json_output)

    # Check for error-level failures
    errors = check_for_errors(summary, enabled_rules)
    if errors:
        error_msg = "PDF validation failed with errors:\n  " + "\n  ".join(errors)
        raise RuntimeError(error_msg)

    return summary

Step 7 — PDF Encryption¶

pipeline.encrypt_notice ¶

Encryption module for immunization PDF notices.

This module provides functions to encrypt PDF notices using client metadata. It's designed to be integrated into the pipeline as an optional step.

Passwords are generated per-client per-PDF using templates defined in config/parameters.yaml under encryption.password.template. Templates support placeholders like {client_id}, {date_of_birth_iso}, {date_of_birth_iso_compact}, {first_name}, {last_name}, {school}, {postal_code}, etc.

Input Contract:

Reads PDF files from disk and client metadata from JSON
Assumes PDF and JSON files exist before encryption
Assumes JSON contains valid client metadata with required fields for password template

Output Contract:

Writes encrypted PDFs to disk with "_encrypted" suffix
Unencrypted originals are preserved (deleted during cleanup step if configured)
Per-PDF failures are logged and skipped (optional feature; some PDFs may not be encrypted)
Pipeline completes even if some PDFs fail to encrypt

Error Handling:

Infrastructure errors (missing PDF/JSON files) raise immediately (fail-fast)
Configuration errors (invalid password template) raise immediately (fail-fast)
Per-PDF failures (encryption error, invalid template data) are logged and skipped
This strategy allows partial success; users are notified with summary of results
Per-PDF recovery is intentional for optional step; allows users to still get output

load_encryption_config ¶

load_encryption_config()

Load and cache encryption configuration from parameters.yaml.

Module-internal helper. Configuration is loaded once and cached globally for subsequent function calls. This avoids repeated file I/O when generating passwords for multiple PDFs.

Returns:

Type	Description
`dict`	Encryption configuration dict (typically contains 'password' key with 'template' sub-key), or empty dict if config file not found.

Source code in pipeline/encrypt_notice.py

def load_encryption_config():
    """Load and cache encryption configuration from parameters.yaml.

    Module-internal helper. Configuration is loaded once and cached globally
    for subsequent function calls. This avoids repeated file I/O when generating
    passwords for multiple PDFs.

    Returns
    -------
    dict
        Encryption configuration dict (typically contains 'password' key with
        'template' sub-key), or empty dict if config file not found.
    """
    global _encryption_config
    if _encryption_config is None:
        try:
            parameters_path = CONFIG_DIR / "parameters.yaml"
            if parameters_path.exists():
                with open(parameters_path) as f:
                    params = yaml.safe_load(f) or {}
                    _encryption_config = params.get("encryption", {})
            else:
                _encryption_config = {}
        except Exception:
            _encryption_config = {}
    return _encryption_config

get_encryption_config ¶

get_encryption_config()

Get the encryption configuration from parameters.yaml.

Returns:

Type	Description
`dict`	Cached encryption configuration.

Source code in pipeline/encrypt_notice.py

def get_encryption_config():
    """Get the encryption configuration from parameters.yaml.

    Returns
    -------
    dict
        Cached encryption configuration.
    """
    return load_encryption_config()

encrypt_pdf ¶

encrypt_pdf(file_path, context)

Encrypt a PDF with a password derived from client context.

Parameters:

Name	Type	Description	Default
`file_path`	`str`	Path to the PDF file to encrypt.	required
`context`	`dict`	Template context dict with client metadata (from build_client_context). Must contain fields referenced in the password template.	required

Returns:

Type	Description
`str`	Path to the encrypted PDF file with _encrypted suffix.

Raises:

Type	Description
`ValueError`	If password template references missing fields or is invalid.

Source code in pipeline/encrypt_notice.py

def encrypt_pdf(file_path: str, context: dict) -> str:
    """Encrypt a PDF with a password derived from client context.

    Parameters
    ----------
    file_path : str
        Path to the PDF file to encrypt.
    context : dict
        Template context dict with client metadata (from build_client_context).
        Must contain fields referenced in the password template.

    Returns
    -------
    str
        Path to the encrypted PDF file with _encrypted suffix.

    Raises
    ------
    ValueError
        If password template references missing fields or is invalid.
    """
    config = get_encryption_config()
    password_config = config.get("password", {})
    template = password_config.get("template", "{date_of_birth_iso_compact}")

    try:
        password = validate_and_format_template(
            template, context, allowed_fields=TemplateField.all_values()
        )
    except (KeyError, ValueError) as e:
        raise ValueError(f"Invalid password template: {e}") from e

    reader = PdfReader(file_path, strict=False)
    writer = PdfWriter()

    # Use pypdf's standard append method
    writer.append(reader)

    if reader.metadata:
        writer.add_metadata(reader.metadata)

    writer.encrypt(user_password=password, owner_password=password)

    src = Path(file_path)
    encrypted_path = src.with_name(f"{src.stem}_encrypted{src.suffix}")
    with open(encrypted_path, "wb") as f:
        writer.write(f)

    return str(encrypted_path)

load_notice_metadata ¶

load_notice_metadata(json_path)

Load client data dict and context from JSON notice metadata.

Module-internal helper for encrypt_notice(). Loads the JSON, extracts the client data dict, builds the templating context, and returns both.

Parameters:

Name	Type	Description	Default
`json_path`	`Path`	Path to JSON metadata file.	required

Returns:

Type	Description
`tuple`	(client_dict: dict, context: dict) for password generation.

Raises:

Type	Description
`ValueError`	If JSON is invalid or has unexpected structure.

Source code in pipeline/encrypt_notice.py

def load_notice_metadata(json_path: Path) -> tuple:
    """Load client data dict and context from JSON notice metadata.

    Module-internal helper for encrypt_notice(). Loads the JSON, extracts
    the client data dict, builds the templating context, and returns both.

    Parameters
    ----------
    json_path : Path
        Path to JSON metadata file.

    Returns
    -------
    tuple
        (client_dict: dict, context: dict) for password generation.

    Raises
    ------
    ValueError
        If JSON is invalid or has unexpected structure.
    """
    try:
        payload = json.loads(json_path.read_text())
    except json.JSONDecodeError as exc:
        raise ValueError(f"Invalid JSON structure ({json_path.name}): {exc}") from exc

    if not payload:
        raise ValueError(f"No client data in {json_path.name}")

    first_key = next(iter(payload))
    client_dict = payload[first_key]

    # Ensure record is a dict
    if not isinstance(client_dict, dict):
        raise ValueError(f"Invalid client record format in {json_path.name}")

    # Build context using shared helper
    context = build_client_context(client_dict)
    return client_dict, context

encrypt_notice ¶

encrypt_notice(json_path, pdf_path, language)

Encrypt a PDF notice using client data from the JSON file.

Returns the path to the encrypted PDF with _encrypted suffix. If the encrypted version already exists and is newer than the source, returns the existing file without re-encrypting.

Parameters:

Name	Type	Description	Default
`json_path`	`Path`	Path to the JSON file containing client metadata	required
`pdf_path`	`Path`	Path to the PDF file to encrypt	required
`language`	`str`	ISO 639-1 language code ('en' for English, 'fr' for French)	required

Returns:

Type	Description
`path`	Path to the encrypted PDF file

Raises:

Type	Description
`FileNotFoundError`	If JSON or PDF file not found
`ValueError`	If JSON is invalid

Source code in pipeline/encrypt_notice.py

def encrypt_notice(json_path: str | Path, pdf_path: str | Path, language: str) -> str:
    """Encrypt a PDF notice using client data from the JSON file.

    Returns the path to the encrypted PDF with _encrypted suffix.
    If the encrypted version already exists and is newer than the source,
    returns the existing file without re-encrypting.

    Parameters
    ----------
    json_path : Path
        Path to the JSON file containing client metadata
    pdf_path : Path
        Path to the PDF file to encrypt
    language: str
        ISO 639-1 language code ('en' for English, 'fr' for French)

    Returns
    -------
    path
        Path to the encrypted PDF file

    Raises
    ------
    FileNotFoundError
        If JSON or PDF file not found
    ValueError
        If JSON is invalid
    """
    json_path = Path(json_path)
    pdf_path = Path(pdf_path)

    if not json_path.exists():
        raise FileNotFoundError(f"JSON file not found: {json_path}")
    if not pdf_path.exists():
        raise FileNotFoundError(f"PDF file not found: {pdf_path}")

    encrypted_path = pdf_path.with_name(f"{pdf_path.stem}_encrypted{pdf_path.suffix}")
    if encrypted_path.exists():
        try:
            if encrypted_path.stat().st_mtime >= pdf_path.stat().st_mtime:
                return str(encrypted_path)
        except OSError:
            pass

    client_data, context = load_notice_metadata(json_path)
    return encrypt_pdf(str(pdf_path), context)

encrypt_pdfs_in_directory ¶

encrypt_pdfs_in_directory(pdf_directory, json_file, language)

Encrypt all PDF notices in a directory using a combined JSON metadata file.

The JSON file should contain a dict where keys are client identifiers and values contain client metadata with DOB information.

PDFs are encrypted in-place with the _encrypted suffix added to filename.

Parameters:

Name	Type	Description	Default
`pdf_directory`	`str`	Directory containing PDF files to encrypt	required
`json_file`	`Path`	Path to the combined JSON file with all client metadata	required
`language`	`str`	ISO 639-1 language code ('en' for English, 'fr' for French)	required

Raises:

Type	Description
`FileNotFoundError`	If PDF directory or JSON file don't exist

Source code in pipeline/encrypt_notice.py

def encrypt_pdfs_in_directory(
    pdf_directory: Path,
    json_file: Path,
    language: str,
) -> None:
    """Encrypt all PDF notices in a directory using a combined JSON metadata file.

    The JSON file should contain a dict where keys are client identifiers and
    values contain client metadata with DOB information.

    PDFs are encrypted in-place with the _encrypted suffix added to filename.

    Parameters
    ----------
    pdf_directory : str
        Directory containing PDF files to encrypt
    json_file : Path
        Path to the combined JSON file with all client metadata
    language :str
        ISO 639-1 language code ('en' for English, 'fr' for French)

    Raises
    ------
    FileNotFoundError
        If PDF directory or JSON file don't exist
    """
    pdf_directory = Path(pdf_directory)
    json_file = Path(json_file)

    if not pdf_directory.exists():
        raise FileNotFoundError(f"PDF directory not found: {pdf_directory}")
    if not json_file.exists():
        raise FileNotFoundError(f"JSON file not found: {json_file}")

    # Load the combined metadata
    try:
        metadata = json.loads(json_file.read_text())
    except json.JSONDecodeError as exc:
        raise ValueError(f"Invalid JSON in {json_file.name}: {exc}") from exc

    # Extract clients from the metadata
    # Handle both preprocessed artifact format (has 'clients' key) and dict of clients
    if isinstance(metadata, dict) and "clients" in metadata:
        clients_data = metadata["clients"]
    else:
        clients_data = metadata

    if not clients_data:
        print("No client data found in JSON file.")
        return

    # Build a lookup dict: client_id -> client_data
    client_lookup = {}
    if isinstance(clients_data, list):
        # Format: list of client dicts with 'client_id' field
        for client in clients_data:
            client_id = client.get("client_id")
            if client_id:
                client_lookup[str(client_id)] = client
    elif isinstance(clients_data, dict):
        # Format: dict keyed by client_id
        client_lookup = {str(k): v for k, v in clients_data.items()}

    # Find PDFs and encrypt them
    pdf_files = sorted(pdf_directory.glob("*.pdf"))
    if not pdf_files:
        print("No PDFs found for encryption.")
        return

    start = time.perf_counter()
    print(
        f"🔐 Encrypting {len(pdf_files)} notices...",
        flush=True,
    )

    successes = 0
    skipped: List[Tuple[str, str]] = []
    failures: List[Tuple[str, str]] = []

    for pdf_path in pdf_files:
        pdf_name = pdf_path.name
        stem = pdf_path.stem

        # Skip conf and already-encrypted files
        if stem == "conf" or stem.endswith("_encrypted"):
            continue

        # Extract client_id from filename (format: en_client_XXXXX_YYYYYYY)
        # The last part after the last underscore is the client_id (OEN)
        parts = stem.split("_")
        if len(parts) >= 3:
            client_id = parts[-1]
        else:
            skipped.append((pdf_name, "Could not extract client_id from filename"))
            continue

        # Look up client data
        client_data = client_lookup.get(client_id)
        if not client_data:
            skipped.append((pdf_name, f"No metadata found for client_id {client_id}"))
            continue

        # Build context directly from client dict using shared helper
        try:
            context = build_client_context(client_data)
        except (ValueError, KeyError) as exc:
            skipped.append((pdf_name, str(exc)))
            continue

        # Encrypt the PDF
        try:
            encrypted_path = pdf_path.with_name(
                f"{pdf_path.stem}_encrypted{pdf_path.suffix}"
            )

            # Skip if encrypted version is newer than source
            if encrypted_path.exists():
                try:
                    if encrypted_path.stat().st_mtime >= pdf_path.stat().st_mtime:
                        successes += 1
                        continue
                except OSError:
                    pass

            encrypt_pdf(str(pdf_path), context)
            # Unencrypted PDF is preserved; deletion is handled in cleanup step
            successes += 1
        except Exception as exc:
            failures.append((pdf_name, str(exc)))

    duration = time.perf_counter() - start
    print(
        f"✅ Encryption complete in {duration:.2f}s "
        f"(success: {successes}, skipped: {len(skipped)}, failed: {len(failures)})"
    )

    for pdf_name, reason in skipped:
        print(f"SKIP: {pdf_name} -> {reason}")

    for pdf_name, reason in failures:
        print(f"WARNING: Encryption failed for {pdf_name}: {reason}")

Step 8 — PDF Bundling¶

pipeline.bundle_pdfs ¶

Bundle per-client PDFs into combined files with manifests.

This module combines individual per-client PDFs into bundled files with accompanying manifest records. It can be invoked as a CLI tool or imported for unit testing. Bundling supports three modes:

Size-based (default): bundle the clients into fixed-size groups, i.e., 100 per bundle.
School-based: group the clients by school code then bundle the clients into fixed-sized groups
Board-based: group the clients by board code then bundle the clients into fixed-sized groups

Each bundle produces a merged PDF inside output/pdf_combined and a manifest JSON record inside output/metadata that captures critical metadata for audits.

Input Contract:

Reads individual PDF files from output/pdf_individual/
Reads client metadata from preprocessed artifact JSON
Assumes bundle_size > 0 in config (bundling is optional; disabled when bundle_size=0)

Output Contract:

Writes merged PDF files to output/pdf_combined/
Writes bundle manifest JSON to output/metadata/
Returns list of created bundle files

Error Handling:

Configuration errors (invalid bundle_size, group_by) raise immediately (infrastructure)
Per-bundle errors (PDF merge failure) log and continue (optional feature)
Pipeline completes even if some bundles fail to create (optional step)

Validation Contract:

What this module validates:

Bundle size is positive (bundle_size > 0)
Group-by strategy is valid (size, school, board, or None)
PDF files can be discovered and merged
Manifest records have required metadata

What this module assumes (validated upstream):

PDF files are valid and readable (validated by count_pdfs step)
Client metadata in artifact is complete (validated by preprocessing step)
Output directory can be created (general I/O)

Note: This is an optional step. Per-bundle errors are logged but don't halt pipeline.

BundleConfig `dataclass` ¶

Configuration for PDF bundling operation.

Attributes:

Name	Type	Description
`output_dir`	`Path`	Root output directory containing pipeline artifacts
`language`	`str`	Language code ('en' or 'fr')
`bundle_size`	`int`	Maximum number of clients per bundle (0 disables bundling)
`bundle_strategy`	`BundleStrategy`	Strategy for grouping PDFs into bundles
`run_id`	`str`	Pipeline run identifier

Source code in pipeline/bundle_pdfs.py

@dataclass(frozen=True)
class BundleConfig:
    """Configuration for PDF bundling operation.

    Attributes
    ----------
    output_dir : Path
        Root output directory containing pipeline artifacts
    language : str
        Language code ('en' or 'fr')
    bundle_size : int
        Maximum number of clients per bundle (0 disables bundling)
    bundle_strategy : BundleStrategy
        Strategy for grouping PDFs into bundles
    run_id : str
        Pipeline run identifier
    """

    output_dir: Path
    language: str
    bundle_size: int
    bundle_strategy: BundleStrategy
    run_id: str

BundlePlan `dataclass` ¶

Plan for a single bundle of PDFs.

Attributes:

Name	Type	Description
`bundle_type`	`BundleType`	Type/strategy used for this bundle
`bundle_identifier`	`str \| None`	School or board code if bundle was grouped, None for size-based
`bundle_number`	`int`	Sequential bundle number
`total_bundles`	`int`	Total number of bundles in this operation
`clients`	`List[PdfRecord]`	List of PDFs and metadata in this bundle

Source code in pipeline/bundle_pdfs.py

@dataclass(frozen=True)
class BundlePlan:
    """Plan for a single bundle of PDFs.

    Attributes
    ----------
    bundle_type : BundleType
        Type/strategy used for this bundle
    bundle_identifier : str | None
        School or board code if bundle was grouped, None for size-based
    bundle_number : int
        Sequential bundle number
    total_bundles : int
        Total number of bundles in this operation
    clients : List[PdfRecord]
        List of PDFs and metadata in this bundle
    """

    bundle_type: BundleType
    bundle_identifier: str | None
    bundle_number: int
    total_bundles: int
    clients: List[PdfRecord]

BundleResult `dataclass` ¶

Result of a completed bundle operation.

Attributes:

Name	Type	Description
`pdf_path`	`Path`	Path to the merged PDF file
`manifest_path`	`Path`	Path to the JSON manifest file
`bundle_plan`	`BundlePlan`	The plan used to create this bundle

Source code in pipeline/bundle_pdfs.py

@dataclass(frozen=True)
class BundleResult:
    """Result of a completed bundle operation.

    Attributes
    ----------
    pdf_path : Path
        Path to the merged PDF file
    manifest_path : Path
        Path to the JSON manifest file
    bundle_plan : BundlePlan
        The plan used to create this bundle
    """

    pdf_path: Path
    manifest_path: Path
    bundle_plan: BundlePlan

bundle_pdfs_with_config ¶

bundle_pdfs_with_config(output_dir, language, run_id, config_path=None)

Bundle PDFs using configuration from parameters.yaml.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root output directory containing pipeline artifacts.	required
`language`	`str`	Language prefix to bundle ('en' or 'fr').	required
`run_id`	`str`	Pipeline run identifier to locate preprocessing artifacts.	required
`config_path`	`Path`	Path to parameters.yaml. If not provided, uses default location.	`None`

Returns:

Type	Description
`List[BundleResult]`	List of bundle results created.

Source code in pipeline/bundle_pdfs.py

def bundle_pdfs_with_config(
    output_dir: Path,
    language: str,
    run_id: str,
    config_path: Path | None = None,
) -> List[BundleResult]:
    """Bundle PDFs using configuration from parameters.yaml.

    Parameters
    ----------
    output_dir : Path
        Root output directory containing pipeline artifacts.
    language : str
        Language prefix to bundle ('en' or 'fr').
    run_id : str
        Pipeline run identifier to locate preprocessing artifacts.
    config_path : Path, optional
        Path to parameters.yaml. If not provided, uses default location.

    Returns
    -------
    List[BundleResult]
        List of bundle results created.
    """
    config = load_config(config_path)

    bundling_config = config.get("bundling", {})
    bundle_size = bundling_config.get("bundle_size", 0)
    group_by = bundling_config.get("group_by", None)

    bundle_strategy = BundleStrategy.from_string(group_by)

    config_obj = BundleConfig(
        output_dir=output_dir.resolve(),
        language=language,
        bundle_size=bundle_size,
        bundle_strategy=bundle_strategy,
        run_id=run_id,
    )

    return bundle_pdfs(config_obj)

main ¶

main(output_dir, language, run_id, config_path=None)

Main entry point for PDF bundling.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root output directory containing pipeline artifacts.	required
`language`	`str`	Language prefix to bundle ('en' or 'fr').	required
`run_id`	`str`	Pipeline run identifier.	required
`config_path`	`Path`	Path to parameters.yaml configuration file.	`None`

Returns:

Type	Description
`List[BundleResult]`	List of bundles created.

Source code in pipeline/bundle_pdfs.py

def main(
    output_dir: Path, language: str, run_id: str, config_path: Path | None = None
) -> List[BundleResult]:
    """Main entry point for PDF bundling.

    Parameters
    ----------
    output_dir : Path
        Root output directory containing pipeline artifacts.
    language : str
        Language prefix to bundle ('en' or 'fr').
    run_id : str
        Pipeline run identifier.
    config_path : Path, optional
        Path to parameters.yaml configuration file.

    Returns
    -------
    List[BundleResult]
        List of bundles created.
    """
    results = bundle_pdfs_with_config(output_dir, language, run_id, config_path)
    if results:
        print(f"Created {len(results)} bundles in {output_dir / 'pdf_combined'}")
    else:
        print("No bundles created.")
    return results

chunked ¶

chunked(iterable, size)

Split an iterable into fixed-size chunks.

Parameters:

Name	Type	Description	Default
`iterable`	`Sequence[T]`	Sequence to chunk.	required
`size`	`int`	Maximum number of items per chunk (must be positive).	required

Returns:

Type	Description
`Iterator[List[T]]`	Iterator yielding lists of up to `size` items.

Raises:

Type	Description
`ValueError`	If size is not positive.

Examples:

>>> list(chunked([1, 2, 3, 4, 5], 2))
[[1, 2], [3, 4], [5]]

Source code in pipeline/bundle_pdfs.py

def chunked(iterable: Sequence[T], size: int) -> Iterator[List[T]]:
    """Split an iterable into fixed-size chunks.

    Parameters
    ----------
    iterable : Sequence[T]
        Sequence to chunk.
    size : int
        Maximum number of items per chunk (must be positive).

    Returns
    -------
    Iterator[List[T]]
        Iterator yielding lists of up to `size` items.

    Raises
    ------
    ValueError
        If size is not positive.

    Examples
    --------
    >>> list(chunked([1, 2, 3, 4, 5], 2))
    [[1, 2], [3, 4], [5]]
    """
    if size <= 0:
        raise ValueError("chunk size must be positive")
    for index in range(0, len(iterable), size):
        yield list(islice(iterable, index, index + size))

slugify ¶

slugify(value)

Convert a string to a URL-safe slug format.

Converts spaces and special characters to underscores, removes consecutive underscores, and lowercases the result. Used for generating bundle filenames from school/board names.

Parameters:

Name	Type	Description	Default
`value`	`str`	String to slugify (e.g., school or board name).	required

Returns:

Type	Description
`str`	Slugified string, or 'unknown' if value is empty/whitespace.

Examples:

>>> slugify("Lincoln High School")
'lincoln_high_school'
>>> slugify("Bd. Métropolitain")
'bd_m_tropolitain'

Source code in pipeline/bundle_pdfs.py

def slugify(value: str) -> str:
    """Convert a string to a URL-safe slug format.

    Converts spaces and special characters to underscores, removes consecutive
    underscores, and lowercases the result. Used for generating bundle filenames
    from school/board names.

    Parameters
    ----------
    value : str
        String to slugify (e.g., school or board name).

    Returns
    -------
    str
        Slugified string, or 'unknown' if value is empty/whitespace.

    Examples
    --------
    >>> slugify("Lincoln High School")
    'lincoln_high_school'
    >>> slugify("Bd. Métropolitain")
    'bd_m_tropolitain'
    """
    cleaned = re.sub(r"[^A-Za-z0-9]+", "_", value.strip())
    return re.sub(r"_+", "_", cleaned).strip("_").lower() or "unknown"

load_artifact ¶

load_artifact(output_dir, run_id)

Load the preprocessed artifact JSON from the output directory.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root output directory containing artifacts.	required
`run_id`	`str`	Pipeline run identifier matching the artifact filename.	required

Returns:

Type	Description
`Dict[str, object]`	Parsed preprocessed artifact with clients and metadata.

Raises:

Type	Description
`FileNotFoundError`	If the preprocessed artifact file does not exist.

Source code in pipeline/bundle_pdfs.py

def load_artifact(output_dir: Path, run_id: str) -> Dict[str, object]:
    """Load the preprocessed artifact JSON from the output directory.

    Parameters
    ----------
    output_dir : Path
        Root output directory containing artifacts.
    run_id : str
        Pipeline run identifier matching the artifact filename.

    Returns
    -------
    Dict[str, object]
        Parsed preprocessed artifact with clients and metadata.

    Raises
    ------
    FileNotFoundError
        If the preprocessed artifact file does not exist.
    """
    artifact_path = output_dir / "artifacts" / f"preprocessed_clients_{run_id}.json"
    if not artifact_path.exists():
        raise FileNotFoundError(f"Preprocessed artifact not found at {artifact_path}")
    payload = json.loads(artifact_path.read_text(encoding="utf-8"))
    return payload

build_client_lookup ¶

build_client_lookup(artifact)

Build a lookup table from artifact clients dict.

Parameters:

Name	Type	Description	Default
`artifact`	`Dict[str, object]`	Preprocessed artifact dictionary	required

Returns:

Type	Description
`Dict[tuple[str, str], dict]`	Lookup table keyed by (sequence, client_id)

Source code in pipeline/bundle_pdfs.py

def build_client_lookup(
    artifact: Dict[str, object],
) -> Dict[tuple[str, str], dict]:
    """Build a lookup table from artifact clients dict.

    Parameters
    ----------
    artifact : Dict[str, object]
        Preprocessed artifact dictionary

    Returns
    -------
    Dict[tuple[str, str], dict]
        Lookup table keyed by (sequence, client_id)
    """
    clients_obj = artifact.get("clients", [])
    clients = clients_obj if isinstance(clients_obj, list) else []
    lookup: Dict[tuple[str, str], dict] = {}
    for client in clients:
        sequence = client.get("sequence")  # type: ignore[attr-defined]
        client_id = client.get("client_id")  # type: ignore[attr-defined]
        lookup[(sequence, client_id)] = client  # type: ignore[typeddict-item]
    return lookup

discover_pdfs ¶

discover_pdfs(output_dir, language)

Discover all individual PDF files for a given language.

Discovers non-encrypted PDF files only. Encrypted PDFs (with _encrypted suffix) are excluded from bundling since bundling operates on the original unencrypted PDFs.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root output directory.	required
`language`	`str`	Language prefix to match (e.g., 'en' or 'fr').	required

Returns:

Type	Description
`List[Path]`	Sorted list of non-encrypted PDF file paths matching the language, or empty list if pdf_individual directory doesn't exist.

Source code in pipeline/bundle_pdfs.py

def discover_pdfs(output_dir: Path, language: str) -> List[Path]:
    """Discover all individual PDF files for a given language.

    Discovers non-encrypted PDF files only. Encrypted PDFs (with _encrypted suffix)
    are excluded from bundling since bundling operates on the original unencrypted PDFs.

    Parameters
    ----------
    output_dir : Path
        Root output directory.
    language : str
        Language prefix to match (e.g., 'en' or 'fr').

    Returns
    -------
    List[Path]
        Sorted list of non-encrypted PDF file paths matching the language, or empty list
        if pdf_individual directory doesn't exist.
    """
    pdf_dir = output_dir / "pdf_individual"
    if not pdf_dir.exists():
        return []
    # Exclude encrypted PDFs (those with _encrypted suffix)
    all_pdfs = pdf_dir.glob(f"{language}_notice_*.pdf")
    return sorted([p for p in all_pdfs if not p.stem.endswith("_encrypted")])

build_pdf_records ¶

build_pdf_records(output_dir, language, clients)

Build a list of PdfRecord objects from discovered PDF files.

Discovers PDFs, extracts metadata from filenames, looks up client data, and constructs PdfRecord objects with page counts and client metadata.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root output directory.	required
`language`	`str`	Language prefix to filter PDFs.	required
`clients`	`Dict[tuple[str, str], dict]`	Lookup table of client data keyed by (sequence, client_id).	required

Returns:

Type	Description
`List[PdfRecord]`	Sorted list of PdfRecord objects by sequence.

Raises:

Type	Description
`KeyError`	If a PDF filename has no matching client in the lookup table.

Source code in pipeline/bundle_pdfs.py

def build_pdf_records(
    output_dir: Path, language: str, clients: Dict[tuple[str, str], dict]
) -> List[PdfRecord]:
    """Build a list of PdfRecord objects from discovered PDF files.

    Discovers PDFs, extracts metadata from filenames, looks up client data,
    and constructs PdfRecord objects with page counts and client metadata.

    Parameters
    ----------
    output_dir : Path
        Root output directory.
    language : str
        Language prefix to filter PDFs.
    clients : Dict[tuple[str, str], dict]
        Lookup table of client data keyed by (sequence, client_id).

    Returns
    -------
    List[PdfRecord]
        Sorted list of PdfRecord objects by sequence.

    Raises
    ------
    KeyError
        If a PDF filename has no matching client in the lookup table.
    """
    pdf_paths = discover_pdfs(output_dir, language)
    records: List[PdfRecord] = []
    for pdf_path in pdf_paths:
        match = PDF_PATTERN.match(pdf_path.name)
        if not match:
            LOG.warning("Skipping unexpected PDF filename: %s", pdf_path.name)
            continue
        sequence = match.group("sequence")
        client_id = match.group("client_id")
        key = (sequence, client_id)
        if key not in clients:
            raise KeyError(f"No client metadata found for PDF {pdf_path.name}")
        reader = PdfReader(str(pdf_path))
        page_count = len(reader.pages)
        records.append(
            PdfRecord(
                sequence=sequence,
                client_id=client_id,
                pdf_path=pdf_path,
                page_count=page_count,
                client=clients[key],
            )
        )
    return sorted(records, key=lambda record: record.sequence)

plan_bundles ¶

plan_bundles(config, records, log_path)

Plan how to group PDFs into bundles based on configuration.

Parameters:

Name	Type	Description	Default
`config`	`BundleConfig`	Bundling configuration including strategy and bundle size	required
`records`	`List[PdfRecord]`	List of PDF records to bundle	required
`log_path`	`Path`	Path to logging file	required

Returns:

Type	Description
`List[BundlePlan]`	List of bundle plans

Source code in pipeline/bundle_pdfs.py

def plan_bundles(
    config: BundleConfig, records: List[PdfRecord], log_path: Path
) -> List[BundlePlan]:
    """Plan how to group PDFs into bundles based on configuration.

    Parameters
    ----------
    config : BundleConfig
        Bundling configuration including strategy and bundle size
    records : List[PdfRecord]
        List of PDF records to bundle
    log_path : Path
        Path to logging file

    Returns
    -------
    List[BundlePlan]
        List of bundle plans
    """
    if config.bundle_size <= 0:
        return []

    plans: List[BundlePlan] = []

    if config.bundle_strategy == BundleStrategy.SCHOOL:
        ensure_ids(records, attr="school", log_path=log_path)
        grouped = group_records(records, "school")
        for identifier, items in grouped.items():
            total_bundles = (len(items) + config.bundle_size - 1) // config.bundle_size
            for index, chunk in enumerate(chunked(items, config.bundle_size), start=1):
                plans.append(
                    BundlePlan(
                        bundle_type=BundleType.SCHOOL_GROUPED,
                        bundle_identifier=identifier,
                        bundle_number=index,
                        total_bundles=total_bundles,
                        clients=chunk,
                    )
                )
        return plans

    if config.bundle_strategy == BundleStrategy.BOARD:
        ensure_ids(records, attr="board", log_path=log_path)
        grouped = group_records(records, "board")
        for identifier, items in grouped.items():
            total_bundles = (len(items) + config.bundle_size - 1) // config.bundle_size
            for index, chunk in enumerate(chunked(items, config.bundle_size), start=1):
                plans.append(
                    BundlePlan(
                        bundle_type=BundleType.BOARD_GROUPED,
                        bundle_identifier=identifier,
                        bundle_number=index,
                        total_bundles=total_bundles,
                        clients=chunk,
                    )
                )
        return plans

    # Size-based bundling (default)
    total_bundles = (len(records) + config.bundle_size - 1) // config.bundle_size
    for index, chunk in enumerate(chunked(records, config.bundle_size), start=1):
        plans.append(
            BundlePlan(
                bundle_type=BundleType.SIZE_BASED,
                bundle_identifier=None,
                bundle_number=index,
                total_bundles=total_bundles,
                clients=chunk,
            )
        )
    return plans

relative ¶

relative(path, root)

Convert path to string relative to root directory.

Module-internal helper for manifest generation. Creates relative path strings for storing in JSON manifests, making paths portable across different base directories.

Parameters:

Name	Type	Description	Default
`path`	`Path`	Absolute path to convert.	required
`root`	`Path`	Root directory to compute relative path from.	required

Returns:

Type	Description
`str`	Relative path as POSIX string.

Source code in pipeline/bundle_pdfs.py

def relative(path: Path, root: Path) -> str:
    """Convert path to string relative to root directory.

    Module-internal helper for manifest generation. Creates relative path strings
    for storing in JSON manifests, making paths portable across different base directories.

    Parameters
    ----------
    path : Path
        Absolute path to convert.
    root : Path
        Root directory to compute relative path from.

    Returns
    -------
    str
        Relative path as POSIX string.
    """
    try:
        return str(path.relative_to(root))
    except ValueError:
        return str(path)

Step 9 — Cleanup¶

pipeline.cleanup ¶

Cleanup module for Step 9: removing intermediate pipeline artifacts.

This step removes intermediate files generated during the pipeline run to reduce storage footprint. Configuration is read from parameters.yaml under pipeline.after_run.

This is distinct from Step 1 (prepare_output), which uses pipeline.before_run.clear_output_directory to clean up old pipeline runs at startup while preserving logs.

Step 1 Configuration (pipeline.before_run in parameters.yaml):

clear_output_directory: when true, removes all output except logs before starting a new run

Step 9 Configuration (pipeline.after_run in parameters.yaml):

remove_artifacts: when true, removes output/artifacts directory
remove_unencrypted_pdfs: when true and (encryption OR batching) is enabled, removes non-encrypted PDFs from pdf_individual/ after encryption completes. If both encryption and batching are disabled, individual non-encrypted PDFs are assumed to be final output and are preserved.

Input Contract:

Reads configuration from parameters.yaml (pipeline.after_run section)
Assumes output directory structure exists (may be partially populated)
Assumes encryption.enabled and bundling.bundle_size from parameters.yaml

Output Contract:

Removes specified directories from output_dir
Removes unencrypted PDFs if conditions are met:
- remove_unencrypted_pdfs=true AND (encryption enabled OR batching enabled)
Does not modify final PDF outputs (unless configured to do so)
Does not halt pipeline if cleanup fails

Error Handling:

File deletion errors are logged and continue (optional step)
Missing directories/files don't cause errors (idempotent)
Pipeline completes even if cleanup partially fails (utility step)

Validation Contract:

What this module validates:

Output directory exists and is writable
Directory/file paths can be safely deleted (exist check before delete)
Configuration values are sensible boolean types and integers

What this module assumes (validated upstream):

Configuration keys are valid and well-formed
Output directory structure is correct (created by prior steps)

Note: This is a utility/cleanup step. Failures don't halt pipeline.

safe_delete ¶

safe_delete(path)

Safely delete a file or directory if it exists.

Parameters:

Name	Type	Description	Default
`path`	`Path`	File or directory to delete.	required

Source code in pipeline/cleanup.py

def safe_delete(path: Path):
    """Safely delete a file or directory if it exists.

    Parameters
    ----------
    path : Path
        File or directory to delete.
    """
    if path.exists():
        if path.is_dir():
            shutil.rmtree(path)
        else:
            path.unlink()

cleanup_with_config ¶

cleanup_with_config(output_dir, config_path=None)

Perform cleanup using configuration from parameters.yaml.

Reads Step 9 (after_run) cleanup configuration from parameters.yaml. This is separate from Step 1's before_run.clear_output_directory setting, which cleans old runs at pipeline start (preserving logs).

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root output directory containing generated files.	required
`config_path`	`Path`	Path to parameters.yaml. If not provided, uses default location.	`None`

Source code in pipeline/cleanup.py

def cleanup_with_config(output_dir: Path, config_path: Path | None = None) -> None:
    """Perform cleanup using configuration from parameters.yaml.

    Reads Step 9 (after_run) cleanup configuration from parameters.yaml.
    This is separate from Step 1's before_run.clear_output_directory setting, which cleans
    old runs at pipeline start (preserving logs).

    Parameters
    ----------
    output_dir : Path
        Root output directory containing generated files.
    config_path : Path, optional
        Path to parameters.yaml. If not provided, uses default location.
    """
    config = load_config(config_path)
    pipeline_config = config.get("pipeline", {})
    after_run_config = pipeline_config.get("after_run", {})
    encryption_enabled = config.get("encryption", {}).get("enabled", False)
    bundling_config = config.get("bundling", {})
    bundle_size = bundling_config.get("bundle_size", 0)
    batching_enabled = bundle_size > 0

    remove_artifacts = after_run_config.get("remove_artifacts", False)
    remove_unencrypted = after_run_config.get("remove_unencrypted_pdfs", False)

    # Remove artifacts directory if configured
    if remove_artifacts:
        safe_delete(output_dir / "artifacts")

    # Delete unencrypted PDFs if:
    # - remove_unencrypted_pdfs is True AND
    # - (encryption is enabled OR batching is enabled)
    # If both encryption and batching are disabled, assume we want the individual non-encrypted PDFs
    if remove_unencrypted and (encryption_enabled or batching_enabled):
        pdf_dir = output_dir / "pdf_individual"
        if pdf_dir.exists():
            for pdf_file in pdf_dir.glob("*.pdf"):
                # Only delete non-encrypted PDFs (skip _encrypted versions)
                if not pdf_file.stem.endswith("_encrypted"):
                    safe_delete(pdf_file)

main ¶

main(output_dir, config_path=None)

Main entry point for cleanup.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Root output directory to clean.	required
`config_path`	`Path`	Path to parameters.yaml configuration file.	`None`

Source code in pipeline/cleanup.py

def main(output_dir: Path, config_path: Path | None = None) -> None:
    """Main entry point for cleanup.

    Parameters
    ----------
    output_dir : Path
        Root output directory to clean.
    config_path : Path, optional
        Path to parameters.yaml configuration file.
    """
    if not output_dir.is_dir():
        raise ValueError(f"The path {output_dir} is not a valid directory.")

    cleanup_with_config(output_dir, config_path)

Utilities¶

pipeline.utils ¶

Utility functions for immunization pipeline processing.

Provides template rendering utilities and context building functions shared across pipeline steps, particularly for QR code generation, PDF encryption, and template variable substitution. All functions handle string conversions and safe formatting of client data for use in downstream templates.

string_or_empty ¶

string_or_empty(value)

Safely convert value to string, returning empty string for None/NaN.

Parameters:

Name	Type	Description	Default
`value`	`Any`	Value to convert (may be None, empty string, or any type)	required

Returns:

Type	Description
`str`	Stringified value or empty string for None/NaN values

Source code in pipeline/utils.py

def string_or_empty(value: Any) -> str:
    """Safely convert value to string, returning empty string for None/NaN.

    Parameters
    ----------
    value : Any
        Value to convert (may be None, empty string, or any type)

    Returns
    -------
    str
        Stringified value or empty string for None/NaN values
    """
    if value is None:
        return ""
    return str(value).strip()

extract_template_fields ¶

extract_template_fields(template)

Extract placeholder names from a format string template.

Parameters:

Name	Type	Description	Default
`template`	`str`	Format string like "https://example.com?id={client_id}&dob={date_of_birth_iso}"	required

Returns:

Type	Description
`set[str]`	Set of placeholder names found in template

Raises:

Type	Description
`ValueError`	If template contains invalid format string syntax

Examples:

>>> extract_template_fields("{client_id}_{date_of_birth_iso}")
{'client_id', 'date_of_birth_iso'}

Source code in pipeline/utils.py

def extract_template_fields(template: str) -> set[str]:
    """Extract placeholder names from a format string template.

    Parameters
    ----------
    template : str
        Format string like "https://example.com?id={client_id}&dob={date_of_birth_iso}"

    Returns
    -------
    set[str]
        Set of placeholder names found in template

    Raises
    ------
    ValueError
        If template contains invalid format string syntax

    Examples
    --------
    >>> extract_template_fields("{client_id}_{date_of_birth_iso}")
    {'client_id', 'date_of_birth_iso'}
    """
    try:
        return {
            field_name
            for _, field_name, _, _ in _FORMATTER.parse(template)
            if field_name
        }
    except ValueError as exc:
        raise ValueError(f"Invalid template format: {exc}") from exc

validate_and_format_template ¶

validate_and_format_template(template, context, allowed_fields=None)

Format template and validate placeholders against allowed set.

Ensures that:

All placeholders in template exist in context
All placeholders are in the allowed_fields set (if provided)
Template is successfully rendered

Parameters:

Name	Type	Description	Default
`template`	`str`	Format string template with placeholders	required
`context`	`dict[str, str]`	Context dict with placeholder values	required
`allowed_fields`	`set[str] \| None`	Set of allowed placeholder names. If None, allows any placeholder that exists in context.	`None`

Returns:

Type	Description
`str`	Rendered template

Raises:

Type	Description
`KeyError`	If template contains placeholders not in context
`ValueError`	If template contains disallowed placeholders (when allowed_fields provided)

Examples:

>>> ctx = {"client_id": "12345", "date_of_birth_iso": "2015-03-15"}
>>> validate_and_format_template(
...     "{client_id}_{date_of_birth_iso}",
...     ctx,
...     allowed_fields={"client_id", "date_of_birth_iso"}
... )
'12345_2015-03-15'

Source code in pipeline/utils.py

def validate_and_format_template(
    template: str,
    context: dict[str, str],
    allowed_fields: set[str] | None = None,
) -> str:
    """Format template and validate placeholders against allowed set.

    Ensures that:

    1. All placeholders in template exist in context
    2. All placeholders are in the allowed_fields set (if provided)
    3. Template is successfully rendered

    Parameters
    ----------
    template : str
        Format string template with placeholders
    context : dict[str, str]
        Context dict with placeholder values
    allowed_fields : set[str] | None
        Set of allowed placeholder names. If None, allows any placeholder
        that exists in context.

    Returns
    -------
    str
        Rendered template

    Raises
    ------
    KeyError
        If template contains placeholders not in context
    ValueError
        If template contains disallowed placeholders (when allowed_fields provided)

    Examples
    --------
    >>> ctx = {"client_id": "12345", "date_of_birth_iso": "2015-03-15"}
    >>> validate_and_format_template(
    ...     "{client_id}_{date_of_birth_iso}",
    ...     ctx,
    ...     allowed_fields={"client_id", "date_of_birth_iso"}
    ... )
    '12345_2015-03-15'
    """
    placeholders = extract_template_fields(template)

    # Check for missing placeholders in context
    unknown_fields = placeholders - context.keys()
    if unknown_fields:
        raise KeyError(
            f"Unknown placeholder(s) {sorted(unknown_fields)} in template. "
            f"Available: {sorted(context.keys())}"
        )

    # Check for disallowed placeholders (if whitelist provided)
    if allowed_fields is not None:
        disallowed = placeholders - allowed_fields
        if disallowed:
            raise ValueError(
                f"Disallowed placeholder(s) {sorted(disallowed)} in template. "
                f"Allowed: {sorted(allowed_fields)}"
            )

    return template.format(**context)

build_client_context ¶

build_client_context(client_data, language=None)

Build template context dict from client metadata for templating.

Extracts and formats all available client fields for use in templates, supporting both QR code payloads and PDF encryption passwords.

Accepts either a dict (from JSON) or a ClientRecord dataclass instance. Both provide the same fields; the function handles both transparently.

Parameters:

Name	Type	Description	Default
`client_data`	`dict or ClientRecord`	Client data as either: - A dict (from preprocessed artifact JSON) with nested structure: { "client_id": "...", "person": {"first_name": "...", "last_name": "...", "date_of_birth_iso": "..."}, "school": {"name": "..."}, "board": {"name": "..."}, "contact": {"postal_code": "...", "city": "...", ...} } - A ClientRecord dataclass instance with same nested fields.	required
`language`	`str`	ISO 639-1 language code ('en' or 'fr'). When omitted, falls back to the client's own language field if present, otherwise an empty string.	`None`

Returns:

Type	Description
`dict[str, str]`	Context dict with keys: - client_id - first_name, last_name, name - date_of_birth (display format) - date_of_birth_iso (YYYY-MM-DD) - date_of_birth_iso_compact (YYYYMMDD) - school, board - postal_code, city, province, street_address - language_code ('en' or 'fr')

Examples:

>>> client_dict = {
...     "client_id": "12345",
...     "person": {"first_name": "John", "last_name": "Doe", "date_of_birth_iso": "2015-03-15"},
...     "school": {"name": "Lincoln School"},
...     "contact": {"postal_code": "M5V 3A8"}
... }
>>> ctx = build_client_context(client_dict)
>>> ctx["client_id"]
'12345'
>>> ctx["first_name"]
'John'

Source code in pipeline/utils.py

def build_client_context(
    client_data,
    language: str | None = None,
) -> dict[str, str]:
    """Build template context dict from client metadata for templating.

    Extracts and formats all available client fields for use in templates,
    supporting both QR code payloads and PDF encryption passwords.

    Accepts either a dict (from JSON) or a ClientRecord dataclass instance.
    Both provide the same fields; the function handles both transparently.

    Parameters
    ----------
    client_data : dict or ClientRecord
        Client data as either:
        - A dict (from preprocessed artifact JSON) with nested structure:
          {
              "client_id": "...",
              "person": {"first_name": "...", "last_name": "...", "date_of_birth_iso": "..."},
              "school": {"name": "..."},
              "board": {"name": "..."},
              "contact": {"postal_code": "...", "city": "...", ...}
          }
        - A ClientRecord dataclass instance with same nested fields.
    language : str, optional
        ISO 639-1 language code ('en' or 'fr'). When omitted, falls back to the
        client's own language field if present, otherwise an empty string.

    Returns
    -------
    dict[str, str]
        Context dict with keys:
        - client_id
        - first_name, last_name, name
        - date_of_birth (display format)
        - date_of_birth_iso (YYYY-MM-DD)
        - date_of_birth_iso_compact (YYYYMMDD)
        - school, board
        - postal_code, city, province, street_address
        - language_code ('en' or 'fr')

    Examples
    --------
    >>> client_dict = {
    ...     "client_id": "12345",
    ...     "person": {"first_name": "John", "last_name": "Doe", "date_of_birth_iso": "2015-03-15"},
    ...     "school": {"name": "Lincoln School"},
    ...     "contact": {"postal_code": "M5V 3A8"}
    ... }
    >>> ctx = build_client_context(client_dict)
    >>> ctx["client_id"]
    '12345'
    >>> ctx["first_name"]
    'John'
    """
    # Handle both dict and ClientRecord: extract nested fields uniformly
    if isinstance(client_data, dict):
        person = client_data.get("person", {})
        contact = client_data.get("contact", {})
        school = client_data.get("school", {})
        board = client_data.get("board", {})
        client_id = client_data.get("client_id", "")
        client_language = client_data.get("language", "")
    else:
        # Assume ClientRecord dataclass
        person = client_data.person or {}
        contact = client_data.contact or {}
        school = client_data.school or {}
        board = client_data.board or {}
        client_id = client_data.client_id
        client_language = client_data.language

    # Get DOB in ISO format
    dob_iso = person.get("date_of_birth_iso") or person.get("date_of_birth", "")
    dob_display = person.get("date_of_birth_display", "") or dob_iso

    # Extract name components (from authoritative first/last fields)
    first_name = person.get("first_name", "")
    last_name = person.get("last_name", "")
    # Combine for display purposes
    full_name = " ".join(filter(None, [first_name, last_name])).strip()

    language_code = string_or_empty(language or client_language)

    # Build context dict for template rendering
    context = {
        "client_id": string_or_empty(client_id),
        "first_name": string_or_empty(first_name),
        "last_name": string_or_empty(last_name),
        "name": string_or_empty(full_name),
        "date_of_birth": string_or_empty(dob_display),
        "date_of_birth_iso": string_or_empty(dob_iso),
        "date_of_birth_iso_compact": string_or_empty(
            dob_iso.replace("-", "") if dob_iso else ""
        ),
        "school": string_or_empty(school.get("name", "")),
        "board": string_or_empty(board.get("name", "")),
        "postal_code": string_or_empty(contact.get("postal_code", "")),
        "city": string_or_empty(contact.get("city", "")),
        "province": string_or_empty(contact.get("province", "")),
        "street_address": string_or_empty(contact.get("street", "")),
        "language_code": language_code,
    }

    return context

deserialize_client_record ¶

deserialize_client_record(client_dict)

Deserialize a dict to a ClientRecord dataclass instance.

Constructs a ClientRecord from a dict (typically from JSON), handling all required and optional fields uniformly. This is the canonical deserialization utility shared across modules for type safety and reduced code duplication.

Parameters:

Name	Type	Description	Default
`client_dict`	`dict`	Client dict with structure: { "sequence": "...", "client_id": "...", "language": "...", "person": {...}, "school": {...}, "board": {...}, "contact": {...}, "vaccines_due": "...", "vaccines_due_list": [...], "received": [...], "metadata": {...}, "qr": {...} (optional) }	required

Returns:

Type	Description
`ClientRecord`	Constructed dataclass instance.

Raises:

Type	Description
`TypeError`	If dict cannot be converted (missing required fields or type mismatch).

Source code in pipeline/utils.py

def deserialize_client_record(client_dict: dict) -> ClientRecord:
    """Deserialize a dict to a ClientRecord dataclass instance.

    Constructs a ClientRecord from a dict (typically from JSON), handling
    all required and optional fields uniformly. This is the canonical
    deserialization utility shared across modules for type safety and
    reduced code duplication.

    Parameters
    ----------
    client_dict : dict
        Client dict with structure:
        {
            "sequence": "...",
            "client_id": "...",
            "language": "...",
            "person": {...},
            "school": {...},
            "board": {...},
            "contact": {...},
            "vaccines_due": "...",
            "vaccines_due_list": [...],
            "received": [...],
            "metadata": {...},
            "qr": {...}  (optional)
        }

    Returns
    -------
    ClientRecord
        Constructed dataclass instance.

    Raises
    ------
    TypeError
        If dict cannot be converted (missing required fields or type mismatch).
    """
    from .data_models import ClientRecord

    try:
        return ClientRecord(
            sequence=client_dict.get("sequence", ""),
            client_id=client_dict.get("client_id", ""),
            language=client_dict.get("language", ""),
            person=client_dict.get("person", {}),
            school=client_dict.get("school", {}),
            board=client_dict.get("board", {}),
            contact=client_dict.get("contact", {}),
            vaccines_due=client_dict.get("vaccines_due"),
            vaccines_due_list=client_dict.get("vaccines_due_list"),
            received=client_dict.get("received"),
            metadata=client_dict.get("metadata", {}),
            qr=client_dict.get("qr"),
        )
    except TypeError as exc:
        raise TypeError(f"Cannot deserialize dict to ClientRecord: {exc}") from exc

Translation Helpers¶

pipeline.translation_helpers ¶

Translation and normalization helpers for disease names.

Provides utilities to normalize input disease names to canonical English forms and translate canonical names to localized display strings for multiple domains (overdue list, immunization history chart).

Contracts:

Canonical disease names are English strings from vaccine_reference.json (e.g., "Diphtheria", "Polio")
Normalization maps raw input strings to canonical names using config/disease_normalization.json
Translation maps canonical names to localized display strings using config/translations/*.json
Missing translations fall back leniently (return canonical name + log warning) unless strict=True
Missing normalization keys return the input unchanged; they may map via disease_map.json later

load_normalization ¶

load_normalization()

Load disease normalization map from config.

Returns:

Type	Description
`Dict[str, str]`	Map from raw disease strings to canonical disease names. Returns empty dict if file does not exist.

Source code in pipeline/translation_helpers.py

def load_normalization() -> Dict[str, str]:
    """Load disease normalization map from config.

    Returns
    -------
    Dict[str, str]
        Map from raw disease strings to canonical disease names.
        Returns empty dict if file does not exist.
    """
    global _NORMALIZATION_CACHE
    if _NORMALIZATION_CACHE is not None:
        return _NORMALIZATION_CACHE

    if not NORMALIZATION_PATH.exists():
        _NORMALIZATION_CACHE = {}
        return _NORMALIZATION_CACHE

    try:
        with open(NORMALIZATION_PATH, encoding="utf-8") as f:
            _NORMALIZATION_CACHE = json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        LOG.warning(f"Failed to load normalization config: {e}")
        _NORMALIZATION_CACHE = {}

    return _NORMALIZATION_CACHE

load_translations ¶

load_translations(domain, lang)

Load translation map for a domain and language from config.

Parameters:

Name	Type	Description	Default
`domain`	`Literal['diseases_overdue', 'diseases_chart']`	Display domain (overdue list or chart).	required
`lang`	`str`	Language code (e.g., "en", "fr").	required

Returns:

Type	Description
`Dict[str, str]`	Map from canonical disease names to localized display strings. Returns empty dict if file does not exist.

Source code in pipeline/translation_helpers.py

def load_translations(
    domain: Literal["diseases_overdue", "diseases_chart"], lang: str
) -> Dict[str, str]:
    """Load translation map for a domain and language from config.

    Parameters
    ----------
    domain : Literal["diseases_overdue", "diseases_chart"]
        Display domain (overdue list or chart).
    lang : str
        Language code (e.g., "en", "fr").

    Returns
    -------
    Dict[str, str]
        Map from canonical disease names to localized display strings.
        Returns empty dict if file does not exist.
    """
    cache_key = (domain, lang)
    if cache_key in _TRANSLATION_CACHES:
        return _TRANSLATION_CACHES[cache_key]

    translation_file = TRANSLATIONS_DIR / f"{lang}_{domain}.json"
    if not translation_file.exists():
        _TRANSLATION_CACHES[cache_key] = {}
        return _TRANSLATION_CACHES[cache_key]

    try:
        with open(translation_file, encoding="utf-8") as f:
            _TRANSLATION_CACHES[cache_key] = json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        LOG.warning(f"Failed to load translations for {lang}_{domain}: {e}")
        _TRANSLATION_CACHES[cache_key] = {}

    return _TRANSLATION_CACHES[cache_key]

normalize_disease ¶

normalize_disease(token)

Normalize a raw disease string to canonical form.

Applies the normalization map from config/disease_normalization.json. If the token is not in the normalization map, returns it unchanged (it may be normalized via disease_map.json later in preprocessing).

Parameters:

Name	Type	Description	Default
`token`	`str`	Raw disease string from input data.	required

Returns:

Type	Description
`str`	Canonical disease name or unchanged token if not found.

Examples:

>>> normalize_disease("Poliomyelitis")
"Polio"
>>> normalize_disease("Unknown Disease")
"Unknown Disease"

Source code in pipeline/translation_helpers.py

def normalize_disease(token: str) -> str:
    """Normalize a raw disease string to canonical form.

    Applies the normalization map from config/disease_normalization.json.
    If the token is not in the normalization map, returns it unchanged (it may
    be normalized via disease_map.json later in preprocessing).

    Parameters
    ----------
    token : str
        Raw disease string from input data.

    Returns
    -------
    str
        Canonical disease name or unchanged token if not found.

    Examples
    --------
    >>> normalize_disease("Poliomyelitis")
    "Polio"
    >>> normalize_disease("Unknown Disease")
    "Unknown Disease"
    """
    token = token.strip()
    normalization = load_normalization()
    return normalization.get(token, token)

display_label ¶

display_label(domain, key, lang, *, strict=False)

Translate a canonical disease name to a localized display label.

Loads translations from config/translations/{domain}.{lang}.json. Falls back leniently to the canonical key if missing (unless strict=True), and logs a single warning per unique missing key.

Parameters:

Name	Type	Description	Default
`domain`	`Literal['diseases_overdue', 'diseases_chart']`	Display domain (overdue list or chart).	required
`key`	`str`	Canonical disease name (English).	required
`lang`	`str`	Language code (e.g., "en", "fr").	required
`strict`	`bool`	If True, raise KeyError on missing translation. If False (default), return the canonical key and log a warning.	`False`

Returns:

Type	Description
`str`	Localized display label or canonical key (if not strict and missing).

Raises:

Type	Description
`KeyError`	If strict=True and translation is missing.

Examples:

>>> display_label("diseases_overdue", "Polio", "en")
"Polio"
>>> display_label("diseases_overdue", "Polio", "fr")
"Poliomyélite"

Source code in pipeline/translation_helpers.py

def display_label(
    domain: Literal["diseases_overdue", "diseases_chart"],
    key: str,
    lang: str,
    *,
    strict: bool = False,
) -> str:
    """Translate a canonical disease name to a localized display label.

    Loads translations from `config/translations/{domain}.{lang}.json`.
    Falls back leniently to the canonical key if missing (unless strict=True),
    and logs a single warning per unique missing key.

    Parameters
    ----------
    domain : Literal["diseases_overdue", "diseases_chart"]
        Display domain (overdue list or chart).
    key : str
        Canonical disease name (English).
    lang : str
        Language code (e.g., "en", "fr").
    strict : bool, optional
        If True, raise KeyError on missing translation. If False (default),
        return the canonical key and log a warning.

    Returns
    -------
    str
        Localized display label or canonical key (if not strict and missing).

    Raises
    ------
    KeyError
        If strict=True and translation is missing.

    Examples
    --------
    >>> display_label("diseases_overdue", "Polio", "en")
    "Polio"
    >>> display_label("diseases_overdue", "Polio", "fr")
    "Poliomyélite"
    """
    translations = load_translations(domain, lang)
    if key in translations:
        return translations[key]

    missing_key = f"{domain}:{lang}:{key}"
    if missing_key not in _LOGGED_MISSING_KEYS:
        _LOGGED_MISSING_KEYS.add(missing_key)
        LOG.warning(
            f"Missing translation for {domain} in language {lang}: {key}. "
            f"Using canonical name."
        )

    if strict:
        raise KeyError(f"Missing translation for {domain} in language {lang}: {key}")

    return key

clear_caches ¶

clear_caches()

Clear all translation and normalization caches.

Useful for testing or reloading configs during runtime.

Source code in pipeline/translation_helpers.py

def clear_caches() -> None:
    """Clear all translation and normalization caches.

    Useful for testing or reloading configs during runtime.
    """
    global _NORMALIZATION_CACHE, _TRANSLATION_CACHES, _LOGGED_MISSING_KEYS
    _NORMALIZATION_CACHE = None
    _TRANSLATION_CACHES.clear()
    _LOGGED_MISSING_KEYS.clear()

PHIX School Validation¶

pipeline.validate_phix ¶

Validate school/daycare names in the input DataFrame against the PHIX mapping file.

The mapping file is produced by a separate script and contains a PHU-keyed dict of normalized school names → PHIX facility IDs.

Match categories

exact: normalized name found in the target PHU's mapping AND facility ID matches

inexact: name found but no ID provided for comparison (name_only), name found but provided ID differs from mapping (id_mismatch), or ID found under a different name (id_only)

no_match: neither name nor ID found for the target PHU

Usage (called from preprocess.run_phix_validation)

df, warnings = validate_schools(
    df=df,
    mapping_path=Path("config/phix_mapping.json"),
    target_phu="Wellington Dufferin Guelph Public Health",
    output_dir=output_dir,
)

PHIXMatchResult `dataclass` ¶

Outcome of matching one input school entry against the PHIX mapping.

Source code in pipeline/validate_phix.py

@dataclass
class PHIXMatchResult:
    """Outcome of matching one input school entry against the PHIX mapping."""

    input_name: str
    input_id: str                    # "" when not provided in input
    match_type: str                  # "exact" | "inexact" | "no_match"
    matched_name: str | None = None  # canonical name from mapping
    matched_id: str | None = None    # facility ID from mapping
    mismatch_reason: str | None = None  # "name_only" | "id_mismatch" | "id_only"

normalize_school_name ¶

normalize_school_name(name)

Uppercase and collapse whitespace.

Must produce the same keys as build_phix_mapping.py so lookups match.

Source code in pipeline/validate_phix.py

def normalize_school_name(name: str) -> str:
    """Uppercase and collapse whitespace.

    Must produce the same keys as build_phix_mapping.py so lookups match.
    """
    if not name:
        return ""
    return re.sub(r"\s+", " ", name.upper().strip())

parse_input_entry ¶

parse_input_entry(entry)

Split a raw input cell into (school_name, facility_id).

Input data sometimes carries the facility ID appended as "SCHOOL NAME - 019186". We split on the rightmost " - " so names that themselves contain dashes are handled correctly.

Returns (entry, "") when no separator is found.

Source code in pipeline/validate_phix.py

def parse_input_entry(entry: str) -> tuple[str, str]:
    """Split a raw input cell into (school_name, facility_id).

    Input data sometimes carries the facility ID appended as
    ``"SCHOOL NAME - 019186"``. We split on the rightmost ``" - "``
    so names that themselves contain dashes are handled correctly.

    Returns ``(entry, "")`` when no separator is found.
    """
    entry = str(entry).strip()
    parts = entry.rsplit(" - ", maxsplit=1)
    if len(parts) == 2:
        return parts[0].strip(), parts[1].strip()
    return entry, ""

load_mapping ¶

load_mapping(mapping_path, target_phu)

Load phix_mapping.json and return lookup dicts for target_phu.

The by_id reverse lookup is built here from the flat name→id mapping.

Parameters:

Name	Type	Description	Default
`mapping_path`	`Path`	Path to the JSON file produced by `build_phix_mapping.py`.	required
`target_phu`	`str`	Exact PHU key string as it appears in the JSON (e.g. `"Wellington Dufferin Guelph Public Health"`).	required

Returns:

Name	Type	Description
`name_to_id`	`dict`	`{normalized_school_name: facility_id}`
`id_to_name`	`dict`	`{facility_id: normalized_school_name}`

Raises:

Type	Description
`FileNotFoundError`	When the mapping file does not exist.
`KeyError`	When target_phu is not present in the mapping, with available PHUs listed in the message.

Source code in pipeline/validate_phix.py

def load_mapping(
    mapping_path: Path,
    target_phu: str,
) -> tuple[dict[str, str], dict[str, str]]:
    """Load ``phix_mapping.json`` and return lookup dicts for *target_phu*.

    The ``by_id`` reverse lookup is built here from the flat name→id mapping.

    Parameters
    ----------
    mapping_path:
        Path to the JSON file produced by ``build_phix_mapping.py``.
    target_phu:
        Exact PHU key string as it appears in the JSON (e.g.
        ``"Wellington Dufferin Guelph Public Health"``).

    Returns
    -------
    name_to_id : dict
        ``{normalized_school_name: facility_id}``
    id_to_name : dict
        ``{facility_id: normalized_school_name}``

    Raises
    ------
    FileNotFoundError
        When the mapping file does not exist.
    KeyError
        When *target_phu* is not present in the mapping, with available PHUs
        listed in the message.
    """
    if not mapping_path.exists():
        raise FileNotFoundError(f"PHIX mapping file not found: {mapping_path}")

    data = json.loads(mapping_path.read_text(encoding="utf-8"))
    phus: dict = data.get("phus", {})

    if target_phu not in phus:
        available = ", ".join(f'"{p}"' for p in sorted(phus))
        raise KeyError(
            f"PHU '{target_phu}' not found in {mapping_path.name}. "
            f"Available PHUs: {available}"
        )

    name_to_id: dict[str, str] = phus[target_phu]
    id_to_name: dict[str, str] = {fid: name for name, fid in name_to_id.items() if fid}
    return name_to_id, id_to_name

classify_match ¶

classify_match(input_name, input_id, name_to_id, id_to_name)

Classify one input entry against the PHU's school/ID mapping.

Parameters:

Name	Type	Description	Default
`input_name`	`str`	Raw school name from the input (before normalisation).	required
`input_id`	`str`	Facility ID extracted from the input cell, or `""` if absent.	required
`name_to_id`	`dict[str, str]`	`{normalized_name: facility_id}` for the target PHU.	required
`id_to_name`	`dict[str, str]`	`{facility_id: normalized_name}` for the target PHU.	required

Source code in pipeline/validate_phix.py

def classify_match(
    input_name: str,
    input_id: str,
    name_to_id: dict[str, str],
    id_to_name: dict[str, str],
) -> PHIXMatchResult:
    """Classify one input entry against the PHU's school/ID mapping.

    Parameters
    ----------
    input_name:
        Raw school name from the input (before normalisation).
    input_id:
        Facility ID extracted from the input cell, or ``""`` if absent.
    name_to_id:
        ``{normalized_name: facility_id}`` for the target PHU.
    id_to_name:
        ``{facility_id: normalized_name}`` for the target PHU.
    """
    normalized = normalize_school_name(input_name)

    if normalized in name_to_id:
        mapping_id = name_to_id[normalized]

        if not input_id:
            return PHIXMatchResult(
                input_name=input_name,
                input_id=input_id,
                match_type="inexact",
                matched_name=input_name,
                matched_id=mapping_id,
                mismatch_reason="name_only",
            )
        if input_id == mapping_id:
            return PHIXMatchResult(
                input_name=input_name,
                input_id=input_id,
                match_type="exact",
                matched_name=input_name,
                matched_id=mapping_id,
            )
        return PHIXMatchResult(
            input_name=input_name,
            input_id=input_id,
            match_type="inexact",
            matched_name=input_name,
            matched_id=mapping_id,
            mismatch_reason="id_mismatch",
        )

    # Name not found — try reverse ID lookup
    if input_id and input_id in id_to_name:
        canonical_name = id_to_name[input_id]
        return PHIXMatchResult(
            input_name=input_name,
            input_id=input_id,
            match_type="inexact",
            matched_name=canonical_name,
            matched_id=input_id,
            mismatch_reason="id_only",
        )

    return PHIXMatchResult(
        input_name=input_name,
        input_id=input_id,
        match_type="no_match",
    )

validate_schools ¶

validate_schools(df, mapping_path, target_phu, output_dir, school_column='SCHOOL_NAME', unmatched_behavior='warn', column_prefix='PHIX_')

Validate school names in df against the PHIX mapping for target_phu.

Adds four columns to the DataFrame (with column_prefix):

{prefix}FACILITY_ID : Matched facility ID, or ""
{prefix}MATCH_TYPE : "exact" | "inexact" | "no_match"
{prefix}MATCHED_NAME : Canonical name from mapping, or ""
{prefix}MATCHED_PHU : target_phu for matched rows, "" otherwise

Writes three CSV files to output_dir:

phix_exact.csv
phix_inexact.csv
phix_no_match.csv

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	Input DataFrame (not mutated; a copy is returned).	required
`mapping_path`	`Path`	Path to `phix_mapping.json`.	required
`target_phu`	`str`	Exact PHU key in the mapping JSON.	required
`output_dir`	`Path`	Directory for CSV output files.	required
`school_column`	`str`	Column in df containing school/daycare names.	`'SCHOOL_NAME'`
`unmatched_behavior`	`str`	`"warn"` — log warning, return all rows. `"error"` — raise `ValueError` if any `no_match` results. `"skip"` — filter out rows whose school has no match.	`'warn'`
`column_prefix`	`str`	Prefix applied to all output column names.	`'PHIX_'`

Returns:

Type	Description
`tuple[DataFrame, list[str]]`	Enriched DataFrame and a list of human-readable warning strings.

Source code in pipeline/validate_phix.py

def validate_schools(
    df: pd.DataFrame,
    mapping_path: Path,
    target_phu: str,
    output_dir: Path,
    school_column: str = "SCHOOL_NAME",
    unmatched_behavior: str = "warn",
    column_prefix: str = "PHIX_",
) -> tuple[pd.DataFrame, list[str]]:
    """Validate school names in *df* against the PHIX mapping for *target_phu*.

    Adds four columns to the DataFrame (with *column_prefix*):

    - ``{prefix}FACILITY_ID`` :   Matched facility ID, or ``""``
    - ``{prefix}MATCH_TYPE`` :    ``"exact"`` | ``"inexact"`` | ``"no_match"``
    - ``{prefix}MATCHED_NAME`` :  Canonical name from mapping, or ``""``
    - ``{prefix}MATCHED_PHU`` :   *target_phu* for matched rows, ``""`` otherwise



    Writes three CSV files to *output_dir*:

    * ``phix_exact.csv``
    * ``phix_inexact.csv``
    * ``phix_no_match.csv``

    Parameters
    ----------
    df:
        Input DataFrame (not mutated; a copy is returned).
    mapping_path:
        Path to ``phix_mapping.json``.
    target_phu:
        Exact PHU key in the mapping JSON.
    output_dir:
        Directory for CSV output files.
    school_column:
        Column in *df* containing school/daycare names.
    unmatched_behavior:

        ``"warn"``  — log warning, return all rows.

        ``"error"`` — raise ``ValueError`` if any ``no_match`` results.

        ``"skip"``  — filter out rows whose school has no match.

    column_prefix:
        Prefix applied to all output column names.

    Returns
    -------
    tuple[DataFrame, list[str]]
        Enriched DataFrame and a list of human-readable warning strings.
    """
    warnings: list[str] = []

    if school_column not in df.columns:
        LOG.warning(
            "Column '%s' not found in DataFrame — skipping PHIX validation.",
            school_column,
        )
        return df, warnings

    name_to_id, id_to_name = load_mapping(mapping_path, target_phu)
    LOG.info("Loaded %d schools for PHU '%s'.", len(name_to_id), target_phu)

    # Classify each unique input value (deduplicated for efficiency)
    results: dict[str, PHIXMatchResult] = {}
    for raw in df[school_column].dropna().unique():
        name, fac_id = parse_input_entry(str(raw))
        results[str(raw)] = classify_match(name, fac_id, name_to_id, id_to_name)

    # Enrich DataFrame
    df = df.copy()

    def _attr(raw_val: object, attr: str, default: str = "") -> str:
        r = results.get(str(raw_val))
        return getattr(r, attr, default) if r else default

    def col(suffix: str) -> str:
        return f"{column_prefix}{suffix}"

    df[col("FACILITY_ID")] = df[school_column].apply(
        lambda x: _attr(x, "matched_id") if pd.notna(x) else ""
    )
    df[col("MATCH_TYPE")] = df[school_column].apply(
        lambda x: _attr(x, "match_type", "no_match") if pd.notna(x) else "no_match"
    )
    df[col("MATCHED_NAME")] = df[school_column].apply(
        lambda x: _attr(x, "matched_name") if pd.notna(x) else ""
    )
    df[col("MATCHED_PHU")] = df[school_column].apply(
        lambda x: target_phu if pd.notna(x) and _attr(x, "match_type", "no_match") != "no_match" else ""
    )

    # Split into categories
    exact    = [r for r in results.values() if r.match_type == "exact"]
    inexact  = [r for r in results.values() if r.match_type == "inexact"]
    no_match = [r for r in results.values() if r.match_type == "no_match"]

    # Write CSVs
    _write_csv(exact,    output_dir / "phix_exact.csv")
    _write_csv(inexact,  output_dir / "phix_inexact.csv")
    _write_csv(no_match, output_dir / "phix_no_match.csv")

    LOG.info(
        "PHIX validation complete: %d exact, %d inexact, %d no_match",
        len(exact), len(inexact), len(no_match),
    )

    if no_match:
        no_match_names = sorted(r.input_name for r in no_match)
        msg = (
            f"{len(no_match)} school(s) had no PHIX match for PHU '{target_phu}'. "
            f"See phix_no_match.csv for details."
        )
        warnings.append(msg)
        LOG.warning("%s First unmatched: %s", msg, no_match_names[:5])

        if unmatched_behavior == "error":
            raise ValueError(msg)
        elif unmatched_behavior == "skip":
            matched_inputs = {
                raw for raw, r in results.items() if r.match_type != "no_match"
            }
            before = len(df)
            df = df[df[school_column].isin(matched_inputs)]
            skipped = before - len(df)
            warnings.append(f"Skipped {skipped} record(s) with no PHIX match.")
            LOG.info("Skipped %d records with no PHIX match, %d remaining.", skipped, len(df))

    return df, warnings

API Reference¶

Orchestrator¶

pipeline.orchestrator ¶

parse_args ¶

validate_args ¶

main ¶

Configuration¶

pipeline.config_loader ¶

load_config ¶

validate_template_fields ¶

validate_config ¶

Data Models¶

pipeline.data_models ¶

ClientRecord dataclass ¶

PreprocessResult dataclass ¶

ArtifactPayload dataclass ¶

PdfRecord dataclass ¶

Enumerations¶

pipeline.enums ¶

BundleStrategy ¶

from_string classmethod ¶

BundleType ¶

Language ¶

from_string classmethod ¶

all_codes classmethod ¶

TemplateField ¶

all_values classmethod ¶

Step 1 — Output Preparation¶

pipeline.prepare_output ¶

is_log_directory ¶

purge_output_directory ¶

default_prompt ¶

prepare_output_directory ¶

Step 2 — Preprocessing¶

pipeline.preprocess ¶

convert_date_string ¶

format_iso_date_for_language ¶

check_addresses_complete ¶

convert_date_iso ¶

over_16_check ¶

configure_logging ¶

detect_file_type ¶

read_input ¶

normalize ¶

split_vaccine_due_entry ¶

hide_vaccine_due_doses ¶

format_vaccine_due_list ¶

map_columns ¶

Example usage (illustrative only):¶

renamed_df, mapping = map_columns(df, required_columns=['state', 'date', 'count'])¶

filter_columns ¶

normalize_dataframe ¶

synthesize_identifier ¶

process_vaccines_due ¶

normalize_validity_status ¶

collapse_validity_statuses ¶

classify_dataset_validity ¶

parse_dose_segments ¶

compute_column_statuses ¶

build_received_rows ¶

build_preprocess_result ¶

run_phix_validation ¶

write_artifact ¶

Step 3 — QR Code Generation¶

pipeline.generate_qr_codes ¶

encode_qr_payload_url ¶

generate_qr_code ¶

read_preprocessed_artifact ¶

load_qr_settings ¶

generate_qr_codes ¶

main ¶

Step 4 — Notice Generation¶

pipeline.generate_notices ¶

load_template_module ¶

build_language_renderers ¶

get_language_renderer ¶

read_artifact ¶

escape_string ¶

to_typ_value ¶

load_and_translate_chart_diseases ¶

ClientRecord `dataclass` ¶

PreprocessResult `dataclass` ¶

ArtifactPayload `dataclass` ¶

PdfRecord `dataclass` ¶

from_string `classmethod` ¶

from_string `classmethod` ¶

all_codes `classmethod` ¶

all_values `classmethod` ¶

ValidationResult `dataclass` ¶

RuleResult `dataclass` ¶

ValidationSummary `dataclass` ¶

BundleConfig `dataclass` ¶

BundlePlan `dataclass` ¶

BundleResult `dataclass` ¶