"""Formula-runtime operation helpers for registry calculations.
The formula evaluator delegates arithmetic dispatch, dated parameter lookup,
rounding, and input validation here while executing
:class:`~domain.calculations.registry.ModeloRevision` formula graphs.
Helpers raise :class:`~domain.calculations.registry.RegistryValidationError`
so :func:`domain.calculations.registry._formula_runtime.calculate_registry_snapshot`
reports contract failures through the registry error channel.
See Also:
:mod:`domain.calculations.registry._formula_runtime`
Snapshot evaluator that calls these helpers while materialising
:class:`~domain.calculations.registry.RegistrySnapshot` outputs.
:mod:`domain.calculations.registry._runtime_graph`
Formula graph walkers that discover the casilla, binding, relation, and
parameter refs consumed before operation dispatch starts.
:class:`domain.calculations.registry.ValidatedRegistryAuthority`
Registry authority loaded by :func:`read_parameter` for ad hoc parameter
reads outside snapshot execution.
"""
from __future__ import annotations
from collections.abc import Mapping
from datetime import date
from decimal import ROUND_HALF_UP, Decimal
from enum import StrEnum
from functools import cache
from pathlib import Path
from typing import TYPE_CHECKING
from ....core.money import round_to_cents as _round_to_cents
from ._casilla_membership import undeclared_casilla_ids
from ._errors import RegistrySnapshotError, RegistryValidationError
from ._ids import CasillaId, validated_casilla_id
from ._schema import DatedValue, ModeloRevision, ParameterDefinition
if TYPE_CHECKING:
from _typeshed import SupportsAllComparisons
from ._authority import ValidatedRegistryAuthority
from ._formula_runtime import _EvalContext
_ZERO = Decimal("0")
_ONE = Decimal("1")
_COMPARISON_OPS = frozenset({"less_than", "less_equal", "greater_than", "greater_equal", "equal"})
_UNARY_PASSTHROUGH_OPS = frozenset({"copy", "lookup_parameter", "previous_period_value", "cross_model_sum"})
[docs]
class RegistryUnresolvedOutcomeReason(StrEnum):
"""Closed reason catalogue for typed formula outcomes with no Decimal value."""
M210_BASELINE_TIPO_DEFERRED = "m210-baseline-tipo-deferred"
M210_CONVENIO_RATE_MISSING = "m210-convenio-rate-missing"
[docs]
def numeric_casilla_value(casilla_id: CasillaId, ctx: _EvalContext) -> Decimal:
"""Read a resolved numeric casilla value from the evaluation context.
Generic accessor shared by the M210/IRNR, M131 módulos, and M303 módulos
IVA formula-op families; each raises :class:`UnresolvedFormulaDependencyError`
the same way for a casilla deferred by a non-blocking source gap.
"""
if casilla_id not in ctx.values:
if casilla_id in ctx.unresolved_casilla_ids:
raise UnresolvedFormulaDependencyError((casilla_id,))
raise RegistryValidationError(
f"casilla {casilla_id!r} referenced before evaluation",
translated_message="errors.calc.casilla_referenced_before_evaluation",
context={"casilla_id": casilla_id},
)
value = ctx.values[casilla_id]
ctx.operand_refs.append(casilla_id)
ctx.operand_casilla_refs.append(casilla_id)
ctx.operand_values.append(value)
return value
[docs]
def evaluate_args_op(op: str, args: list[Decimal]) -> Decimal:
"""Evaluate a resolved formula operation over decimal operands.
Operation names mirror
:class:`~domain.calculations.registry.FormulaExpression` ``op`` values
consumed by
:func:`domain.calculations.registry._formula_runtime.calculate_registry_snapshot`.
"""
if op in {"add", "sum", "previous_period_sum"}:
if op == "previous_period_sum":
_require_non_empty(op, args)
return sum(args, _ZERO)
if op in _COMPARISON_OPS:
_require_arg_count(op, args, 2)
return _ONE if _compare(op, args[0], args[1]) else _ZERO
if op in _UNARY_PASSTHROUGH_OPS:
_require_arg_count(op, args, 1)
return args[0]
return _dispatch_named_arithmetic_op(op, args)
def _dispatch_named_arithmetic_op(op: str, args: list[Decimal]) -> Decimal:
"""Dispatch non-comparison arithmetic operations for :func:`evaluate_args_op`."""
match op:
case "subtract":
_require_arg_count(op, args, 2)
return args[0] - args[1]
case "multiply":
result = _ONE
for arg in args:
result *= arg
return result
case "divide":
_require_arg_count(op, args, 2)
if args[1] == _ZERO:
raise RegistryValidationError(
"formula expression divides by zero",
translated_message="errors.calc.divide_by_zero",
)
return args[0] / args[1]
case "percent":
_require_arg_count(op, args, 2)
return args[0] * args[1] / Decimal("100")
case "min":
_require_non_empty(op, args)
return min(args)
case "max":
_require_non_empty(op, args)
return max(args)
case "clamp":
_require_arg_count(op, args, 3)
return max(args[1], min(args[0], args[2]))
case "negate":
_require_arg_count(op, args, 1)
return -args[0]
case _:
raise RegistryValidationError(f"formula expression uses unsupported op {op!r}")
def _compare(op: str, left: Decimal, right: Decimal) -> bool:
"""Evaluate a comparison operation from a registry formula expression."""
if op == "less_than":
return left < right
if op == "less_equal":
return left <= right
if op == "greater_than":
return left > right
if op == "greater_equal":
return left >= right
if op == "equal":
return left == right
raise RegistryValidationError(f"formula expression uses unsupported comparison op {op!r}")
[docs]
def resolve_bracket(
parameter: ParameterDefinition,
base: Decimal,
date_context: Mapping[str, date],
) -> Decimal:
"""Resolve a bracket-table :class:`ParameterDefinition` for a base amount.
The parameter's bracket date axis must be present in ``date_context`` so
registry-authored validity windows select exactly one bracket row.
"""
if parameter.data_type != "bracket_table":
raise RegistryValidationError(
f"parameter {parameter.id!r} must declare data_type='bracket_table' to use lookup_bracket",
)
if parameter.bracket_axis is None:
raise RegistryValidationError(f"parameter {parameter.id!r} bracket_table requires bracket_axis")
if parameter.bracket_axis not in date_context:
raise RegistryValidationError(f"parameter {parameter.id!r} requires date axis {parameter.bracket_axis!r}")
selected = date_context[parameter.bracket_axis]
candidates = [
b for b in parameter.brackets if b.valid_from <= selected and (b.valid_to is None or selected <= b.valid_to)
]
if not candidates:
raise RegistryValidationError(
f"parameter {parameter.id!r} has no bracket valid for {selected.isoformat()}",
translated_message="errors.calc.bracket_no_window",
context={"parameter_id": parameter.id, "as_of": selected.isoformat()},
)
base = Decimal(base)
if base < Decimal("0"):
raise RegistryValidationError(
f"parameter {parameter.id!r} lookup_bracket received negative base {base}",
translated_message="errors.calc.bracket_negative_base",
context={"parameter_id": parameter.id, "base": str(base)},
)
sorted_brackets = sorted(candidates, key=lambda b: b.lower_bound)
selected_entry = None
for entry in sorted_brackets:
if entry.lower_bound <= base and (entry.upper_bound is None or base <= entry.upper_bound):
selected_entry = entry
break
if selected_entry is None:
raise RegistryValidationError(
f"parameter {parameter.id!r} has no bracket covering base {base}",
translated_message="errors.calc.bracket_no_coverage",
context={"parameter_id": parameter.id, "base": str(base)},
)
return selected_entry.fixed_addition + selected_entry.marginal_rate * (base - selected_entry.lower_bound)
[docs]
def resolve_parameter(parameter: ParameterDefinition, date_context: Mapping[str, date]) -> Decimal:
"""Resolve one dated value from a :class:`ParameterDefinition`.
Exactly one :class:`~domain.calculations.registry.DatedValue` must match
the selected date axes for the parameter lookup to be deterministic.
"""
if not parameter.values:
raise RegistryValidationError(f"parameter {parameter.id!r} has no dated values")
matches: list[DatedValue] = []
for value in parameter.values:
if value.date_axis not in date_context:
raise RegistryValidationError(f"parameter {parameter.id!r} requires date axis {value.date_axis!r}")
selected = date_context[value.date_axis]
if value.valid_from <= selected and (value.valid_to is None or selected <= value.valid_to):
matches.append(value)
if len(matches) != 1:
raise RegistryValidationError(
f"parameter {parameter.id!r} expected exactly one dated value, found {len(matches)}",
)
return matches[0].value
[docs]
def apply_rounding(value: Decimal, rounding: str | None) -> Decimal:
"""Apply a registry rounding rule to a decimal formula result.
``money-2`` uses :func:`core.money.round_to_cents`; ``integer`` uses
half-up quantization for registry-authored integer targets.
"""
if rounding is None:
return value
if rounding == "money-2":
return _round_to_cents(value)
if rounding == "integer":
return value.quantize(Decimal("1"), rounding=ROUND_HALF_UP)
raise RegistryValidationError(f"unsupported rounding rule {rounding!r}")
[docs]
def reject_non_decimal[Key](items: Mapping[Key, Decimal], label: str) -> None:
"""Reject non-decimal values before formula runtime consumption."""
for key, value in items.items():
if isinstance(value, bool) or not isinstance(value, Decimal):
raise RegistryValidationError(f"{label} {key!r} must be a Decimal")
[docs]
def reject_non_string[Key](values: Mapping[Key, str], label: str) -> None:
"""Reject empty or non-string external values before registry validation."""
for key, value in values.items():
if not isinstance(value, str) or not value:
raise RegistryValidationError(f"{label} {key!r} must be a non-empty string")
[docs]
def reject_unknown_external_values[Key: SupportsAllComparisons](
items: Mapping[Key, Decimal],
known_ids: set[Key],
label: str,
) -> None:
"""Reject external ids not declared by the current registry snapshot."""
unknown = sorted(set(items).difference(known_ids))
if unknown:
raise RegistryValidationError(f"unknown registry {label} ids: {unknown!r}")
def _require_arg_count(op: str, args: list[Decimal], count: int) -> None:
"""Require an exact operand count for a formula operation."""
if len(args) != count:
raise RegistryValidationError(f"formula op {op!r} expects {count} args, got {len(args)}")
def _require_non_empty(op: str, args: list[Decimal]) -> None:
"""Require at least one operand for aggregate formula operations."""
if not args:
raise RegistryValidationError(f"formula op {op!r} expects at least one arg")
@cache
def _default_read_parameter_authority(root: Path, source_root: Path) -> ValidatedRegistryAuthority:
from ._authority import ValidatedRegistryAuthority
return ValidatedRegistryAuthority.load(root, source_root=source_root)
[docs]
def read_parameter(
modelo_id: str,
revision_id: str,
parameter_id: str,
*,
date_context: Mapping[str, date],
registry_root: Path | None = None,
) -> Decimal:
"""Read a registry parameter through :class:`ValidatedRegistryAuthority`.
The ad hoc public helper loads the same validated registry authority used by
snapshot callers, narrows to the selected
:class:`~domain.calculations.registry.ModeloRevision`, and delegates the
dated value lookup to :func:`resolve_parameter`.
"""
from ....core.resources import bundled_path
from ._authority import ValidatedRegistryAuthority
source_root = bundled_path()
if registry_root is None:
root = bundled_path("registry", "aeat")
authority = _default_read_parameter_authority(root, source_root)
else:
root = registry_root
authority = ValidatedRegistryAuthority.load(root, source_root=source_root)
try:
modelo_match = authority.modelo(modelo_id)
except RegistrySnapshotError as exc:
raise RegistryValidationError(f"modelo {modelo_id!r} not registered in {root}") from exc
revision = modelo_match.revisions.get(revision_id)
if revision is None:
raise RegistryValidationError(f"modelo {modelo_id!r} has no revision {revision_id!r}")
parameter = next((p for p in revision.parameters if p.id == parameter_id), None)
if parameter is None:
raise RegistryValidationError(
f"parameter {parameter_id!r} not registered under modelo {modelo_id!r} revision {revision_id!r}",
)
return resolve_parameter(parameter, date_context)