westgard-python 0.2.3

Typed Westgard QC rule evaluation, sigma mappings, and planning helpers for Python.

Westgard-Python

westgard-python is a pure-Python package for evaluating Westgard rules, named multirule QC procedures, Westgard Sigma Rules mappings, and reference-backed planning guidance.

The package is built directly from the canonical reference in the master reference. It treats QC semantics as explicit API surface rather than implicit chart logic.

Status

The project is in active early development and targets 0.2.3.

Installation

pip install westgard-python

Optional dataframe helpers:

pip install "westgard-python[pandas]"

API Layers

Top-level API (import westgard):

Low-level rule primitives:

• get_rule_definition(rule)
• evaluate_rule(rule, observations, *, scope, stream_name)

Mid-level current-run preset evaluation:

• get_preset_definition(preset)
• evaluate_multirule(*, context, preset, interpretation_mode=InterpretationMode.MODERN)
• evaluate_custom_rules(*, context, rule_set)

High-level historical builders and scanners:

• observation_from_value(*, analyte, material, run_id, sequence, value, mean, standard_deviation, metadata=None)
• series_from_rows(rows)
• evaluate_multirule_sequence(*, series, preset, interpretation_mode=InterpretationMode.MODERN, enabled_scopes=None)
• scan_multirule_series(*, series, preset, interpretation_mode=InterpretationMode.MODERN, enabled_scopes=None, emit=HistoricalEmitMode.ALL_TRIGGERS)
• evaluate_custom_rule_sequence(*, series, rule_set)
• scan_custom_rule_series(*, series, rule_set, emit=HistoricalEmitMode.ALL_TRIGGERS)
• custom_rule_set_from_rules(*, rules, enabled_scopes=(within_run, within_material), interpretation_mode=modern, name=None)
• validate_custom_rule_set(rule_set)

Stable JSON serializers:

• serialize_multirule_result(result, *, rule_name_mode=RuleNameOutputMode.CANONICAL)
• serialize_event_multirule_result(result, *, rule_name_mode=RuleNameOutputMode.CANONICAL)
• serialize_series_multirule_result(result, *, rule_name_mode=RuleNameOutputMode.CANONICAL)
• serialize_series_multirule_report(result, *, rule_name_mode=RuleNameOutputMode.CANONICAL)
• format_rule_name(rule, mode=RuleNameOutputMode.CANONICAL)

Sigma and planning helpers:

• calculate_sigma_metrics(*, total_allowable_error, bias, coefficient_of_variation, analyte=None, level=None)
• recommend_sigma_rules(*, sigma, control_levels)
• plan_qc_strategy(*, sigma_metrics, control_levels, mode=PlanningMode.MONITOR, residual_risk_target=1.0)

Optional adapter API (from westgard.adapters import ...):

• dataframe_to_context(frame, *, current_run_id, across_material=False)
• dataframe_to_series(frame)

Choosing The Right API

• Use evaluate_rule() when you already know the exact stream and exact rule to test.
• Use evaluate_multirule() when you need one decision for one current QC event and you already have explicit current-run and history streams.
• Use custom rule sets when you need validated, user-selected rule bundles rather than named preset families.
• Use series_from_rows() plus evaluate_multirule_sequence() when you need one result per event across a chronological history.
• Use series_from_rows() plus scan_multirule_series() when you need retrospective trigger discovery for reports, chart annotations, or persistence.

Raw Values And Multi-Material Runs

• Raw QC values are supported through observation_from_value() or row inputs containing value, mean, and standard_deviation. The package computes z_score = (value - mean) / standard_deviation internally.
• standard_deviation must be positive. Invalid or nonpositive SDs raise ValueError.
• Multi-material runs are represented by sharing the same run_id across one observation per material. Sequences should remain chronological across the series.
• High-level row and dataframe builders auto-group mixed-analyte tables into one QCSeries per analyte.

Interpretation And Output Notes

• modern evaluates configured rejection rules directly.
• classic preserves the historical 1_2s warning gate. In sequence/retrospective APIs, downstream rules appear explicitly as skipped_by_warning_gate when the gate does not trigger.
• For custom rule sets in classic mode, the warning gate is applied only if 1_2s is explicitly selected in current_run_rules.
• Rule outcomes distinguish violated, not_violated, insufficient_data, and skipped_by_warning_gate.
• Rule metadata includes severity classification: warning, rejection, or preset_dependent.
• Historical APIs default to enabled_scopes={within_run, within_material}. across_material is explicit opt-in.
• Retrospective scans support emit modes: all_triggers, first_trigger_per_rule, first_trigger_per_rule_per_scope, and all_triggers_grouped_by_window.
• Serializer functions provide the stable JSON contract. Rule-bearing payloads always include canonical machine identifiers in rule and display labels in rule_label.
• rule_name_mode controls labels only (canonical, clinical, or compact) and does not affect keys, IDs, grouping, or comparisons.
• Result identity field preset now distinguishes named presets and custom evaluation (custom).
• Serializers attach a nested custom_rule_set object only when preset == "custom":
  • serialize_multirule_result
  • serialize_event_multirule_result
  • serialize_series_multirule_result
  • serialize_series_multirule_report

Historical Scope Semantics

• within_run: evaluates current-event rules such as same-run R_4s.
• within_material: evaluates history rules per material stream across events.
• across_material: evaluates history rules over an explicit combined multi-material stream and is disabled by default in historical APIs.
• Same-run multi-control behavior comes from within_run; cross-history cross-material behavior comes from across_material.

End-To-End Historical Example

from westgard import (
    InterpretationMode,
    MultiruleVariant,
    scan_multirule_series,
    series_from_rows,
    serialize_series_multirule_result,
)

rows = [
    {
        "analyte": "glucose",
        "material": "L1",
        "run_id": "run-001",
        "sequence": 1,
        "value": 102.0,
        "mean": 100.0,
        "standard_deviation": 2.0,
    },
    {
        "analyte": "glucose",
        "material": "L2",
        "run_id": "run-001",
        "sequence": 1,
        "z_score": -0.3,
    },
    {
        "analyte": "glucose",
        "material": "L1",
        "run_id": "run-002",
        "sequence": 2,
        "z_score": 2.2,
    },
    {
        "analyte": "glucose",
        "material": "L2",
        "run_id": "run-002",
        "sequence": 2,
        "z_score": 0.2,
    },
]

series = series_from_rows(rows)["glucose"]
retrospective = scan_multirule_series(
    series=series,
    preset=MultiruleVariant.MODERN_2_CONTROL,
    interpretation_mode=InterpretationMode.MODERN,
)
payload = serialize_series_multirule_result(retrospective)

print(payload["accepted"])
print(payload["violations"])
print(payload["first_violations_by_rule"])

Common Setup

All examples below use these shared imports/helpers.

from westgard import (
    ControlObservation,
    EvaluationScope,
    InterpretationMode,
    MultiruleContext,
    MultiruleVariant,
    PlanningMode,
    RuleName,
    RunIdentifier,
    calculate_sigma_metrics,
    evaluate_multirule,
    evaluate_rule,
    get_preset_definition,
    get_rule_definition,
    plan_qc_strategy,
    recommend_sigma_rules,
)


def obs(run_id: str, material: str, sequence: int, z: float) -> ControlObservation:
    return ControlObservation(
        analyte="glucose",
        material=material,
        run_id=RunIdentifier(run_id),
        sequence=sequence,
        z_score=z,
    )

How-To: get_rule_definition

Signature:

get_rule_definition(rule: RuleName) -> RuleDefinition

Use when you need static metadata for one primitive rule: taxonomy, minimum observations, default scope, intended error type, and description.

definition = get_rule_definition(RuleName.R_4S)
print(definition.minimum_observations)  # 2
print(definition.default_scope.value)   # within_run
print(definition.intended_error)        # random error

Raises:

• KeyError if the rule is not present in RULE_DEFINITIONS.

How-To: evaluate_rule

Signature:

evaluate_rule(
    rule: RuleName,
    observations: tuple[ControlObservation, ...],
    *,
    scope: EvaluationScope,
    stream_name: str,
) -> EvaluationResult

Use when you want to evaluate exactly one rule against one explicit stream.

result = evaluate_rule(
    RuleName.TWO_2S,
    (
        obs("run-001", "L1", 1, 2.2),
        obs("run-002", "L1", 2, 2.3),
    ),
    scope=EvaluationScope.WITHIN_MATERIAL,
    stream_name="material:L1",
)
print(result.violated)      # True
print(result.enough_data)   # True
print(result.reason)

Behavior notes:

• Observations are sorted by sequence before evaluation.
• If there are too few observations, it returns violated=False, enough_data=False.
• Limit rules use strict inequality. Boundary values at exactly +2, -2, +3, or -3 do not violate.
• R_4s must be evaluated with scope=within_run.

Raises:

• ValueError for invalid rule/scope combinations (for example R_4s outside within_run).

How-To: get_preset_definition

Signature:

get_preset_definition(preset: MultiruleVariant) -> PresetDefinition

Use when you need the exact rule bundle for a named preset before evaluating.

preset = get_preset_definition(MultiruleVariant.MODERN_2_CONTROL)
print(preset.current_run_rules)  # (1_3s, R_4s)
print(preset.history_rules)      # (2_2s, 4_1s, 8_x)
print(preset.classic_warning_gate)  # False

Raises:

• KeyError if the preset key is not defined.

How-To: evaluate_multirule

Signature:

evaluate_multirule(
    *,
    context: MultiruleContext,
    preset: MultiruleVariant,
    interpretation_mode: InterpretationMode = InterpretationMode.MODERN,
) -> MultiruleResult

Use when you want end-to-end multirule evaluation with preset rule families.

l1 = (
    obs("run-001", "L1", 1, 0.2),
    obs("run-002", "L1", 3, 0.4),
    obs("run-003", "L1", 5, 3.2),
)
l2 = (
    obs("run-001", "L2", 2, -0.1),
    obs("run-002", "L2", 4, 0.2),
    obs("run-003", "L2", 6, 0.1),
)

context = MultiruleContext(
    current_run=(l1[-1], l2[-1]),
    material_streams={"L1": l1, "L2": l2},
)

result = evaluate_multirule(
    context=context,
    preset=MultiruleVariant.MODERN_2_CONTROL,
    interpretation_mode=InterpretationMode.MODERN,
)
print(result.accepted)
print([v.rule.value for v in result.violations])

Behavior notes:

• modern evaluates configured rejection rules directly.
• classic first evaluates a current-run 1_2s warning gate.
• If classic and no warning gate trigger, the run is accepted early.
• Cross-material history is evaluated only if context.across_material_stream is supplied.

How-To: calculate_sigma_metrics

Signature:

calculate_sigma_metrics(
    *,
    total_allowable_error: float,
    bias: float,
    coefficient_of_variation: float,
    analyte: str | None = None,
    level: str | None = None,
) -> SigmaMetrics

Use when you have TEa/bias/CV and need sigma for rule-selection workflows.

metrics = calculate_sigma_metrics(
    total_allowable_error=10.0,
    bias=-2.0,
    coefficient_of_variation=2.0,
    analyte="glucose",
    level="L1",
)
print(metrics.sigma)  # 4.0

Formula:

• sigma = (TEa - abs(bias)) / CV

Raises:

• ValueError if total_allowable_error <= 0
• ValueError if coefficient_of_variation <= 0

How-To: recommend_sigma_rules

Signature:

recommend_sigma_rules(*, sigma: float, control_levels: int) -> SigmaRuleMapping

Use when you need a reference-backed sigma-band mapping to a rule set.

mapping = recommend_sigma_rules(sigma=4.5, control_levels=2)
print(mapping.sigma_band)          # 4 to <5
print(mapping.recommended_rules)   # 1_3s, 2_2s, R_4s, 4_1s
print(mapping.control_measurements_per_event)
print(mapping.max_runs_between_qc_events)

Raises:

• ValueError if control_levels is not 2 or 3.

How-To: plan_qc_strategy

Signature:

plan_qc_strategy(
    *,
    sigma_metrics: SigmaMetrics,
    control_levels: int,
    mode: PlanningMode = PlanningMode.MONITOR,
    residual_risk_target: float = 1.0,
) -> PlanningRecommendation

Use when you need deterministic planning notes around startup vs routine monitor usage.

metrics = calculate_sigma_metrics(
    total_allowable_error=8.0,
    bias=1.0,
    coefficient_of_variation=1.0,
)

recommendation = plan_qc_strategy(
    sigma_metrics=metrics,
    control_levels=2,
    mode=PlanningMode.STARTUP,
    residual_risk_target=1.0,
)
print(recommendation.mode.value)                # startup
print(recommendation.sigma_mapping.sigma_band)  # based on computed sigma
print(recommendation.qc_event_note)
print(recommendation.out_of_control_recovery)

Raises:

• ValueError if residual_risk_target <= 0.
• ValueError if control_levels is invalid (propagated from recommend_sigma_rules).

How-To: dataframe_to_context (Optional pandas Adapter)

Signature:

dataframe_to_context(
    frame,
    *,
    current_run_id: str,
    across_material: bool = False,
) -> MultiruleContext

Use when your QC data starts as a pandas DataFrame and you want to feed it into evaluate_multirule.

import pandas as pd
from westgard.adapters import dataframe_to_context

frame = pd.DataFrame(
    [
        {"analyte": "glu", "material": "L1", "run_id": "run-1", "sequence": 1, "z_score": 0.1},
        {"analyte": "glu", "material": "L2", "run_id": "run-1", "sequence": 2, "z_score": -0.1},
        {"analyte": "glu", "material": "L1", "run_id": "run-2", "sequence": 3, "z_score": 2.3},
        {"analyte": "glu", "material": "L2", "run_id": "run-2", "sequence": 4, "z_score": 0.2},
    ]
)

context = dataframe_to_context(frame, current_run_id="run-2", across_material=True)
print(len(context.current_run))  # 2

Required columns:

• analyte
• material
• run_id
• sequence
• z_score

Raises:

• ImportError if pandas is not installed.
• TypeError if frame is not a pandas DataFrame.
• ValueError if required columns are missing.
• ValueError if no rows match current_run_id.

How-To: EvaluationResult.to_violation (Method)

Signature:

EvaluationResult.to_violation() -> RuleViolation | None

Use when you want a normalized violation payload for downstream reporting.

result = evaluate_rule(
    RuleName.ONE_3S,
    (obs("run-009", "L1", 1, 3.5),),
    scope=EvaluationScope.WITHIN_RUN,
    stream_name="current_run:run-009",
)
violation = result.to_violation()
print(violation.rule.value if violation else None)

Behavior notes:

• Returns None when result.violated is False.
• Returns RuleViolation with rule/scope/stream and triggering observation IDs when violated.

Design Principles

• modern interpretation is the default for computerized evaluation.
• classic interpretation is supported explicitly and preserves 1_2s warning-gate behavior.
• R_4s is enforced as within-run only.
• Cross-run and cross-material logic are explicit inputs to the engine. The package does not infer run boundaries from timestamps.

Documentation

• Canonical reference: master reference
• Release runbook: publishing guide
• Package docs: MkDocs Material site under docs/.

Local Release Validation

Run the strict local deployability gate with:

python scripts/validate_release.py

This command runs linting, typing, tests with branch coverage gate (>=95%), build, twine checks, and wheel smoke install. It writes a local summary report to reports/test-results-summary.md.

License

Apache-2.0