"""One observation-fold helper for cross-filing fold-ins.
The single gather-and-fold primitive shared by both relation fold paths (the
application-layer relation prefill and the domain-layer
:func:`domain.calculations.registry.relations.resolve_relation_values_from_observations`).
It matches the source filings a
:class:`~domain.calculations.registry.relations.RegistryFoldRequirement` declares and
extracts the source casilla value per period, then folds the gathered values
through the requirement's declared ``copy`` / ``sum`` aggregation to one
:class:`~decimal.Decimal`.
Lives in the domain registry package (not the application source mesh) because
the domain relation resolver consumes it and the hexagonal direction forbids a
domain module importing the application layer.
See Also:
:mod:`domain.calculations.registry.relations`
Domain relation resolver that gathers and folds requirements here.
:mod:`domain.calculations.registry.bindings_previous_filing`
Previous-filing binding resolver that reuses :func:`fold_sum_or_copy`.
"""
from __future__ import annotations
from decimal import Decimal
from typing import TYPE_CHECKING
from ....core.period import Period
from .errors import RegistryValidationError
if TYPE_CHECKING:
from .bindings import RegistryModeloObservation
from .relations import RegistryFoldRequirement
[docs]
def gather_observed_requirement_values(
requirement: RegistryFoldRequirement,
observations: tuple[RegistryModeloObservation, ...],
) -> tuple[Decimal, ...]:
"""Return the source-casilla values matched for one fold requirement, per period.
Matches exactly one
:class:`~domain.calculations.registry.bindings.RegistryModeloObservation` per
declared source period and extracts the requirement's single source
:class:`~core.casilla_id.CasillaId` value from each. The
returned tuple carries one value per ``requirement.periods`` entry, in
declaration order, ready for :func:`fold_observed_requirement_values`.
"""
source_casilla_id = requirement.source_casilla_ids[0]
values: list[Decimal] = []
for source_period in _filed_cadence_periods(requirement, observations):
matches = tuple(
observation
for observation in observations
if observation.modelo == requirement.source_modelo
and observation.filing_year == requirement.filing_year
and observation.period == source_period
)
if len(matches) != 1:
raise RegistryValidationError(
f"relation requirement {_requirement_ids(requirement)!r} expected one observed filing "
f"{requirement.source_modelo!r}/{requirement.filing_year}/{source_period!r}, found {len(matches)}",
)
value = matches[0].casilla_values.get(source_casilla_id)
if value is None:
raise RegistryValidationError(
f"relation requirement {_requirement_ids(requirement)!r} requires observed source casilla id "
f"{source_casilla_id!r} from "
f"{requirement.source_modelo!r}/{requirement.filing_year}/{source_period!r}",
)
values.append(value)
return tuple(values)
def _filed_cadence_periods(
requirement: RegistryFoldRequirement,
observations: tuple[RegistryModeloObservation, ...],
) -> tuple[str, ...]:
"""Return the declared periods of the one cadence the source was filed under.
A source modelo may be declared over alternative cadences -- Modelo 111 is
filed quarterly or, by a large company, monthly (RIRPF art. 108) -- and a
filer files one of them for the year, never both. When the declared periods
span more than one cadence, exactly one cadence must be completely observed;
none or several leave the requirement unresolved rather than folding a mix.
A single-cadence declaration is returned unchanged.
"""
groups: dict[str, list[str]] = {}
for source_period in requirement.periods:
kind = Period.from_year_and_code(requirement.filing_year, source_period).kind
groups.setdefault(str(kind), []).append(source_period)
if len(groups) < 2:
return requirement.periods
def observed(source_period: str) -> bool:
return any(
observation.modelo == requirement.source_modelo
and observation.filing_year == requirement.filing_year
and observation.period == source_period
for observation in observations
)
filed = [tuple(periods) for periods in groups.values() if all(observed(period) for period in periods)]
if len(filed) != 1:
raise RegistryValidationError(
f"relation requirement {_requirement_ids(requirement)!r} needs exactly one completely filed cadence of "
f"{requirement.source_modelo!r}/{requirement.filing_year}; found {len(filed)}",
)
return filed[0]
[docs]
def fold_sum_or_copy(
op: str,
values: tuple[Decimal, ...] | list[Decimal],
*,
subject: str,
copy_unit: str,
) -> Decimal:
"""Fold a value sequence through the shared ``sum`` / ``copy`` arithmetic.
The one place ``sum`` (add every value) and ``copy`` (require exactly one
value and return it) are implemented. ``subject`` and ``copy_unit`` carry the
caller's diagnostic vocabulary so a refused copy names the right entity
(a relation requirement's observation, or a binding's source casilla). An op
that is neither ``sum`` nor ``copy`` is rejected; callers handle any other op
(e.g. the Modelo 130 ``prior_pagos_fraccionados`` identity) before delegating.
Used by both :func:`fold_observed_requirement_values` and
:func:`domain.calculations.registry.bindings_previous_filing.resolve_previous_filing_binding_values`.
"""
if op == "copy":
if len(values) != 1:
raise RegistryValidationError(f"{subject} copy aggregation requires one {copy_unit}")
return values[0]
if op == "sum":
return sum(values, Decimal("0"))
raise RegistryValidationError(f"{subject} uses unsupported aggregation op {op!r}")
[docs]
def fold_observed_requirement_values(
requirement: RegistryFoldRequirement,
values: tuple[Decimal, ...],
) -> Decimal:
"""Fold per-period source values through a requirement's ``copy`` / ``sum`` op.
``copy`` requires exactly one gathered value and returns it; ``sum`` adds the
gathered values. This is the one fold both relation paths apply to a
:class:`~domain.calculations.registry.relations.RegistryFoldRequirement`; the
period match is :func:`gather_observed_requirement_values`.
"""
aggregation_op = requirement.aggregation_op
# Every relation-derived requirement declares its aggregation op; only the
# unrelated same-modelo previous_filing producer leaves it unset, and this
# fold is the one both RELATION paths apply, per the docstring above.
if aggregation_op is None:
raise RegistryValidationError(
f"relation requirement {_requirement_ids(requirement)!r} declares no aggregation op to fold with",
)
return fold_sum_or_copy(
aggregation_op,
values,
subject=f"relation requirement {_requirement_ids(requirement)!r}",
copy_unit="observation",
)
def _requirement_ids(requirement: RegistryFoldRequirement) -> tuple[str, ...]:
"""Return the binding ids carried by either kind of fold requirement."""
return requirement.target_bindings or requirement.binding_ids
[docs]
def resolve_observed_requirement_value(
requirement: RegistryFoldRequirement,
observations: tuple[RegistryModeloObservation, ...],
) -> Decimal:
"""Gather and fold one requirement to a single :class:`~decimal.Decimal`.
Convenience wrapper for callers that already hold the normalized
:class:`~domain.calculations.registry.bindings.RegistryModeloObservation` rows.
"""
return fold_observed_requirement_values(
requirement,
gather_observed_requirement_values(requirement, observations),
)