From 15ab6b77551b8ea19016532d77d627f5ea507e56 Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 00:04:33 -0600 Subject: [PATCH 1/9] adds dimension-constrained Number type --- ucon/pydantic.py | 108 ++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 93 insertions(+), 15 deletions(-) diff --git a/ucon/pydantic.py b/ucon/pydantic.py index f97fdb7..8004152 100644 --- a/ucon/pydantic.py +++ b/ucon/pydantic.py @@ -33,7 +33,7 @@ """ -from typing import Annotated, Any +from typing import Annotated, Any, Generic, TypeVar try: from pydantic import GetCoreSchemaHandler, GetJsonSchemaHandler @@ -45,7 +45,7 @@ "Install with: pip install ucon[pydantic]" ) from e -from ucon.core import Number as _Number +from ucon.core import Dimension, Number as _Number from ucon.units import UnknownUnitError, get_unit_by_name @@ -122,6 +122,17 @@ def _serialize_number(n: _Number) -> dict: } +def _make_dimension_validator(dimension: Dimension): + """Create a validator function for a specific dimension.""" + def validate_dimension(n: _Number) -> _Number: + if n.dimension != dimension: + raise ValueError( + f"expected dimension '{dimension.name}', got '{n.dimension.name}'" + ) + return n + return validate_dimension + + class _NumberPydanticAnnotation: """ Pydantic annotation helper for ucon Number type. @@ -131,9 +142,13 @@ class _NumberPydanticAnnotation: the internal Unit/UnitProduct types. """ - @classmethod + dimension: Dimension | None = None + + def __init__(self, dimension: Dimension | None = None): + self.dimension = dimension + def __get_pydantic_core_schema__( - cls, + self, _source_type: Any, _handler: GetCoreSchemaHandler, ) -> CoreSchema: @@ -143,8 +158,18 @@ def __get_pydantic_core_schema__( Uses no_info_plain_validator_function to bypass Pydantic's default introspection of the Number class fields. """ + if self.dimension is not None: + # Chain dimension validation after number parsing + dim_validator = _make_dimension_validator(self.dimension) + def validate_with_dimension(v: Any) -> _Number: + n = _validate_number(v) + return dim_validator(n) + validator = validate_with_dimension + else: + validator = _validate_number + return core_schema.no_info_plain_validator_function( - _validate_number, + validator, serialization=core_schema.plain_serializer_function_ser_schema( _serialize_number, info_arg=False, @@ -152,14 +177,13 @@ def __get_pydantic_core_schema__( ), ) - @classmethod def __get_pydantic_json_schema__( - cls, + self, _core_schema: CoreSchema, handler: GetJsonSchemaHandler, ) -> JsonSchemaValue: """Generate JSON schema for OpenAPI documentation.""" - return { + schema = { "type": "object", "properties": { "quantity": {"type": "number"}, @@ -168,16 +192,46 @@ def __get_pydantic_json_schema__( }, "required": ["quantity"], } + if self.dimension is not None: + schema["description"] = f"Number with dimension '{self.dimension.name}'" + return schema + + +class _NumberType: + """ + Subscriptable Number type for Pydantic models. + + Supports both unconstrained and dimension-constrained usage: + + value: Number # Any dimension + length: Number[Dimension.length] # Must be length dimension + + When subscripted with a Dimension, validation will fail if the + parsed Number has a different dimension. + """ + def __class_getitem__(cls, dimension: Dimension) -> type: + """Return an Annotated type with dimension validation.""" + if not isinstance(dimension, Dimension): + raise TypeError( + f"Number[...] requires a Dimension, got {type(dimension).__name__}" + ) + return Annotated[_Number, _NumberPydanticAnnotation(dimension)] -Number = Annotated[_Number, _NumberPydanticAnnotation] + def __new__(cls) -> type: + """When used without subscript, return the base Annotated type.""" + return Annotated[_Number, _NumberPydanticAnnotation()] + + +# Export Number as the subscriptable type +Number = _NumberType """ -Pydantic-compatible Number type. +Pydantic-compatible Number type with optional dimension constraints. Use this as a type hint in Pydantic models to enable automatic validation and JSON serialization of ucon Number instances. -Example:: +Basic usage (any dimension):: from pydantic import BaseModel from ucon.pydantic import Number @@ -185,15 +239,39 @@ def __get_pydantic_json_schema__( class Measurement(BaseModel): value: Number - # From dict - m = Measurement(value={"quantity": 5, "unit": "m"}) + m = Measurement(value={"quantity": 5, "unit": "km"}) + print(m.value) # <5 km> + +With dimension constraint:: + + from ucon import Dimension + from ucon.pydantic import Number + + class Vehicle(BaseModel): + mass: Number[Dimension.mass] + speed: Number[Dimension.velocity] + + # Valid + v = Vehicle( + mass={"quantity": 1500, "unit": "kg"}, + speed={"quantity": 100, "unit": "km/h"} + ) + + # Invalid - wrong dimension + Vehicle( + mass={"quantity": 5, "unit": "m"}, # ValueError: expected 'mass', got 'length' + speed={"quantity": 100, "unit": "km/h"} + ) + +From Number instance:: - # From Number instance from ucon import units m2 = Measurement(value=units.meter(10)) - # Serialize to JSON +Serialize to JSON:: + print(m.model_dump_json()) + # {"value": {"quantity": 5.0, "unit": "km", "uncertainty": null}} """ __all__ = ["Number"] From 2086a74f51598fefda7c3ef8188ff98afe44c6b8 Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 00:06:00 -0600 Subject: [PATCH 2/9] adds constrained_number factory for composable Pydantic validators --- ucon/pydantic.py | 34 +++++++++++++++++++++++++++++++++- 1 file changed, 33 insertions(+), 1 deletion(-) diff --git a/ucon/pydantic.py b/ucon/pydantic.py index 8004152..216346c 100644 --- a/ucon/pydantic.py +++ b/ucon/pydantic.py @@ -209,6 +209,11 @@ class _NumberType: When subscripted with a Dimension, validation will fail if the parsed Number has a different dimension. """ + _extra_validators: tuple = () + + def __init_subclass__(cls, **kwargs): + super().__init_subclass__(**kwargs) + # Subclasses can add validators def __class_getitem__(cls, dimension: Dimension) -> type: """Return an Annotated type with dimension validation.""" @@ -216,13 +221,40 @@ def __class_getitem__(cls, dimension: Dimension) -> type: raise TypeError( f"Number[...] requires a Dimension, got {type(dimension).__name__}" ) - return Annotated[_Number, _NumberPydanticAnnotation(dimension)] + # Build the Annotated type with base annotation + extra validators + annotations = [_NumberPydanticAnnotation(dimension)] + list(cls._extra_validators) + return Annotated[_Number, *annotations] def __new__(cls) -> type: """When used without subscript, return the base Annotated type.""" + if cls._extra_validators: + return Annotated[_Number, _NumberPydanticAnnotation(), *cls._extra_validators] return Annotated[_Number, _NumberPydanticAnnotation()] +def constrained_number(*validators): + """ + Factory to create subscriptable Number types with additional validators. + + Usage:: + + from pydantic.functional_validators import AfterValidator + + def must_be_positive(n): + if n.quantity <= 0: + raise ValueError("must be positive") + return n + + PositiveNumber = constrained_number(AfterValidator(must_be_positive)) + + class Model(BaseModel): + value: PositiveNumber[Dimension.time] # positive time value + """ + class ConstrainedNumber(_NumberType): + _extra_validators = validators + return ConstrainedNumber + + # Export Number as the subscriptable type Number = _NumberType """ From cc8c89f067012b08f67a714ba28ab14dcf409e68 Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 00:06:56 -0600 Subject: [PATCH 3/9] exports constrained_number from ucon.pydantic --- ucon/pydantic.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ucon/pydantic.py b/ucon/pydantic.py index 216346c..604a0b9 100644 --- a/ucon/pydantic.py +++ b/ucon/pydantic.py @@ -306,4 +306,4 @@ class Vehicle(BaseModel): # {"value": {"quantity": 5.0, "unit": "km", "uncertainty": null}} """ -__all__ = ["Number"] +__all__ = ["Number", "constrained_number"] From 882efba05f6dbfe731d1d5da815c667519de008f Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 00:15:13 -0600 Subject: [PATCH 4/9] dimension validation handles unitless Numbers --- ucon/pydantic.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/ucon/pydantic.py b/ucon/pydantic.py index 604a0b9..3f46ac3 100644 --- a/ucon/pydantic.py +++ b/ucon/pydantic.py @@ -125,9 +125,10 @@ def _serialize_number(n: _Number) -> dict: def _make_dimension_validator(dimension: Dimension): """Create a validator function for a specific dimension.""" def validate_dimension(n: _Number) -> _Number: - if n.dimension != dimension: + actual_dim = n.unit.dimension if n.unit else Dimension.none + if actual_dim != dimension: raise ValueError( - f"expected dimension '{dimension.name}', got '{n.dimension.name}'" + f"expected dimension '{dimension.name}', got '{actual_dim.name}'" ) return n return validate_dimension From ff0aa6e0021298e9facfa66ac96feb2459c41a31 Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 00:54:04 -0600 Subject: [PATCH 5/9] adds a 'schema getter' to _NumberType for when no subscript type is passed --- ucon/pydantic.py | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/ucon/pydantic.py b/ucon/pydantic.py index 3f46ac3..4ce9c1f 100644 --- a/ucon/pydantic.py +++ b/ucon/pydantic.py @@ -216,6 +216,7 @@ def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs) # Subclasses can add validators + @classmethod def __class_getitem__(cls, dimension: Dimension) -> type: """Return an Annotated type with dimension validation.""" if not isinstance(dimension, Dimension): @@ -226,11 +227,21 @@ def __class_getitem__(cls, dimension: Dimension) -> type: annotations = [_NumberPydanticAnnotation(dimension)] + list(cls._extra_validators) return Annotated[_Number, *annotations] - def __new__(cls) -> type: - """When used without subscript, return the base Annotated type.""" - if cls._extra_validators: - return Annotated[_Number, _NumberPydanticAnnotation(), *cls._extra_validators] - return Annotated[_Number, _NumberPydanticAnnotation()] + @classmethod + def __get_pydantic_core_schema__( + cls, + _source_type: Any, + _handler: GetCoreSchemaHandler, + ) -> CoreSchema: + """ + Generate Pydantic core schema when Number is used without subscript. + + This allows both: + value: Number # Any dimension + length: Number[Dimension.length] # Constrained dimension + """ + annotation = _NumberPydanticAnnotation() + return annotation.__get_pydantic_core_schema__(_source_type, _handler) def constrained_number(*validators): From 11303b35d0172c3e9ca45c0aa0985447840f0aad Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 03:02:43 -0600 Subject: [PATCH 6/9] defines the power operation for Number --- ucon/core.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/ucon/core.py b/ucon/core.py index c769cce..bb88c15 100644 --- a/ucon/core.py +++ b/ucon/core.py @@ -1821,6 +1821,30 @@ def __eq__(self, other: Quantifiable) -> bool: return True + def __pow__(self, power: Union[int, float]) -> 'Number': + """Raise Number to a power. + + Examples + -------- + >>> from ucon import units + >>> v = units.meter(3) / units.second(1) + >>> v ** 2 + <9 m²/s²> + """ + new_quantity = self.quantity ** power + new_unit = self.unit ** power + + # Uncertainty propagation: δ(x^n) = |n| * x^(n-1) * δx = |n| * (x^n / x) * δx + new_uncertainty = None + if self.uncertainty is not None and self.quantity != 0: + new_uncertainty = abs(power) * abs(new_quantity / self.quantity) * self.uncertainty + + return Number( + quantity=new_quantity, + unit=new_unit, + uncertainty=new_uncertainty, + ) + def __repr__(self): if self.uncertainty is not None: if not self.unit.dimension: From 2e4a50c725f321288c96870e70d5561a797b8579 Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 10:00:11 -0600 Subject: [PATCH 7/9] adds some helpful pydantic examples --- examples/pydantic/__init__.py | 5 + examples/pydantic/pharma/README.md | 133 ++++++++++ examples/pydantic/pharma/__init__.py | 5 + examples/pydantic/pharma/config.yaml | 43 +++ examples/pydantic/pharma/main.py | 177 +++++++++++++ examples/pydantic/pharma/models.py | 190 ++++++++++++++ examples/pydantic/sre/README.md | 220 ++++++++++++++++ examples/pydantic/sre/__init__.py | 5 + examples/pydantic/sre/config.yaml | 98 +++++++ examples/pydantic/sre/main.py | 168 ++++++++++++ examples/pydantic/sre/models.py | 375 +++++++++++++++++++++++++++ examples/pydantic/toy/README.md | 79 ++++++ examples/pydantic/toy/__init__.py | 5 + examples/pydantic/toy/config.yaml | 29 +++ examples/pydantic/toy/main.py | 98 +++++++ examples/pydantic/toy/models.py | 46 ++++ examples/pydantic/toy/physics.py | 49 ++++ 17 files changed, 1725 insertions(+) create mode 100644 examples/pydantic/__init__.py create mode 100644 examples/pydantic/pharma/README.md create mode 100644 examples/pydantic/pharma/__init__.py create mode 100644 examples/pydantic/pharma/config.yaml create mode 100644 examples/pydantic/pharma/main.py create mode 100644 examples/pydantic/pharma/models.py create mode 100644 examples/pydantic/sre/README.md create mode 100644 examples/pydantic/sre/__init__.py create mode 100644 examples/pydantic/sre/config.yaml create mode 100644 examples/pydantic/sre/main.py create mode 100644 examples/pydantic/sre/models.py create mode 100644 examples/pydantic/toy/README.md create mode 100644 examples/pydantic/toy/__init__.py create mode 100644 examples/pydantic/toy/config.yaml create mode 100644 examples/pydantic/toy/main.py create mode 100644 examples/pydantic/toy/models.py create mode 100644 examples/pydantic/toy/physics.py diff --git a/examples/pydantic/__init__.py b/examples/pydantic/__init__.py new file mode 100644 index 0000000..c16aa63 --- /dev/null +++ b/examples/pydantic/__init__.py @@ -0,0 +1,5 @@ +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +"""Pydantic integration examples for ucon.""" diff --git a/examples/pydantic/pharma/README.md b/examples/pydantic/pharma/README.md new file mode 100644 index 0000000..345a35d --- /dev/null +++ b/examples/pydantic/pharma/README.md @@ -0,0 +1,133 @@ +# Pharmaceutical Compounding Example + +An advanced example demonstrating sophisticated Pydantic patterns with `Number[Dimension]` for safety-critical pharmaceutical calculations. + +## Overview + +This example shows: + +1. **Composite dimension validation** — `mg/kg/day` dose rates +2. **Custom validators** — Positivity constraints, range checking +3. **Cross-field validation** — Ensuring `min < max` for temperature ranges +4. **Uncertainty propagation** — Drug concentration uncertainty flows through calculations +5. **Computed properties** — Derived values like total infusion volume + +## Files + +- `config.yaml` — Drug, patient, and dosing parameters +- `models.py` — Advanced Pydantic models with custom validation +- `main.py` — Example usage with error demonstrations + +## Usage + +```bash +# Install dependencies +pip install pyyaml + +# Run with default config +python main.py + +# Run with custom config +python main.py /path/to/custom/config.yaml +``` + +## Example Output + +``` +=== Drug Information === + Name: Amoxicillin + Concentration: <250 ± 5 mg/mL> + Stability: <14 day> + Storage temp: <2 degC> to <8 degC> + +=== Calculated Values === + Daily dose: <1750 ± 35 mg> + Single dose: <500 ± 10 mg> # Clamped to max + Volume/dose: <2.0 ± 0.05 mL> +``` + +## Key Patterns + +### Composite Dimension Validation + +```python +@field_validator('target_dose') +@classmethod +def validate_dose_rate_dimension(cls, v: Number) -> Number: + expected = Dimension.mass / Dimension.mass / Dimension.time + if v.dimension != expected: + raise ValueError(f"expected {expected.name}, got {v.dimension.name}") + return v +``` + +### Custom Validators with Annotated + +```python +def must_be_positive(n: Number) -> Number: + if n.quantity <= 0: + raise ValueError(f"must be positive, got {n.quantity}") + return n + +PositiveNumber = Annotated[PydanticNumber, AfterValidator(must_be_positive)] +``` + +### Cross-Field Validation + +```python +class TemperatureRange(BaseModel): + min: PydanticNumber[Dimension.temperature] + max: PydanticNumber[Dimension.temperature] + + @model_validator(mode='after') + def min_less_than_max(self) -> 'TemperatureRange': + min_k = self.min.to(units.kelvin).quantity + max_k = self.max.to(units.kelvin).quantity + if min_k >= max_k: + raise ValueError("min must be less than max") + return self +``` + +### Uncertainty Propagation + +```yaml +# config.yaml +concentration: { quantity: 250, unit: "mg/mL", uncertainty: 5 } +``` + +```python +# Uncertainty flows through calculations automatically +volume = settings.calculate_volume_per_dose() +print(volume) # <2.0 ± 0.05 mL> +``` + +### Computed Methods with Bounds Clamping + +```python +def calculate_single_dose(self) -> Number: + daily_dose = self.calculate_daily_dose() + single_dose = daily_dose / self.dosing.frequency + + # Clamp to configured bounds + clamped = max(min_mg, min(max_mg, dose_mg)) + return mg(clamped, uncertainty=single_dose.uncertainty) +``` + +## Validation Examples + +The example demonstrates catching various errors: + +```python +# Wrong dimension +concentration: { quantity: 250, unit: "mg" } # Missing /mL +# → ValidationError: concentration must have dimension 'density' + +# Range violation +storage_temp: + min: { quantity: 10, unit: "degC" } + max: { quantity: 5, unit: "degC" } +# → ValidationError: min must be less than max + +# Negative value +weight: { quantity: -70, unit: "kg" } +# → ValidationError: must be positive +``` diff --git a/examples/pydantic/pharma/__init__.py b/examples/pydantic/pharma/__init__.py new file mode 100644 index 0000000..3c6fa61 --- /dev/null +++ b/examples/pydantic/pharma/__init__.py @@ -0,0 +1,5 @@ +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +"""Pharmaceutical compounding example demonstrating advanced Pydantic patterns.""" diff --git a/examples/pydantic/pharma/config.yaml b/examples/pydantic/pharma/config.yaml new file mode 100644 index 0000000..de71a29 --- /dev/null +++ b/examples/pydantic/pharma/config.yaml @@ -0,0 +1,43 @@ +drug: + name: "Amoxicillin" + concentration: + quantity: 250 + unit: "mg/mL" + uncertainty: 5 + stability_duration: + quantity: 14 + unit: "day" + storage_temp: + min: + quantity: 2 + unit: "degC" + max: + quantity: 8 + unit: "degC" + +patient: + weight: + quantity: 70 + unit: "kg" + uncertainty: 0.5 + age_category: "adult" + +dosing: + target_dose: + quantity: 25 + unit: "mg/kg/day" + frequency: 3 + max_single_dose: + quantity: 500 + unit: "mg" + min_single_dose: + quantity: 100 + unit: "mg" + +infusion: + rate: + quantity: 100 + unit: "mL/h" + duration: + quantity: 0.5 + unit: "h" diff --git a/examples/pydantic/pharma/main.py b/examples/pydantic/pharma/main.py new file mode 100644 index 0000000..e93642f --- /dev/null +++ b/examples/pydantic/pharma/main.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +""" +Pharmaceutical compounding example. + +Demonstrates: +- Complex Pydantic models with Number[Dimension] validation +- Custom dimension validation (mg/kg/day) +- Cross-field validation (min < max) +- Uncertainty propagation through calculations +- Computed properties and methods + +Usage: + python main.py + python main.py path/to/custom/config.yaml +""" + +import sys +from pathlib import Path + +import yaml + +from ucon import Scale, units + +from models import PharmacySettings + +# Define milliliter for convenience +milliliter = Scale.milli * units.liter + + +def load_settings(config_path: Path) -> PharmacySettings: + """Load and validate configuration from YAML.""" + with open(config_path) as f: + data = yaml.safe_load(f) + return PharmacySettings(**data) + + +def main(config_path: Path) -> None: + print(f"Loading configuration from {config_path}\n") + settings = load_settings(config_path) + + print("=== Drug Information ===") + print(f" Name: {settings.drug.name}") + print(f" Concentration: {settings.drug.concentration}") + print(f" Stability: {settings.drug.stability_duration}") + print(f" Storage temp: {settings.drug.storage_temp.min} to " + f"{settings.drug.storage_temp.max}") + print() + + print("=== Patient Information ===") + print(f" Weight: {settings.patient.weight}") + print(f" Age category: {settings.patient.age_category}") + print() + + print("=== Dosing Parameters ===") + print(f" Target dose: {settings.dosing.target_dose}") + print(f" Frequency: {settings.dosing.frequency} doses/day") + print(f" Dose range: {settings.dosing.min_single_dose} to " + f"{settings.dosing.max_single_dose}") + print() + + # Calculate doses (uncertainty propagates automatically) + daily_dose = settings.calculate_daily_dose() + single_dose = settings.calculate_single_dose() + volume = settings.calculate_volume_per_dose() + + print("=== Calculated Values ===") + print(f" Daily dose: {daily_dose}") + print(f" Single dose: {single_dose}") + print(f" Volume/dose: {volume}") + print(f" {volume.to(milliliter)}") + print() + + # Calculate doses per vial + vial_volume = milliliter(100) + doses_per_vial = settings.calculate_doses_per_vial(vial_volume) + print(f" Doses per {vial_volume} vial: {doses_per_vial}") + print() + + # Infusion calculations (if configured) + if settings.infusion: + print("=== Infusion ===") + print(f" Rate: {settings.infusion.rate}") + print(f" Duration: {settings.infusion.duration}") + print(f" Total volume: {settings.infusion.total_volume}") + print() + + # Demonstrate validation errors + print("=== Validation Examples ===") + demonstrate_validation_errors() + + +def demonstrate_validation_errors() -> None: + """Show how various validation errors are caught.""" + import yaml + from pydantic import ValidationError + + # Wrong dimension for concentration + bad_config_1 = """ +drug: + name: "Test" + concentration: { quantity: 250, unit: "mg" } + stability_duration: { quantity: 14, unit: "day" } + storage_temp: + min: { quantity: 2, unit: "degC" } + max: { quantity: 8, unit: "degC" } +patient: + weight: { quantity: 70, unit: "kg" } + age_category: "adult" +dosing: + target_dose: { quantity: 25, unit: "mg/kg/day" } + frequency: 3 + max_single_dose: { quantity: 500, unit: "mg" } + min_single_dose: { quantity: 100, unit: "mg" } +""" + try: + PharmacySettings(**yaml.safe_load(bad_config_1)) + except ValidationError as e: + print(f" Wrong dimension: {e.errors()[0]['msg'][:60]}...") + + # Temperature range violation + bad_config_2 = """ +drug: + name: "Test" + concentration: { quantity: 250, unit: "mg/mL" } + stability_duration: { quantity: 14, unit: "day" } + storage_temp: + min: { quantity: 10, unit: "degC" } + max: { quantity: 5, unit: "degC" } +patient: + weight: { quantity: 70, unit: "kg" } + age_category: "adult" +dosing: + target_dose: { quantity: 25, unit: "mg/kg/day" } + frequency: 3 + max_single_dose: { quantity: 500, unit: "mg" } + min_single_dose: { quantity: 100, unit: "mg" } +""" + try: + PharmacySettings(**yaml.safe_load(bad_config_2)) + except ValidationError as e: + print(f" Range violation: {e.errors()[0]['msg'][:60]}...") + + # Negative value + bad_config_3 = """ +drug: + name: "Test" + concentration: { quantity: 250, unit: "mg/mL" } + stability_duration: { quantity: 14, unit: "day" } + storage_temp: + min: { quantity: 2, unit: "degC" } + max: { quantity: 8, unit: "degC" } +patient: + weight: { quantity: -70, unit: "kg" } + age_category: "adult" +dosing: + target_dose: { quantity: 25, unit: "mg/kg/day" } + frequency: 3 + max_single_dose: { quantity: 500, unit: "mg" } + min_single_dose: { quantity: 100, unit: "mg" } +""" + try: + PharmacySettings(**yaml.safe_load(bad_config_3)) + except ValidationError as e: + print(f" Negative value: {e.errors()[0]['msg'][:60]}...") + + +if __name__ == "__main__": + if len(sys.argv) > 1: + config_file = Path(sys.argv[1]) + else: + config_file = Path(__file__).parent / "config.yaml" + + main(config_file) diff --git a/examples/pydantic/pharma/models.py b/examples/pydantic/pharma/models.py new file mode 100644 index 0000000..f76be0a --- /dev/null +++ b/examples/pydantic/pharma/models.py @@ -0,0 +1,190 @@ +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +""" +Pharmaceutical compounding configuration models. + +Demonstrates advanced Pydantic patterns with Number[Dimension]: +- Composite dimension validation +- Custom validators (positivity, range checking) +- Cross-field validation (min < max) +- Computed properties +- Uncertainty propagation through calculations +""" + +from typing import Annotated, Literal + +from pydantic import BaseModel, field_validator, model_validator +from pydantic.functional_validators import AfterValidator + +from ucon import Dimension, Number, Scale, units +from ucon.pydantic import Number as PydanticNumber, constrained_number + +# Define milliliter for convenience +milliliter = Scale.milli * units.liter + + +# --------------------------------------------------------------------------- +# Custom validators +# --------------------------------------------------------------------------- + + +def must_be_positive(n: Number) -> Number: + """Validator: quantity must be > 0.""" + if n.quantity <= 0: + raise ValueError(f"must be positive, got {n.quantity}") + return n + + +# Subscriptable positive number type +PositiveNumber = constrained_number(AfterValidator(must_be_positive)) + + +# --------------------------------------------------------------------------- +# Nested models with constrained ranges +# --------------------------------------------------------------------------- + + +class TemperatureRange(BaseModel): + """Temperature bounds with cross-field validation.""" + + min: PydanticNumber[Dimension.temperature] + max: PydanticNumber[Dimension.temperature] + + @model_validator(mode='after') + def min_less_than_max(self) -> 'TemperatureRange': + """Ensure min < max after converting to common unit.""" + min_k = self.min.to(units.kelvin).quantity + max_k = self.max.to(units.kelvin).quantity + if min_k >= max_k: + raise ValueError( + f"min ({self.min}) must be less than max ({self.max})" + ) + return self + + +# --------------------------------------------------------------------------- +# Domain models with dimensional constraints +# --------------------------------------------------------------------------- + + +class DrugConfig(BaseModel): + """Drug formulation properties.""" + + name: str + concentration: PositiveNumber[Dimension.density] + stability_duration: PositiveNumber[Dimension.time] + storage_temp: TemperatureRange + + +class PatientConfig(BaseModel): + """Patient demographics.""" + + weight: PositiveNumber[Dimension.mass] + age_category: Literal["pediatric", "adult", "geriatric"] + + @property + def weight_kg(self) -> float: + """Weight in kg for calculations.""" + return self.weight.to(units.kilogram).quantity + + +class DosingConfig(BaseModel): + """Dosing parameters.""" + + target_dose: PositiveNumber # mg/kg/day - validated below + frequency: int + max_single_dose: PositiveNumber[Dimension.mass] + min_single_dose: PositiveNumber[Dimension.mass] + + @field_validator('target_dose') + @classmethod + def validate_dose_rate_dimension(cls, v: Number) -> Number: + """Ensure target_dose has dose-rate dimension (mass/mass/time).""" + expected = Dimension.mass / Dimension.mass / Dimension.time + actual = v.unit.dimension if v.unit else Dimension.none + if actual != expected: + raise ValueError( + f"target_dose must have dimension '{expected.name}', " + f"got '{actual.name}'" + ) + return v + + @field_validator('frequency') + @classmethod + def validate_frequency_positive(cls, v: int) -> int: + """Ensure frequency is positive.""" + if v <= 0: + raise ValueError(f"frequency must be positive, got {v}") + return v + + @model_validator(mode='after') + def min_less_than_max(self) -> 'DosingConfig': + """Ensure min_single_dose < max_single_dose.""" + mg = Scale.milli * units.gram + min_mg = self.min_single_dose.to(mg).quantity + max_mg = self.max_single_dose.to(mg).quantity + if min_mg >= max_mg: + raise ValueError( + f"min_single_dose ({self.min_single_dose}) must be less than " + f"max_single_dose ({self.max_single_dose})" + ) + return self + + +class InfusionConfig(BaseModel): + """IV infusion parameters.""" + + rate: PositiveNumber[Dimension.volume / Dimension.time] + duration: PositiveNumber[Dimension.time] + + @property + def total_volume(self) -> Number: + """Computed: total infusion volume.""" + return self.rate * self.duration + + +# --------------------------------------------------------------------------- +# Root settings with computed methods +# --------------------------------------------------------------------------- + + +class PharmacySettings(BaseModel): + """Root configuration for pharmaceutical compounding.""" + + drug: DrugConfig + patient: PatientConfig + dosing: DosingConfig + infusion: InfusionConfig | None = None + + def calculate_daily_dose(self) -> Number: + """Calculate total daily dose based on patient weight.""" + return self.dosing.target_dose * self.patient.weight * units.day(1) + + def calculate_single_dose(self) -> Number: + """Calculate single dose with bounds clamping.""" + daily_dose = self.calculate_daily_dose() + single_dose = daily_dose / self.dosing.frequency + + # Clamp to bounds + mg = Scale.milli * units.gram + dose_mg = single_dose.to(mg).quantity + min_mg = self.dosing.min_single_dose.to(mg).quantity + max_mg = self.dosing.max_single_dose.to(mg).quantity + + clamped = max(min_mg, min(max_mg, dose_mg)) + + # Preserve uncertainty if present + return mg(clamped, uncertainty=single_dose.uncertainty) + + def calculate_volume_per_dose(self) -> Number: + """Calculate volume of drug solution per dose.""" + dose = self.calculate_single_dose() + return dose / self.drug.concentration + + def calculate_doses_per_vial(self, vial_volume: Number) -> int: + """Calculate number of doses available from a vial.""" + volume_per_dose = self.calculate_volume_per_dose() + return int(vial_volume.to(milliliter).quantity / + volume_per_dose.to(milliliter).quantity) diff --git a/examples/pydantic/sre/README.md b/examples/pydantic/sre/README.md new file mode 100644 index 0000000..9a8ff7a --- /dev/null +++ b/examples/pydantic/sre/README.md @@ -0,0 +1,220 @@ +# SRE Error Budget & SLO Tracking Example + +A comprehensive example demonstrating dimensional analysis for Site Reliability Engineering: SLO tracking, error budget management, burn rate calculations, and capacity planning. + +## Overview + +This example shows: + +1. **Availability in multiple notations** — percent ↔ nines conversion +2. **Error budget calculations** — total, consumed, remaining, time-to-exhaustion +3. **Burn rate analysis** — detecting unsustainable consumption patterns +4. **Ratio/percentage handling** — error rates, availability, headroom +5. **Throughput dimensions** — requests/second for capacity planning +6. **Uncertainty as confidence intervals** — measurement precision from monitoring +7. **Alert threshold evaluation** — multi-dimensional alerting logic + +## Files + +- `config.yaml` — Service SLOs, current metrics, alert thresholds, capacity params +- `models.py` — Pydantic models with error budget computation methods +- `main.py` — Dashboard-style output with recommendations + +## Usage + +```bash +# Install dependencies +pip install pyyaml + +# Run with default config +python main.py + +# Run with custom config +python main.py /path/to/custom/config.yaml +``` + +## Example Output + +``` +============================================================ +Service: payments-api (tier: critical) +============================================================ + +=== SLO Targets === + Availability: <99.95 %> (<3.3 nines>) + Latency p99: < <200 ms> + Error rate: < <0.1 %> + Throughput: > <5000 Hz> + +=== Current Measurements === + Availability: <99.92 ± 0.01 %> + Latency p99: <180 ± 10 ms> + Throughput: <4800 ± 100 Hz> + +=== SLO Compliance === + Availability ✗ FAIL + Latency p99 ✓ PASS + Error rate ✓ PASS + Throughput ✗ FAIL + +=== Error Budget Analysis === + Window: <30 day> + Total budget: <21.6 min> + Consumed: <17.28 min> + Remaining: <4.32 min> (<20 %>) + Burn rate: 1.6x + Time to exhaust: <2.7 day> + +=== Alert Status: ⚠ WARNING === +``` + +## Key Patterns + +### Availability Nines ↔ Percent + +```python +availability: Percentage # accepts %, nines, fraction + +# In config: +target: { quantity: 99.95, unit: "%" } +# or +target: { quantity: 3.3, unit: "nines" } + +# In code: +avail_pct = config.slos.availability.target.to(units.percent) +avail_nines = config.slos.availability.target.to(units.nines) +``` + +### Error Budget Calculations + +```python +def total_error_budget(self) -> Number: + """Total allowed downtime in the window.""" + budget_fraction = 1 - self.target.to(units.fraction).quantity + return self.window * budget_fraction + +def error_budget_consumed(self) -> Number: + """Actual downtime so far.""" + actual_avail = self.current.availability.to(units.fraction).quantity + downtime_fraction = 1 - actual_avail + return self.current.window_elapsed * downtime_fraction +``` + +### Burn Rate + +```python +def burn_rate(self) -> Number: + """ + Burn rate = actual_consumption / expected_consumption + + 1.0 = on track to use exactly the budget + >1.0 = burning faster than sustainable + <1.0 = under budget + """ + window_fraction = elapsed / total_window + expected = total_budget * window_fraction + actual = consumed + return actual / expected +``` + +### Time to Exhaustion + +```python +def time_to_exhaustion(self) -> Number | None: + """At current burn rate, when does budget hit zero?""" + remaining = self.error_budget_remaining() + burn = self.burn_rate().quantity + + budget_per_day = total_budget / window_days + days_remaining = remaining / (budget_per_day * burn) + return units.day(days_remaining) +``` + +### Capacity Planning with Throughput + +```python +class CapacityConfig(BaseModel): + current_instances: int + requests_per_instance: PositiveNumber[Dimension.frequency] + headroom_target: Percentage + + @property + def total_capacity(self) -> Number: + return self.requests_per_instance * self.current_instances + + def required_instances(self, target_throughput: Number) -> int: + headroom = 1 + self.headroom_target.to(units.fraction).quantity + required = target_throughput.quantity * headroom + return ceil(required / self.requests_per_instance.quantity) +``` + +### Alert Level Evaluation + +```python +def alert_level(self) -> Literal["ok", "warning", "critical"]: + """Multi-dimensional alert logic.""" + burn = self.burn_rate().quantity + remaining_pct = self.error_budget_remaining_percent().quantity + + # Critical if either condition is met + if burn >= self.alerting.burn_rate_critical.quantity: + return "critical" + if remaining_pct <= self.alerting.budget_remaining_critical.quantity: + return "critical" + + # Warning thresholds + if burn >= self.alerting.burn_rate_warning.quantity: + return "warning" + if remaining_pct <= self.alerting.budget_remaining_warning.quantity: + return "warning" + + return "ok" +``` + +### Uncertainty from Monitoring + +```yaml +# Measurements include confidence intervals +availability: + quantity: 99.92 + unit: "%" + uncertainty: 0.01 # ±0.01% measurement precision +``` + +```python +# Uncertainty propagates through calculations +remaining = config.error_budget_remaining() +print(remaining) # <4.32 ± 0.2 min> +``` + +## Dimensional Safety + +The SRE domain involves several dimensions: + +| Metric | Dimension | Units | +|--------|-----------|-------| +| Availability | `ratio` | %, nines, fraction | +| Error rate | `ratio` | %, ppm | +| Latency | `time` | ms, s | +| Throughput | `frequency` | req/s, Hz | +| Window | `time` | day, hour | +| Error budget | `time` | min, hour | +| Burn rate | `none` | dimensionless multiplier | + +Dimensional validation catches errors like: + +```yaml +# ❌ Wrong dimension - latency should be time, not percentage +latency_p99: + target: { quantity: 99, unit: "%" } +# → ValidationError: expected dimension 'time', got 'ratio' +``` + +## SRE Concepts Modeled + +1. **SLOs (Service Level Objectives)** — targets for availability, latency, error rate +2. **SLIs (Service Level Indicators)** — current measurements with uncertainty +3. **Error Budget** — total allowed "badness" in a window +4. **Burn Rate** — rate of error budget consumption vs expected +5. **Multi-Burn-Rate Alerting** — Google SRE pattern for actionable alerts +6. **Capacity Planning** — instances needed to meet throughput SLO with headroom diff --git a/examples/pydantic/sre/__init__.py b/examples/pydantic/sre/__init__.py new file mode 100644 index 0000000..0cc1764 --- /dev/null +++ b/examples/pydantic/sre/__init__.py @@ -0,0 +1,5 @@ +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +"""SRE error budget and SLO tracking example.""" diff --git a/examples/pydantic/sre/config.yaml b/examples/pydantic/sre/config.yaml new file mode 100644 index 0000000..f99d1d3 --- /dev/null +++ b/examples/pydantic/sre/config.yaml @@ -0,0 +1,98 @@ +# Service Level Objectives for a production API +service: + name: "payments-api" + tier: "critical" + +# SLO definitions +slos: + availability: + target: + quantity: 99.95 + unit: "%" + window: + quantity: 30 + unit: "day" + + latency_p50: + target: + quantity: 50 + unit: "ms" + threshold_type: "upper" + + latency_p99: + target: + quantity: 200 + unit: "ms" + threshold_type: "upper" + + error_rate: + target: + quantity: 0.1 + unit: "%" + threshold_type: "upper" + + throughput: + target: + quantity: 5000 + unit: "Hz" + threshold_type: "lower" + +# Current measurements (from monitoring) +current: + availability: + quantity: 99.92 + unit: "%" + uncertainty: 0.01 # measurement confidence + + latency_p50: + quantity: 45 + unit: "ms" + uncertainty: 2 + + latency_p99: + quantity: 180 + unit: "ms" + uncertainty: 10 + + error_rate: + quantity: 0.08 + unit: "%" + uncertainty: 0.005 + + throughput: + quantity: 4800 + unit: "Hz" + uncertainty: 100 + + # Time elapsed in current window + window_elapsed: + quantity: 15 + unit: "day" + +# Alert thresholds +alerting: + # Burn rate multipliers that trigger alerts + burn_rate_critical: + quantity: 14.4 + unit: "1" + burn_rate_warning: + quantity: 6 + unit: "1" + + # Error budget remaining threshold for alerts + budget_remaining_critical: + quantity: 10 + unit: "%" + budget_remaining_warning: + quantity: 25 + unit: "%" + +# Capacity planning +capacity: + current_instances: 10 + requests_per_instance: + quantity: 600 + unit: "Hz" + headroom_target: + quantity: 30 + unit: "%" diff --git a/examples/pydantic/sre/main.py b/examples/pydantic/sre/main.py new file mode 100644 index 0000000..a895b96 --- /dev/null +++ b/examples/pydantic/sre/main.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +""" +SRE error budget and SLO tracking example. + +Demonstrates: +- Loading SLO/SLI configuration from YAML +- Error budget calculations with dimensional safety +- Burn rate and time-to-exhaustion computations +- Availability in nines notation +- Capacity planning with throughput dimensions +- Alert level evaluation + +Usage: + python main.py + python main.py path/to/custom/config.yaml +""" + +import sys +from pathlib import Path + +import yaml + +from ucon import units + +from models import SREConfig + + +def load_config(config_path: Path) -> SREConfig: + """Load and validate SRE configuration from YAML.""" + with open(config_path) as f: + data = yaml.safe_load(f) + return SREConfig(**data) + + +def format_duration(n) -> str: + """Format a duration nicely.""" + if n is None: + return "∞ (not burning)" + + hours = n.to(units.hour).quantity + if hours < 1: + return f"{n.to(units.minute)}" + elif hours < 24: + return f"{n.to(units.hour)}" + else: + return f"{n.to(units.day)}" + + +def main(config_path: Path) -> None: + print(f"Loading SRE configuration from {config_path}\n") + config = load_config(config_path) + + # Service info + print(f"{'=' * 60}") + print(f"Service: {config.service.name} (tier: {config.service.tier})") + print(f"{'=' * 60}\n") + + # SLO targets + print("=== SLO Targets ===") + avail_pct = config.slos.availability.target.to(units.percent) + avail_nines = config.slos.availability.target.to(units.nines) + print(f" Availability: {avail_pct} ({avail_nines})") + print(f" Latency p50: < {config.slos.latency_p50.target}") + print(f" Latency p99: < {config.slos.latency_p99.target}") + print(f" Error rate: < {config.slos.error_rate.target}") + print(f" Throughput: > {config.slos.throughput.target}") + print() + + # Current measurements + print("=== Current Measurements ===") + print(f" Availability: {config.current.availability}") + print(f" Latency p50: {config.current.latency_p50}") + print(f" Latency p99: {config.current.latency_p99}") + print(f" Error rate: {config.current.error_rate}") + print(f" Throughput: {config.current.throughput}") + print(f" Window elapsed: {config.current.window_elapsed}") + print() + + # SLO compliance + print("=== SLO Compliance ===") + checks = [ + ("Availability", config.availability_compliance()), + ("Latency p99", config.latency_p99_compliance()), + ("Error rate", config.error_rate_compliance()), + ("Throughput", config.throughput_compliance()), + ] + for name, passed in checks: + status = "✓ PASS" if passed else "✗ FAIL" + print(f" {name:15} {status}") + print() + + # Error budget analysis + print("=== Error Budget Analysis ===") + total_budget = config.error_budget_total() + consumed = config.error_budget_consumed() + remaining = config.error_budget_remaining() + remaining_pct = config.error_budget_remaining_percent() + burn_rate = config.burn_rate() + tte = config.time_to_exhaustion() + + print(f" Window: {config.slos.availability.window}") + print(f" Total budget: {format_duration(total_budget)}") + print(f" Consumed: {format_duration(consumed)}") + print(f" Remaining: {format_duration(remaining)} ({remaining_pct})") + print(f" Burn rate: {burn_rate.quantity:.2f}x") + print(f" Time to exhaust: {format_duration(tte)}") + print() + + # Alert status + alert_level = config.alert_level() + alert_symbol = {"ok": "✓", "warning": "⚠", "critical": "🔴"}[alert_level] + print(f"=== Alert Status: {alert_symbol} {alert_level.upper()} ===") + print(f" Burn rate thresholds:") + print(f" Warning: {config.alerting.burn_rate_warning.quantity}x") + print(f" Critical: {config.alerting.burn_rate_critical.quantity}x") + print(f" Budget remaining thresholds:") + print(f" Warning: < {config.alerting.budget_remaining_warning}") + print(f" Critical: < {config.alerting.budget_remaining_critical}") + print() + + # Capacity analysis + print("=== Capacity Analysis ===") + total_cap = config.capacity.total_capacity + headroom = config.capacity_headroom() + instances_needed = config.instances_for_slo() + + print(f" Current instances: {config.capacity.current_instances}") + print(f" Per-instance capacity: {config.capacity.requests_per_instance}") + print(f" Total capacity: {total_cap}") + print(f" Current load: {config.current.throughput}") + print(f" Headroom: {headroom}") + print(f" Target headroom: {config.capacity.headroom_target}") + print(f" Instances for SLO: {instances_needed}") + print() + + # Recommendations + print("=== Recommendations ===") + if alert_level == "critical": + print(" 🔴 CRITICAL: Error budget nearly exhausted!") + print(" - Investigate recent incidents") + print(" - Consider freezing deployments") + print(" - Page on-call if not already") + elif alert_level == "warning": + print(" ⚠ WARNING: Error budget burning faster than expected") + print(" - Review recent changes") + print(" - Monitor closely") + else: + print(" ✓ All systems nominal") + + if config.capacity.current_instances < instances_needed: + print(f" ⚠ CAPACITY: Need {instances_needed} instances to meet SLO with headroom") + + if not config.all_slos_met(): + failing = [name for name, passed in checks if not passed] + print(f" ⚠ SLO BREACH: {', '.join(failing)}") + + +if __name__ == "__main__": + if len(sys.argv) > 1: + config_file = Path(sys.argv[1]) + else: + config_file = Path(__file__).parent / "config.yaml" + + main(config_file) diff --git a/examples/pydantic/sre/models.py b/examples/pydantic/sre/models.py new file mode 100644 index 0000000..0baaa7e --- /dev/null +++ b/examples/pydantic/sre/models.py @@ -0,0 +1,375 @@ +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +""" +SRE configuration models for SLO tracking and error budget management. + +Demonstrates: +- Ratio dimension for availability and error rates +- Time-based calculations for error budgets +- Burn rate and exhaustion time computations +- Capacity planning with throughput dimensions +- Uncertainty as confidence intervals on measurements +""" + +from enum import Enum +from typing import Literal + +from pydantic import BaseModel, model_validator +from pydantic.functional_validators import AfterValidator + +from ucon import Dimension, Number, Scale, units +from ucon.pydantic import constrained_number + + +# Define millisecond for convenience +millisecond = Scale.milli * units.second + +# --------------------------------------------------------------------------- +# Custom validators +# --------------------------------------------------------------------------- + + +def must_be_positive(n: Number) -> Number: + """Validator: quantity must be > 0.""" + if n.quantity <= 0: + raise ValueError(f"must be positive, got {n.quantity}") + return n + + +def must_be_percentage(n: Number) -> Number: + """Validator: must be in ratio dimension and reasonable range.""" + dim = n.unit.dimension if n.unit else Dimension.none + if dim != Dimension.ratio: + raise ValueError(f"must be a ratio/percentage, got {dim.name}") + pct = n.to(units.percent).quantity + if pct < 0 or pct > 100: + raise ValueError(f"percentage must be 0-100, got {pct}") + return n + + +# Subscriptable types with additional validators +PositiveNumber = constrained_number(AfterValidator(must_be_positive)) +Percentage = constrained_number( + AfterValidator(must_be_positive), + AfterValidator(must_be_percentage), +) + + +# --------------------------------------------------------------------------- +# SLO definitions +# --------------------------------------------------------------------------- + + +class ThresholdType(str, Enum): + """Whether the SLO is an upper or lower bound.""" + upper = "upper" # metric should be below target (latency, error rate) + lower = "lower" # metric should be above target (availability, throughput) + + +class AvailabilitySLO(BaseModel): + """Availability SLO with target and measurement window.""" + + target: Percentage[Dimension.ratio] + window: PositiveNumber[Dimension.time] + + @property + def target_nines(self) -> float: + """Target availability in nines notation.""" + return self.target.to(units.nines).quantity + + @property + def error_budget_ratio(self) -> Number: + """Allowed error ratio (1 - availability).""" + target_fraction = self.target.to(units.fraction).quantity + return units.fraction(1 - target_fraction) + + def total_error_budget(self) -> Number: + """Total error budget in time units.""" + budget_fraction = 1 - self.target.to(units.fraction).quantity + return self.window * budget_fraction + + +class LatencySLO(BaseModel): + """Latency SLO (p50, p99, etc.).""" + + target: PositiveNumber[Dimension.time] + threshold_type: Literal["upper"] = "upper" + + +class ErrorRateSLO(BaseModel): + """Error rate SLO.""" + + target: Percentage[Dimension.ratio] + threshold_type: Literal["upper"] = "upper" + + +class ThroughputSLO(BaseModel): + """Throughput SLO (requests per second).""" + + target: PositiveNumber[Dimension.frequency] + threshold_type: Literal["lower"] = "lower" + + +class SLOConfig(BaseModel): + """All SLOs for a service.""" + + availability: AvailabilitySLO + latency_p50: LatencySLO + latency_p99: LatencySLO + error_rate: ErrorRateSLO + throughput: ThroughputSLO + + +# --------------------------------------------------------------------------- +# Current measurements +# --------------------------------------------------------------------------- + + +class CurrentMetrics(BaseModel): + """Current measured values from monitoring.""" + + availability: Percentage[Dimension.ratio] + latency_p50: PositiveNumber[Dimension.time] + latency_p99: PositiveNumber[Dimension.time] + error_rate: Percentage[Dimension.ratio] + throughput: PositiveNumber[Dimension.frequency] + window_elapsed: PositiveNumber[Dimension.time] + + +# --------------------------------------------------------------------------- +# Alerting configuration +# --------------------------------------------------------------------------- + + +class AlertingConfig(BaseModel): + """Alert thresholds for error budget burn rate.""" + + burn_rate_critical: PositiveNumber[Dimension.ratio] + burn_rate_warning: PositiveNumber[Dimension.ratio] + budget_remaining_critical: Percentage[Dimension.ratio] + budget_remaining_warning: Percentage[Dimension.ratio] + + @model_validator(mode='after') + def critical_more_severe(self) -> 'AlertingConfig': + """Ensure critical thresholds are more severe than warning.""" + if self.burn_rate_critical.quantity <= self.burn_rate_warning.quantity: + raise ValueError("burn_rate_critical must be > burn_rate_warning") + crit_pct = self.budget_remaining_critical.to(units.percent).quantity + warn_pct = self.budget_remaining_warning.to(units.percent).quantity + if crit_pct >= warn_pct: + raise ValueError("budget_remaining_critical must be < warning") + return self + + +# --------------------------------------------------------------------------- +# Capacity planning +# --------------------------------------------------------------------------- + + +class CapacityConfig(BaseModel): + """Capacity planning parameters.""" + + current_instances: int + requests_per_instance: PositiveNumber[Dimension.frequency] + headroom_target: Percentage[Dimension.ratio] + + @property + def total_capacity(self) -> Number: + """Total capacity across all instances.""" + return self.requests_per_instance * self.current_instances + + def required_instances(self, target_throughput: Number) -> int: + """Calculate instances needed for target throughput with headroom.""" + headroom_mult = 1 + self.headroom_target.to(units.fraction).quantity + required_capacity = target_throughput.quantity * headroom_mult + per_instance = self.requests_per_instance.quantity + return int((required_capacity / per_instance) + 0.999) # ceil + + +# --------------------------------------------------------------------------- +# Service configuration +# --------------------------------------------------------------------------- + + +class ServiceConfig(BaseModel): + """Service metadata.""" + + name: str + tier: Literal["critical", "high", "medium", "low"] + + +# --------------------------------------------------------------------------- +# Root configuration with computed methods +# --------------------------------------------------------------------------- + + +class SREConfig(BaseModel): + """Root SRE configuration with error budget calculations.""" + + service: ServiceConfig + slos: SLOConfig + current: CurrentMetrics + alerting: AlertingConfig + capacity: CapacityConfig + + # ------------------------------------------------------------------------- + # Error budget calculations + # ------------------------------------------------------------------------- + + def error_budget_total(self) -> Number: + """Total error budget for the window (in time units).""" + return self.slos.availability.total_error_budget() + + def error_budget_consumed(self) -> Number: + """Error budget consumed so far (in time units).""" + # Actual downtime = elapsed_time * (1 - actual_availability) + actual_avail = self.current.availability.to(units.fraction).quantity + downtime_fraction = 1 - actual_avail + return self.current.window_elapsed * downtime_fraction + + def error_budget_remaining(self) -> Number: + """Remaining error budget (in time units).""" + total = self.error_budget_total() + consumed = self.error_budget_consumed() + remaining_qty = total.quantity - consumed.quantity + # Propagate uncertainty if present + unc = None + if consumed.uncertainty: + unc = consumed.uncertainty # dominated by measurement uncertainty + return Number( + quantity=max(0, remaining_qty), + unit=total.unit, + uncertainty=unc, + ) + + def error_budget_remaining_percent(self) -> Number: + """Remaining error budget as percentage of total.""" + total = self.error_budget_total().quantity + remaining = self.error_budget_remaining().quantity + if total == 0: + return units.percent(0) + return units.percent(100 * remaining / total) + + def burn_rate(self) -> Number: + """Current burn rate (1.0 = on track, >1.0 = burning too fast).""" + # Expected consumption at this point in window + window_fraction = ( + self.current.window_elapsed.quantity / + self.slos.availability.window.to(self.current.window_elapsed.unit).quantity + ) + expected_consumed = self.error_budget_total().quantity * window_fraction + + # Actual consumption + actual_consumed = self.error_budget_consumed().quantity + + if expected_consumed == 0: + return units.fraction(0) + + rate = actual_consumed / expected_consumed + return Number( + quantity=rate, + unit=units.fraction, + uncertainty=None, # Could compute from availability uncertainty + ) + + def time_to_exhaustion(self) -> Number | None: + """Time until error budget is exhausted at current burn rate.""" + remaining = self.error_budget_remaining() + if remaining.quantity <= 0: + return units.second(0) + + burn = self.burn_rate().quantity + if burn <= 0: + return None # Not burning budget + + # At current rate, how long until remaining is consumed? + # Remaining budget / (budget_per_unit_time * burn_rate) + window = self.slos.availability.window + total_budget = self.error_budget_total() + budget_per_day = total_budget.quantity / window.to(units.day).quantity + + if budget_per_day * burn <= 0: + return None + + days_remaining = remaining.quantity / (budget_per_day * burn) + return units.day(days_remaining) + + # ------------------------------------------------------------------------- + # SLO compliance checks + # ------------------------------------------------------------------------- + + def availability_compliance(self) -> bool: + """Is current availability meeting SLO?""" + current = self.current.availability.to(units.percent).quantity + target = self.slos.availability.target.to(units.percent).quantity + return current >= target + + def latency_p99_compliance(self) -> bool: + """Is current p99 latency meeting SLO?""" + current = self.current.latency_p99.to(millisecond).quantity + target = self.slos.latency_p99.target.to(millisecond).quantity + return current <= target + + def error_rate_compliance(self) -> bool: + """Is current error rate meeting SLO?""" + current = self.current.error_rate.to(units.percent).quantity + target = self.slos.error_rate.target.to(units.percent).quantity + return current <= target + + def throughput_compliance(self) -> bool: + """Is current throughput meeting SLO?""" + current = self.current.throughput.to(units.hertz).quantity + target = self.slos.throughput.target.to(units.hertz).quantity + return current >= target + + def all_slos_met(self) -> bool: + """Are all SLOs currently being met?""" + return ( + self.availability_compliance() and + self.latency_p99_compliance() and + self.error_rate_compliance() and + self.throughput_compliance() + ) + + # ------------------------------------------------------------------------- + # Alert evaluation + # ------------------------------------------------------------------------- + + def alert_level(self) -> Literal["ok", "warning", "critical"]: + """Current alert level based on burn rate and remaining budget.""" + burn = self.burn_rate().quantity + remaining_pct = self.error_budget_remaining_percent().quantity + + crit_burn = self.alerting.burn_rate_critical.quantity + warn_burn = self.alerting.burn_rate_warning.quantity + crit_remaining = self.alerting.budget_remaining_critical.to(units.percent).quantity + warn_remaining = self.alerting.budget_remaining_warning.to(units.percent).quantity + + # Critical if burn rate is critical OR remaining budget is critical + if burn >= crit_burn or remaining_pct <= crit_remaining: + return "critical" + + # Warning if burn rate is warning OR remaining budget is warning + if burn >= warn_burn or remaining_pct <= warn_remaining: + return "warning" + + return "ok" + + # ------------------------------------------------------------------------- + # Capacity analysis + # ------------------------------------------------------------------------- + + def capacity_headroom(self) -> Number: + """Current capacity headroom as percentage.""" + total_cap = self.capacity.total_capacity.quantity + current_load = self.current.throughput.quantity + if total_cap == 0: + return units.percent(0) + headroom = (total_cap - current_load) / total_cap * 100 + return units.percent(headroom) + + def instances_for_slo(self) -> int: + """Instances needed to meet throughput SLO with headroom.""" + return self.capacity.required_instances(self.slos.throughput.target) diff --git a/examples/pydantic/toy/README.md b/examples/pydantic/toy/README.md new file mode 100644 index 0000000..17704ec --- /dev/null +++ b/examples/pydantic/toy/README.md @@ -0,0 +1,79 @@ +# Vehicle Simulation Example + +A toy example demonstrating `Number[Dimension]` type hints with Pydantic for loading dimensionally-validated configuration from YAML. + +## Overview + +This example shows: + +1. **Pydantic models with dimensional constraints** — `Number[Dimension.mass]`, `Number[Dimension.velocity]`, etc. +2. **YAML configuration loading** — Automatic deserialization and validation +3. **Physics calculations** — `@enforce_dimensions` decorator for runtime safety + +## Files + +- `config.yaml` — Vehicle and simulation parameters +- `models.py` — Pydantic models with `Number[Dimension]` fields +- `physics.py` — Dimensionally-validated physics functions +- `main.py` — Example usage + +## Usage + +```bash +# Install dependencies +pip install pyyaml + +# Run with default config +python main.py + +# Run with custom config +python main.py /path/to/custom/config.yaml +``` + +## Example Output + +``` +=== Vehicle Properties === + Mass: <1500 kg> + Max speed: <120 km/h> + Drag coefficient: <0.32> + Frontal area: <2.2 m²> + +=== Calculations === + Kinetic energy at max speed: + <833333.33... J> + <833.33... kJ> + + Stopping distance (8 m/s² braking): + <69.44... m> +``` + +## Key Patterns + +### Dimensional Type Hints + +```python +class VehicleConfig(BaseModel): + mass: Number[Dimension.mass] # kg, lb, etc. + max_speed: Number[Dimension.velocity] # m/s, km/h, mph +``` + +### Dimension-Enforced Functions + +```python +@enforce_dimensions +def kinetic_energy( + mass: Number[Dimension.mass], + velocity: Number[Dimension.velocity], +) -> Number[Dimension.energy]: + return 0.5 * mass * velocity ** 2 +``` + +### Validation at Load Time + +If the YAML has wrong dimensions, Pydantic rejects it: + +```yaml +# ❌ This would fail validation +max_speed: { quantity: 120, unit: "kg" } # kg is mass, not velocity +``` diff --git a/examples/pydantic/toy/__init__.py b/examples/pydantic/toy/__init__.py new file mode 100644 index 0000000..db122f7 --- /dev/null +++ b/examples/pydantic/toy/__init__.py @@ -0,0 +1,5 @@ +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +"""Vehicle simulation example demonstrating Number[Dimension] with Pydantic.""" diff --git a/examples/pydantic/toy/config.yaml b/examples/pydantic/toy/config.yaml new file mode 100644 index 0000000..659bae2 --- /dev/null +++ b/examples/pydantic/toy/config.yaml @@ -0,0 +1,29 @@ +vehicle: + mass: + quantity: 1500 + unit: "kg" + max_speed: + quantity: 33.33 # ~120 km/h + unit: "m/s" + drag_coefficient: + quantity: 0.32 + unit: "1" + frontal_area: + quantity: 2.2 + unit: "m^2" + +simulation: + time_step: + quantity: 0.01 + unit: "s" + duration: + quantity: 60 + unit: "s" + +environment: + air_density: + quantity: 1.225 + unit: "kg/m^3" + gravity: + quantity: 9.81 + unit: "m/s^2" diff --git a/examples/pydantic/toy/main.py b/examples/pydantic/toy/main.py new file mode 100644 index 0000000..2bbe0f4 --- /dev/null +++ b/examples/pydantic/toy/main.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +""" +Vehicle simulation example. + +Demonstrates: +- Loading YAML config into Pydantic models with Number[Dimension] validation +- Dimensional validation at function boundaries via @enforce_dimensions +- Unit conversion and arithmetic + +Usage: + python main.py + python main.py path/to/custom/config.yaml +""" + +import sys +from pathlib import Path + +import yaml + +from ucon import units + +from models import Settings +from physics import drag_force, kinetic_energy, stopping_distance, time_to_stop + + +def load_settings(config_path: Path) -> Settings: + """Load and validate configuration from YAML.""" + with open(config_path) as f: + data = yaml.safe_load(f) + return Settings(**data) + + +def main(config_path: Path) -> None: + print(f"Loading configuration from {config_path}\n") + settings = load_settings(config_path) + + # Extract values (already validated as correct dimensions) + mass = settings.vehicle.mass + max_speed = settings.vehicle.max_speed + drag_coeff = settings.vehicle.drag_coefficient + area = settings.vehicle.frontal_area + air_density = settings.environment.air_density + gravity = settings.environment.gravity + + print("=== Vehicle Properties ===") + print(f" Mass: {mass}") + print(f" Max speed: {max_speed}") + print(f" Drag coefficient: {drag_coeff}") + print(f" Frontal area: {area}") + print() + + print("=== Environment ===") + print(f" Air density: {air_density}") + print(f" Gravity: {gravity}") + print() + + # Calculate kinetic energy at max speed + ke = kinetic_energy(mass, max_speed) + print("=== Calculations ===") + print(f" Kinetic energy at max speed:") + print(f" {ke}") + print() + + # Calculate drag force at max speed + drag = drag_force(air_density, max_speed, drag_coeff, area) + print(f" Drag force at max speed:") + print(f" {drag}") + print() + + # Calculate stopping distance with 8 m/s² braking + braking = (units.meter / units.second ** 2)(8) + distance = stopping_distance(max_speed, braking) + time = time_to_stop(max_speed, braking) + print(f" Emergency braking ({braking}):") + print(f" Stopping distance: {distance.to(units.meter)}") + print(f" Time to stop: {time.to(units.second)}") + print() + + # Demonstrate dimension error catching + print("=== Dimension Safety ===") + try: + # Intentionally swap mass and velocity + kinetic_energy(max_speed, mass) + except (ValueError, TypeError) as e: + print(f" Caught dimension error: {e}") + + +if __name__ == "__main__": + if len(sys.argv) > 1: + config_file = Path(sys.argv[1]) + else: + config_file = Path(__file__).parent / "config.yaml" + + main(config_file) diff --git a/examples/pydantic/toy/models.py b/examples/pydantic/toy/models.py new file mode 100644 index 0000000..c2dc2ac --- /dev/null +++ b/examples/pydantic/toy/models.py @@ -0,0 +1,46 @@ +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +""" +Vehicle simulation configuration models. + +Demonstrates using Number[Dimension] type hints with Pydantic +for dimensionally-validated configuration loading. +""" + +from pydantic import BaseModel + +from ucon import Dimension +from ucon.pydantic import Number + + +class VehicleConfig(BaseModel): + """Vehicle physical properties.""" + + mass: Number[Dimension.mass] + max_speed: Number[Dimension.velocity] + drag_coefficient: Number[Dimension.ratio] # dimensionless coefficient + frontal_area: Number[Dimension.area] + + +class SimulationConfig(BaseModel): + """Simulation parameters.""" + + time_step: Number[Dimension.time] + duration: Number[Dimension.time] + + +class EnvironmentConfig(BaseModel): + """Environmental conditions.""" + + air_density: Number[Dimension.density] + gravity: Number[Dimension.acceleration] + + +class Settings(BaseModel): + """Root configuration model.""" + + vehicle: VehicleConfig + simulation: SimulationConfig + environment: EnvironmentConfig diff --git a/examples/pydantic/toy/physics.py b/examples/pydantic/toy/physics.py new file mode 100644 index 0000000..3071383 --- /dev/null +++ b/examples/pydantic/toy/physics.py @@ -0,0 +1,49 @@ +# © 2026 The Radiativity Company +# Licensed under the Apache License, Version 2.0 +# See the LICENSE file for details. + +""" +Physics calculations with dimensional validation. + +All functions use @enforce_dimensions to catch unit errors at call time. +""" + +from ucon import Dimension, Number, enforce_dimensions + + +@enforce_dimensions +def kinetic_energy( + mass: Number[Dimension.mass], + velocity: Number[Dimension.velocity], +) -> Number[Dimension.energy]: + """Calculate kinetic energy: KE = ½mv²""" + return mass * velocity ** 2 / 2 + + +@enforce_dimensions +def drag_force( + density: Number[Dimension.density], + velocity: Number[Dimension.velocity], + drag_coeff: Number[Dimension.ratio], + area: Number[Dimension.area], +) -> Number[Dimension.force]: + """Calculate aerodynamic drag: F_drag = ½ρv²CdA""" + return density * velocity ** 2 * drag_coeff * area / 2 + + +@enforce_dimensions +def stopping_distance( + velocity: Number[Dimension.velocity], + deceleration: Number[Dimension.acceleration], +) -> Number[Dimension.length]: + """Calculate stopping distance: d = v² / 2a""" + return velocity ** 2 / deceleration / 2 + + +@enforce_dimensions +def time_to_stop( + velocity: Number[Dimension.velocity], + deceleration: Number[Dimension.acceleration], +) -> Number[Dimension.time]: + """Calculate time to stop: t = v / a""" + return velocity / deceleration From 6fd7a22a5e5de23831dcb6967cf6f15e85941005 Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 15:08:30 -0600 Subject: [PATCH 8/9] avoids unpacking for Annotations --- ucon/pydantic.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/ucon/pydantic.py b/ucon/pydantic.py index 4ce9c1f..03d3ac6 100644 --- a/ucon/pydantic.py +++ b/ucon/pydantic.py @@ -224,8 +224,10 @@ def __class_getitem__(cls, dimension: Dimension) -> type: f"Number[...] requires a Dimension, got {type(dimension).__name__}" ) # Build the Annotated type with base annotation + extra validators - annotations = [_NumberPydanticAnnotation(dimension)] + list(cls._extra_validators) - return Annotated[_Number, *annotations] + # Use __class_getitem__ directly for Python 3.10 compatibility + # (unpacking in Annotated[...] requires Python 3.11+) + annotations = tuple([_NumberPydanticAnnotation(dimension)] + list(cls._extra_validators)) + return Annotated.__class_getitem__((_Number,) + annotations) @classmethod def __get_pydantic_core_schema__( From 43047b249161fff76c664b08ad8a3b0c7029a6f5 Mon Sep 17 00:00:00 2001 From: "Emmanuel I. Obi" Date: Fri, 13 Feb 2026 15:38:01 -0600 Subject: [PATCH 9/9] optional type hint instead of pipe --- ucon/pydantic.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/ucon/pydantic.py b/ucon/pydantic.py index 03d3ac6..21ec9ec 100644 --- a/ucon/pydantic.py +++ b/ucon/pydantic.py @@ -33,7 +33,7 @@ """ -from typing import Annotated, Any, Generic, TypeVar +from typing import Annotated, Any, Generic, Optional, TypeVar try: from pydantic import GetCoreSchemaHandler, GetJsonSchemaHandler @@ -143,9 +143,9 @@ class _NumberPydanticAnnotation: the internal Unit/UnitProduct types. """ - dimension: Dimension | None = None + dimension: Optional[Dimension] = None - def __init__(self, dimension: Dimension | None = None): + def __init__(self, dimension: Optional[Dimension] = None): self.dimension = dimension def __get_pydantic_core_schema__(