"""Off-load-path record-design coverage and calculation-closure derivations.
:func:`calculation_closure_casilla_ids` and
:func:`calculation_closure_legal_refs` derive bounded closure projections from
a :class:`ModeloRevision`; :class:`DisenoCoverageReport` remains the advisory
full-Diseño inventory.
"""
from __future__ import annotations
import re
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from ._bindings import binding_source_casilla_ids, binding_source_modelo
from ._casilla_membership import casillas_by_id
from ._errors import RegistryValidationError
from ._ids import CasillaId, LegalRefId
from ._record_design_schema import RecordDesignSheet
from ._runtime_graph import expression_casilla_refs
from ._schema import CasillaDefinition, DataBindingDefinition, ModeloRevision
def _extract_record_design(path: Path) -> tuple[RecordDesignSheet, ...]:
from ._record_design import extract_record_design
return extract_record_design(path)
# ---------------------------------------------------------------------------
# Calculation-completeness manifest derivation and Diseño extraction
# (off-load-path)
# ---------------------------------------------------------------------------
#
# The derivations below run off the snapshot-build hot path; they are
# called only by manifest-authoring scripts, the off-load-path coverage
# report, and the drift re-verification test.
#
# - ``calculation_closure_casilla_ids`` enumerates a revision's calculation
# closure as canonical ``casilla.id`` values only. Reference tokens that do
# not name a declared casilla id remain unresolved and fail the manifest
# derivation instead of being reinterpreted as display metadata.
#
# - ``calculation_closure_record_design_metadata`` projects that canonical
# closure through each closure casilla's own registry ``(segmento, number)``
# metadata. This metadata is reviewed against Diseño where needed, while
# the manifest row also carries the canonical ``casilla.id`` that consumers
# resolve.
#
# - ``derive_calculation_completeness_casillas`` derives the
# *calculation-completeness manifest* casilla set from that closure:
# the modelo's calculation surface keyed on canonical ``casilla.id`` and
# carrying the registry metadata each closure casilla declares. For a multi-segment modelo it optionally
# verifies the derived record segments against the AEAT Diseño de
# Registros. This is the set the load-blocking completeness gate
# enforces.
#
# - ``derive_diseno_coverage_casillas`` extracts the *full* Diseño
# casilla set — every five-digit casilla tag AEAT embeds in a field
# description, accounting-statement data-entry fields included. It
# parses the multi-megabyte Diseño corpus and is the input to the
# off-load-path advisory coverage report that inventories form-level
# data coverage; it is NOT a load-blocking gate.
_CASILLA_TAG_RE = re.compile(r"\[(\d{5})\]")
"""Matches the five-digit casilla tag AEAT embeds in Diseño field text.
The official AEAT Diseño de Registros workbooks annotate every casilla
field with its five-digit casilla number in square brackets within the
field description (e.g. ``Liquidación III - ... - Base imponible
[00552]``). This regex extracts those tags so a derivation can enumerate
the ``(segmento, number)`` casilla set.
"""
[docs]
@dataclass(frozen=True)
class DerivedDisenoCasilla:
"""One casilla derived from a registry closure or AEAT Diseño workbook.
``segmento`` carries the AEAT record-segment code (the workbook sheet
name) for multi-segment modelos and is ``None`` for single-segment
modelos. ``number`` is the registry/display metadata under review.
``casilla_id`` is present only when the row came from a registry
calculation closure; full-Diseño coverage rows may have no registry
casilla yet and therefore leave it unset.
"""
segmento: str | None
number: str
casilla_id: CasillaId | None = None
def _binding_is_cross_modelo(binding: DataBindingDefinition, modelo_id: str) -> bool:
"""Return whether a binding names a foreign source modelo.
A binding is *cross-modelo* when its typed selector helper reports a
``source_modelo`` that is not the modelo whose closure is being derived.
A binding with no source modelo, or one set to ``modelo_id``, is a
*within-modelo* binding: its ``source_casilla_ids`` / ``source_casilla_id``
name casillas on the modelo being derived, and those casillas belong in
the modelo's own calculation closure.
"""
source_modelo = binding_source_modelo(binding)
if source_modelo is None:
return False
return source_modelo != modelo_id
def _walk_calculation_closure(
revision: ModeloRevision,
modelo_id: str,
*,
visit_token: Callable[[CasillaId], None],
visit_endpoint: Callable[[CasillaDefinition], None],
) -> None:
"""Walk the within-modelo calculation closure, dispatching each member.
Shared by :func:`calculation_closure_casilla_ids` and
:func:`calculation_closure_record_design_metadata`; ``visit_endpoint`` receives every
formula/binding endpoint casilla and ``visit_token`` every referenced
casilla token (formula targets, transitive expression refs,
verification-expectation operands, and within-modelo binding/relation
selectors).
"""
for casilla in revision.casillas:
if casilla.formula is not None or casilla.binding is not None:
visit_endpoint(casilla)
for formula in revision.formulas:
visit_token(formula.target_casilla_id)
for ref in expression_casilla_refs(formula.expression):
visit_token(ref)
for expectation in revision.verification_expectations:
for ref in expectation.computed_casilla_ids:
visit_token(ref)
for ref in expectation.reconciliation_total_casilla_ids.values():
visit_token(ref)
for binding in revision.bindings:
if _binding_is_cross_modelo(binding, modelo_id):
continue
for token in binding_source_casilla_ids(binding):
visit_token(token)
for relation in revision.relations:
if relation.source_modelo == modelo_id:
visit_token(relation.source_casilla_id)
[docs]
def calculation_closure_casilla_ids(revision: ModeloRevision, modelo_id: str) -> frozenset[CasillaId]:
"""Return canonical casilla ids in a revision's calculation closure.
The *calculation closure* is the set of casillas the cross-connecting
calculation engine traverses **within this modelo revision**:
- every ``formula.target_casilla_id`` casilla;
- every casilla referenced inside any ``formula.expression``, walked
transitively via the runtime-graph ``expression_casilla_refs``
walker;
- every casilla that declares a ``formula`` (a computed endpoint) or
a ``binding`` (a bound endpoint) — the engine-visible casillas;
- every verification-expectation operand casilla
(``computed_casilla_ids`` and the ``reconciliation_total_casilla_ids`` targets);
- every *within-modelo* binding ``source_casilla_ids`` / ``source_casilla_id``
selector casilla, and every *within-modelo*
``RelationDefinition.source_casilla_id``.
A binding ``source_casilla_ids`` / ``source_casilla_id`` selector — and a
``RelationDefinition.source_casilla_id`` — is excluded from this closure
**only when it is genuinely cross-modelo**: when the selector
explicitly names a ``source_modelo`` that differs from ``modelo_id``.
A cross-modelo selector's ``source_casilla_ids`` / ``source_casilla_id``
name casillas on that *foreign* modelo, not on the modelo whose
closure is being derived; the cross-modelo edge enters the current
modelo through the *bound* casilla — the current-modelo casilla that
declares the binding (or, for a relation, ``relation.target_binding``)
— which is already counted above as a binding endpoint. Folding a
foreign-modelo casilla id into this closure would make the
completeness gate demand it from the wrong modelo's registry.
A selector that omits ``source_modelo`` or sets it equal to
``modelo_id`` is a *within-modelo* selector: a ``previous_filing``
self-binding or a ``previous_period`` self-relation names a casilla
on the modelo being derived, so that casilla is a genuine closure
member and is kept.
References are not normalised through record-design metadata. A formula,
binding, relation, or verification token must already be the canonical
``casilla.id`` for this revision. A reference token that matches no
declared casilla id is kept verbatim so the manifest derivation and
registry validation fail loudly on the unresolved canonical reference.
Args:
revision: The :class:`ModeloRevision` whose formula and binding graph
is walked to derive the closure.
modelo_id: The AEAT modelo identifier used to exclude cross-modelo
selector casillas from the closure.
"""
closure: set[CasillaId] = set()
_walk_calculation_closure(
revision,
modelo_id,
visit_token=closure.add,
visit_endpoint=lambda casilla: closure.add(casilla.id),
)
return frozenset(closure)
[docs]
def calculation_closure_legal_refs(revision: ModeloRevision, modelo_id: str) -> frozenset[LegalRefId]:
"""Return legal references carried by a revision's calculation closure.
Legal-ref projection of :func:`calculation_closure_casilla_ids`. The
completeness manifest is the reviewed ledger for the calculation closure,
so its own ``legal_refs`` must equal the refs carried by the closure
casillas plus their endpoint formula/binding definitions and verification
expectations.
Unresolved closure tokens are skipped here because
:func:`derive_calculation_completeness_casillas` and the manifest drift
gate already surface missing required casillas explicitly.
Args:
revision: The :class:`ModeloRevision` whose calculation closure legal
references are projected.
modelo_id: Modelo identifier used to scope cross-modelo selectors.
"""
declared_by_id = casillas_by_id(revision)
formulas_by_id = {formula.id: formula for formula in revision.formulas}
bindings_by_id = {binding.id: binding for binding in revision.bindings}
legal_refs: set[LegalRefId] = set()
for casilla_id in calculation_closure_casilla_ids(revision, modelo_id):
casilla = declared_by_id.get(casilla_id)
if casilla is None:
continue
legal_refs.update(casilla.legal_refs)
if casilla.formula is not None:
legal_refs.update(formulas_by_id[casilla.formula].legal_refs)
if casilla.binding is not None:
legal_refs.update(bindings_by_id[casilla.binding].legal_refs)
for binding_id in casilla.alternate_bindings:
legal_refs.update(bindings_by_id[binding_id].legal_refs)
for expectation in revision.verification_expectations:
legal_refs.update(expectation.legal_refs)
return frozenset(legal_refs)
[docs]
def derive_calculation_completeness_casillas(
revision: ModeloRevision,
modelo_id: str,
*,
multi_segment: bool,
diseno_path: Path | None = None,
) -> tuple[DerivedDisenoCasilla, ...]:
r"""Return the calculation-completeness manifest casilla set for a revision.
Derives the modelo's *calculation closure*
(:func:`calculation_closure_casilla_ids`) and carries each closure casilla's
own registry ``(segmento, number)`` metadata. The closure
bounds the manifest to exactly the casillas the cross-connecting
calculation engine traverses; the registry's canonical ``casilla.id`` —
not a five-digit AEAT Diseño tag — names each manifest entry.
This derivation is *vocabulary-agnostic*. Only Modelo 200's registry
casilla ``number``\\ s are genuine five-digit AEAT Diseño tags; the
other calculation-bearing modelos identify casillas by semantic slug
(``iva.cuota-devengada-total``) or short ordinal (``01``-``19``). The
manifest is therefore derived from the modelo's calculation surface
keyed on canonical ``casilla.id`` and carrying the registry metadata
each closure casilla declares, so it can be authored for any
calculation-bearing modelo regardless of its casilla vocabulary.
For a ``multi_segment`` modelo the result is *segment-aware*. A
multi-segment modelo reuses the same casilla number across distinct
record segments and its formulas reference casillas by the
segment-carrying composite ``id``, so the metadata-preserving closure
(:func:`calculation_closure_record_design_metadata`) already pins each closure
casilla to the exact record segment the calculation surface uses.
When ``diseno_path`` is supplied each segment-scoped metadata pair is
additionally **verified against the AEAT Diseño de Registros**: the
Diseño remains authoritative on which record segment carries a
number, and a pinned ``(segmento, number)`` absent from the Diseño is
a derivation error.
For a single-segment modelo ``segmento`` is left unset and the
closure casilla's registry ``number`` alone carries the reviewed metadata; no Diseño
is required because a single-segment modelo's metadata is unambiguous
without one.
A closure reference that resolves to no declared casilla is omitted
from the derived set — the calculation-completeness gate then fires
on the missing required casilla when it compares the manifest to the
declared casillas, which is the gate fulfilling its mission. The
drift / coverage tests surface such gaps explicitly.
This is an off-load-path tool. When ``diseno_path`` is supplied it
parses the multi-megabyte Diseño corpus and must never run on the
snapshot-build path.
Args:
revision: The :class:`ModeloRevision` whose calculation closure to derive into a manifest.
modelo_id: Modelo identifier; cross-modelo selectors whose
``source_modelo`` differs from ``modelo_id`` are excluded from
the closure.
multi_segment: When True, the manifest is segment-aware and a casilla
number repeated across distinct segments produces distinct manifest
rows; when False, segment metadata is dropped from the manifest key.
diseno_path: Optional path to an AEAT Diseño workbook used to
cross-check that every derived manifest casilla also appears on
the published record design.
Returns:
Tuple of :class:`DerivedDisenoCasilla` representing the calculation-completeness manifest.
"""
declared_by_id = casillas_by_id(revision)
diseno_pairs: frozenset[tuple[str, str]] | None = None
if diseno_path is not None:
diseno_pairs = frozenset(
(sheet.name, number)
for sheet in _extract_record_design(diseno_path)
for number in _sheet_record_numbers(sheet)
)
ordered: list[DerivedDisenoCasilla] = []
for casilla_id in sorted(calculation_closure_casilla_ids(revision, modelo_id)):
casilla = declared_by_id.get(casilla_id)
if casilla is None:
raise RegistryValidationError(
f"calculation-completeness derivation: closure reference {casilla_id!r} "
"is not a declared canonical casilla.id",
)
segmento = casilla.segmento
number = casilla.number
if not multi_segment:
if casilla.internal_only:
# App-internal computed casilla intentionally absent from the
# AEAT-published structure (e.g. a regulatory ceiling materialised
# so verification predicates can bound an operator-elective
# amount). The schema validator guarantees it carries no
# export_refs and is formula-derived; it is not an AEAT box, so it
# never appears in the completeness manifest.
continue
ordered.append(DerivedDisenoCasilla(segmento=None, number=number, casilla_id=casilla.id))
continue
if casilla.internal_only:
# App-internal computed casilla intentionally absent from the
# AEAT-published Diseño de Registros (e.g. a regulatory
# ceiling materialised so verification predicates can bound
# an operator-elective amount). The schema validator
# guarantees such a casilla carries no export_refs and is
# formula-derived; the Diseño-presence check is skipped while
# the segment-carrying metadata is preserved for downstream
# manifest consumers.
ordered.append(DerivedDisenoCasilla(segmento=segmento, number=number, casilla_id=casilla.id))
continue
if diseno_pairs is not None and segmento is not None and (segmento, number) not in diseno_pairs:
raise RegistryValidationError(
f"calculation-completeness derivation: casilla {number!r} is "
f"declared under segmento {segmento!r} but the AEAT Diseño de "
"Registros does not carry it under that segment",
)
ordered.append(DerivedDisenoCasilla(segmento=segmento, number=number, casilla_id=casilla.id))
return tuple(ordered)
[docs]
def derive_diseno_coverage_casillas(
path: Path,
*,
multi_segment: bool,
) -> tuple[DerivedDisenoCasilla, ...]:
"""Return :class:`DerivedDisenoCasilla` items for the full casilla set declared by a Diseño.
Runs read-only record-design extraction against the official AEAT
Diseño de Registros source at ``path`` and collects *every*
five-digit casilla tag embedded in the field descriptions, including
the accounting-statement data-entry fields that feed no calculation.
This is the input to the off-load-path advisory coverage report that
inventories form-level data coverage. It is intentionally NOT a
load-blocking gate: a modelo whose registry is not yet exhaustively
backfilled against the full Diseño is reported as having a coverage
gap, not failed at load. The load-blocking gate is keyed on the
bounded calculation closure
(:func:`derive_calculation_completeness_casillas`) instead.
For a ``multi_segment`` modelo (e.g. Modelo 200, which reuses the
same casilla number across distinct record segments) every casilla
carries the workbook sheet name as its ``segmento``, so the same
number under two segments yields two distinct metadata pairs. For a
single-segment modelo ``segmento`` is left unset and the bare number
alone identifies the casilla; a number that recurs across sheets of a
single-segment Diseño collapses to one metadata pair, matching the
bare-number registry behaviour.
This is an off-load-path tool: it parses the multi-megabyte Diseño
corpus and must never run on the snapshot-build path.
"""
sheets = _extract_record_design(path)
if multi_segment:
seen: set[tuple[str | None, str]] = set()
ordered: list[DerivedDisenoCasilla] = []
for sheet in sheets:
for number in _sheet_record_numbers(sheet):
metadata_pair = (sheet.name, number)
if metadata_pair in seen:
continue
seen.add(metadata_pair)
ordered.append(DerivedDisenoCasilla(segmento=sheet.name, number=number))
return tuple(ordered)
seen_numbers: set[str] = set()
bare: list[DerivedDisenoCasilla] = []
for sheet in sheets:
for number in _sheet_record_numbers(sheet):
if number in seen_numbers:
continue
seen_numbers.add(number)
bare.append(DerivedDisenoCasilla(segmento=None, number=number))
return tuple(bare)
[docs]
@dataclass(frozen=True)
class DisenoCoverageReport:
"""An off-load-path advisory inventory of one revision's Diseño coverage.
Compares a modelo revision's declared casillas against the *full*
AEAT Diseño de Registros casilla set — every five-digit casilla tag
AEAT embeds in the form's field descriptions, accounting-statement
data-entry fields included.
This report is **advisory**: it is produced off the snapshot-build
load path and never reds a load. A modelo whose registry is not yet
exhaustively backfilled against the full Diseño is reported here as
having a coverage gap, surfaced as information for follow-up
authoring. The load-blocking gate is the bounded
calculation-completeness gate, not this full-Diseño inventory — that
Calculation-completeness is enforced at load; full-Diseño coverage is
inventoried off-load-path as an advisory follow-up surface.
Fields:
- ``modelo_id`` / ``revision_id`` identify the revision inventoried.
- ``diseno_casillas`` is the full ``(segmento, number)`` set the
Diseño declares.
- ``covered_casillas`` is the subset the registry also declares —
the Diseño casillas the registry has backfilled.
- ``coverage_gap_casillas`` is the subset the Diseño declares that
the registry does not — the advisory follow-up inventory.
"""
modelo_id: str
revision_id: str
diseno_casillas: tuple[DerivedDisenoCasilla, ...]
covered_casillas: tuple[DerivedDisenoCasilla, ...]
coverage_gap_casillas: tuple[DerivedDisenoCasilla, ...]
@property
def diseno_casilla_count(self) -> int:
"""Total ``(segmento, number)`` casillas the Diseño declares."""
return len(self.diseno_casillas)
@property
def covered_count(self) -> int:
"""Diseño casillas the registry also declares."""
return len(self.covered_casillas)
@property
def coverage_gap_count(self) -> int:
"""Diseño casillas the registry does not yet declare."""
return len(self.coverage_gap_casillas)
[docs]
def build_diseno_coverage_report(
path: Path,
modelo_id: str,
revision: ModeloRevision,
*,
multi_segment: bool,
) -> DisenoCoverageReport:
"""Return the off-load-path full-Diseño coverage advisory report for a revision.
Extracts the full AEAT Diseño de Registros casilla set
(:func:`derive_diseno_coverage_casillas`) and compares it against the
revision's declared casillas, keyed on the ``(segmento, number)``
metadata. The result is a :class:`DisenoCoverageReport` that
inventories how much of the form's data surface the registry covers
and which Diseño casillas remain to be authored.
This is an **advisory** inventory, never a load gate. It is produced
off the snapshot-build path — it parses the multi-megabyte Diseño
corpus — and must never run on the load path. A coverage gap reported
here does not fail any modelo: the load-blocking enforcement is the
bounded calculation-completeness gate; calculation-completeness is
enforced at load while full-Diseño coverage is inventoried
off-load-path.
For a ``multi_segment`` modelo the comparison is segment-aware: a
Diseño casilla under segment ``S`` is "covered" only when the
registry declares a casilla with the same ``(S, number)`` metadata. For
a single-segment modelo ``segmento`` is unset on both sides and the
bare number alone identifies the casilla.
Args:
path: Path to the official AEAT Diseño de Registros source file.
modelo_id: The AEAT modelo identifier for the coverage report.
revision: The :class:`ModeloRevision` whose declared casillas are
compared against the extracted Diseño casilla set.
multi_segment: Whether the modelo uses segment-qualified casilla ids.
"""
diseno = derive_diseno_coverage_casillas(path, multi_segment=multi_segment)
declared_metadata = {(casilla.segmento, casilla.number) for casilla in revision.casillas}
covered: list[DerivedDisenoCasilla] = []
gap: list[DerivedDisenoCasilla] = []
for casilla in diseno:
if (casilla.segmento, casilla.number) in declared_metadata:
covered.append(casilla)
else:
gap.append(casilla)
return DisenoCoverageReport(
modelo_id=modelo_id,
revision_id=revision.id,
diseno_casillas=diseno,
covered_casillas=tuple(covered),
coverage_gap_casillas=tuple(gap),
)
def _sheet_record_numbers(sheet: RecordDesignSheet) -> tuple[str, ...]:
"""Return the official record-design numeric tags declared in one sheet, in field order."""
numbers: list[str] = []
seen: set[str] = set()
for design_field in sheet.fields:
for text in (design_field.description, design_field.validation, design_field.content):
if not text:
continue
for match in _CASILLA_TAG_RE.finditer(text):
number = match.group(1)
if number in seen:
continue
seen.add(number)
numbers.append(number)
return tuple(numbers)
__all__ = [
"DerivedDisenoCasilla",
"DisenoCoverageReport",
"build_diseno_coverage_report",
"calculation_closure_casilla_ids",
"calculation_closure_legal_refs",
"calculation_closure_record_design_metadata",
"derive_calculation_completeness_casillas",
"derive_diseno_coverage_casillas",
]