"""Private backend for assembling the canonical Modelo work review.
Core types:
:class:`~cadrumo.domain.modelos.calculation_revision.CalculationRevision`,
:class:`~cadrumo.domain.calculations.registry.bindings.CasillaObservation`,
:class:`~cadrumo.domain.calculations.registry.schema.RegistrySnapshot`.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from typing import TYPE_CHECKING
from ...core.casilla_id import CasillaId
from ...core.estado_casilla_oficial import EstadoCasillaOficial
from ...core.identity.bucket import BucketId
from ...core.modelo_work_progress_state import ModeloWorkProgressState
from ...core.period import Period
from ...domain.calculations.registry.binding_targets import casillas_by_binding
from ...domain.calculations.registry.bindings import CasillaObservation
from ...domain.calculations.registry.export import (
clasificar_casillas_oficiales,
derive_export_layouts_from_bindings,
)
from ...domain.calculations.registry.export_parse import xml_dictionary_entries
from ...domain.calculations.registry.handoffs import (
RelationConsumptionChannel,
relation_consumption_channels,
relation_consumption_index,
)
from ...domain.calculations.registry.ids import (
BindingId,
FormulaId,
RevisionId,
)
from ...domain.calculations.registry.relations import relation_prefill_bindings_for_period
from ...domain.calculations.registry.runtime_graph import (
enum_consumed_binding_ids,
expression_binding_refs,
expression_casilla_refs,
revision_date_binding_ids,
)
from ...domain.calculations.registry.schema import BindingDefinition, FormulaDefinition, RegistrySnapshot
from ...domain.calculations.registry.schema_input_kind import InputKind
from ...domain.calculations.registry.schema_surfaces import CasillaDefinition
from ...domain.calculations.registry.source_byte_availability import layout_embedded_source_ids
from ...domain.filing.schema import ModeloScalar, ModeloValueKind
from ...domain.modelos.calculation_revision import CalculationRevision
from ...domain.modelos.codes import ModeloCode
from ...domain.modelos.protocols import (
CalculationRevisionCatalogueRepositoryProtocol,
VerificationReportCatalogueRepositoryProtocol,
)
from ...domain.modelos.verification_report import (
OPERATOR_ACTION_BY_MODELO_VERIFICATION_FINDING_KIND,
ModeloVerificationFinding,
ModeloVerificationFindingSeverity,
VerificationCompletenessStatus,
VerificationReport,
)
from ...domain.modelos.work_unit import WorkUnit, WorkUnitCatalogue
from ...domain.modelos.work_unit_repository import WorkUnitCatalogueRepositoryProtocol
from ..calculations.verification_report_gate import require_verification_report_coordinates_current
from ._row_source_identity_replay import revision_row_source_fingerprints_for_review
from .action_errors import CalculationRevisionNotFoundError, StoredCalculationDriftError, WorkUnitNotFoundError
from .calculation_revision_gate import require_calculation_revision_coordinates_current
from .work_review import (
BlockerRef,
ModeloWorkBindingOrigin,
ModeloWorkFormulaOrigin,
ModeloWorkOriginAnomaly,
ModeloWorkProgress,
ModeloWorkProgressDenominator,
ModeloWorkRelationConsumption,
ModeloWorkReview,
ModeloWorkReviewCasilla,
)
from .work_selection import ModeloWorkSelectorRequest, ModeloWorkSelectorState, select_modelo_work_resolution
if TYPE_CHECKING:
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
@dataclass(frozen=True, slots=True)
class _ReviewRowContext:
"""Precomputed registry and persistence facts shared by every review row."""
revision: CalculationRevision | None
bindings_by_id: Mapping[BindingId, BindingDefinition]
formulas_by_id: Mapping[FormulaId, FormulaDefinition]
binding_to_casillas: Mapping[BindingId, tuple[CasillaId, ...]]
fold_binding_ids: frozenset[BindingId]
relation_channels: Mapping[BindingId, tuple[RelationConsumptionChannel, ...]]
persisted_decimal_bindings: Mapping[BindingId, Decimal]
persisted_binding_ids: frozenset[BindingId]
estados_casillas_oficiales: Mapping[CasillaId, EstadoCasillaOficial]
official_references: Mapping[CasillaId, str | None]
blocking_findings: tuple[ModeloVerificationFinding, ...]
def _work_unit_for_target(
*,
bucket_id: BucketId,
modelo: ModeloCode,
filing_year: int,
period: Period,
registry_revision_id: RevisionId,
catalogue: WorkUnitCatalogue,
) -> WorkUnit:
resolution = select_modelo_work_resolution(
ModeloWorkSelectorRequest(
bucket_id=bucket_id,
modelo=modelo,
filing_year=filing_year,
period=period,
revision_id=registry_revision_id,
),
catalogue=catalogue,
bucket_id=bucket_id,
)
if resolution.state is ModeloWorkSelectorState.ABSENT or resolution.work_unit is None:
raise WorkUnitNotFoundError(
translated_message="application.modelo.errors.work_unit_not_found",
context={
"bucket_id": bucket_id,
"modelo": modelo,
"filing_year": filing_year,
"period": period.registry_token,
},
)
return resolution.work_unit
def _current_revision(
work_unit: WorkUnit,
repository: CalculationRevisionCatalogueRepositoryProtocol,
*,
operation: PinnedAuthorityOperation,
) -> CalculationRevision | None:
revision_id = work_unit.current_calculation_revision_id
if revision_id is None:
return None
revision = repository.load().get(revision_id)
if revision is None:
raise CalculationRevisionNotFoundError(
translated_message="application.modelo.errors.calculation_revision_not_found",
context={"calculation_revision_id": revision_id, "work_unit_id": work_unit.work_unit_id},
)
if revision.work_unit_id != work_unit.work_unit_id:
raise CalculationRevisionNotFoundError(
translated_message="application.modelo.errors.calculation_revision_not_found",
context={
"calculation_revision_id": revision.calculation_revision_id,
"work_unit_id": work_unit.work_unit_id,
"stored_work_unit_id": revision.work_unit_id,
},
)
require_calculation_revision_coordinates_current(revision, operation=operation)
return revision
def _latest_verification(
revision: CalculationRevision | None,
repository: VerificationReportCatalogueRepositoryProtocol,
*,
operation: PinnedAuthorityOperation,
) -> VerificationReport | None:
if revision is None:
return None
reports = require_verification_report_coordinates_current(
repository.load(),
operation=operation,
).for_calculation_revision(revision.calculation_revision_id)
return reports[-1] if reports else None
def _persisted_decimal_bindings(
*, snapshot: RegistrySnapshot, revision: CalculationRevision | None
) -> Mapping[BindingId, Decimal]:
"""Read only historical decimal replay facts, never live profile state."""
if revision is None:
empty_bindings: dict[BindingId, Decimal] = {}
return empty_bindings
enum_ids = enum_consumed_binding_ids(snapshot.revision)
date_ids = revision_date_binding_ids(snapshot.revision)
decimal_bindings: dict[BindingId, Decimal] = {}
for binding_id, raw_value in revision.binding_overrides.items():
if binding_id in enum_ids or binding_id in date_ids:
continue
try:
parsed_value: Decimal = Decimal(str(raw_value))
decimal_bindings[binding_id] = parsed_value
except InvalidOperation as exc:
raise StoredCalculationDriftError(
translated_message="errors.storage.stored_data_validation_boundary",
context={
"calculation_revision_id": revision.calculation_revision_id,
"binding_id": binding_id,
},
) from exc
return decimal_bindings
def _official_references(
snapshot: RegistrySnapshot,
estados_casillas_oficiales: Mapping[CasillaId, EstadoCasillaOficial],
*,
source_payloads: Mapping[str, bytes],
) -> Mapping[CasillaId, str | None]:
layouts = derive_export_layouts_from_bindings(snapshot.revision)
xml_paths: dict[CasillaId, str] = {}
for layout in layouts:
if layout.dictionary_source_ref is None:
continue
for entry in xml_dictionary_entries(
layout,
sources=snapshot.sources,
source_payloads=source_payloads,
):
if entry.casilla_id is not None:
xml_paths.setdefault(entry.casilla_id, entry.path)
return {
casilla.id: xml_paths.get(casilla.id, casilla.number)
if estados_casillas_oficiales[casilla.id] is EstadoCasillaOficial.ADDRESSED
else None
for casilla in snapshot.revision.casillas
}
def _blocker_ref(finding: ModeloVerificationFinding) -> BlockerRef:
return BlockerRef(
axis=OPERATOR_ACTION_BY_MODELO_VERIFICATION_FINDING_KIND[finding.kind],
native_code=finding.kind.value,
facts=finding.message_facts,
)
def _casilla_observation(
revision: CalculationRevision | None,
casilla_id: CasillaId,
) -> CasillaObservation | None:
if revision is None:
return None
return next(
(item for item in revision.observations if item.casilla_id == casilla_id),
None,
)
def _persisted_binding_values(
binding_ids: tuple[BindingId, ...],
resolved_bindings: Mapping[BindingId, Decimal],
) -> tuple[Decimal, ...]:
return tuple(resolved_bindings[item] for item in binding_ids if item in resolved_bindings)
def _classify_observed_value(
*,
observation: CasillaObservation,
input_kind: InputKind,
formula_id: FormulaId | None,
binding_ids: tuple[BindingId, ...],
resolved_bindings: Mapping[BindingId, Decimal],
) -> tuple[ModeloValueKind, ModeloScalar, ModeloWorkOriginAnomaly | None]:
if formula_id is not None:
return ModeloValueKind.COMPUTED, observation.value, None
if input_kind is not InputKind.BOUND:
return ModeloValueKind.LITERAL, observation.value, None
if observation.absent_by_design:
return ModeloValueKind.INHERITED, observation.value, None
binding_values = _persisted_binding_values(binding_ids, resolved_bindings)
if binding_values and observation.value not in binding_values:
return ModeloValueKind.LITERAL, observation.value, ModeloWorkOriginAnomaly.OPERATOR_OVERRIDE
return ModeloValueKind.INHERITED, observation.value, None
def _realised_value(
*,
casilla_id: CasillaId,
input_kind: InputKind,
formula_id: FormulaId | None,
binding_ids: tuple[BindingId, ...],
revision: CalculationRevision | None,
resolved_bindings: Mapping[BindingId, Decimal],
) -> tuple[ModeloValueKind, ModeloScalar, ModeloWorkOriginAnomaly | None]:
"""Classify the realised value from persisted observations and replay facts.
``OPERATOR_OVERRIDE`` is emitted only for an observable disagreement between
a persisted bound value and the persisted casilla observation. Caller
casilla inputs have final precedence during calculation, so that mismatch is
durable evidence of an override. An equal-value explicit input is not
distinguishable after persistence and therefore does not claim an anomaly.
"""
observation = _casilla_observation(revision, casilla_id)
if observation is None:
anomaly = ModeloWorkOriginAnomaly.BROKEN_CALCULATION_CHAIN if input_kind is InputKind.COMPUTED else None
return ModeloValueKind.EMPTY, None, anomaly
return _classify_observed_value(
observation=observation,
input_kind=input_kind,
formula_id=formula_id,
binding_ids=binding_ids,
resolved_bindings=resolved_bindings,
)
def _casilla_binding_ids(
casilla_id: CasillaId,
binding_to_casillas: Mapping[BindingId, tuple[CasillaId, ...]],
) -> tuple[BindingId, ...]:
return tuple(binding_id for binding_id, casilla_ids in binding_to_casillas.items() if casilla_id in casilla_ids)
def _formula_origin(formula: FormulaDefinition | None) -> ModeloWorkFormulaOrigin | None:
if formula is None:
return None
return ModeloWorkFormulaOrigin(
formula_id=formula.id,
operand_refs=tuple(
dict.fromkeys(
(
*expression_casilla_refs(formula.expression),
*expression_binding_refs(formula.expression),
),
),
),
)
def _relation_consumptions(
*,
binding_ids: tuple[BindingId, ...],
formula: FormulaDefinition | None,
context: _ReviewRowContext,
) -> tuple[ModeloWorkRelationConsumption, ...]:
formula_binding_ids: set[BindingId] = set() if formula is None else set(expression_binding_refs(formula.expression))
candidate_ids = {
binding_id for binding_id in (*binding_ids, *formula_binding_ids) if binding_id in context.fold_binding_ids
}
return tuple(
ModeloWorkRelationConsumption(
relation_id=binding_id,
channels=context.relation_channels[binding_id],
)
for binding_id in sorted(context.fold_binding_ids)
if binding_id in candidate_ids
)
def _binding_origins(
binding_ids: tuple[BindingId, ...],
context: _ReviewRowContext,
) -> tuple[ModeloWorkBindingOrigin, ...]:
return tuple(
ModeloWorkBindingOrigin(
binding_id=binding_id,
source=context.bindings_by_id[binding_id].source,
resolved=binding_id in context.persisted_binding_ids,
)
for binding_id in binding_ids
)
def _review_casilla(
casilla: CasillaDefinition,
context: _ReviewRowContext,
) -> ModeloWorkReviewCasilla:
binding_ids = _casilla_binding_ids(casilla.id, context.binding_to_casillas)
formula = context.formulas_by_id.get(casilla.formula) if casilla.formula is not None else None
realised_kind, value, anomaly = _realised_value(
casilla_id=casilla.id,
input_kind=casilla.input_kind,
formula_id=casilla.formula,
binding_ids=binding_ids,
revision=context.revision,
resolved_bindings=context.persisted_decimal_bindings,
)
return ModeloWorkReviewCasilla(
casilla_id=casilla.id,
number=casilla.number,
segmento=casilla.segmento,
official_reference=context.official_references[casilla.id],
section_path=casilla.section,
label=casilla.label,
data_type=casilla.data_type,
constraints=casilla.constraints,
declared_input_kind=casilla.input_kind,
concrete_bindings=_binding_origins(binding_ids, context),
concrete_formula=_formula_origin(formula),
relation_consumption=_relation_consumptions(
binding_ids=binding_ids,
formula=formula,
context=context,
),
realised_kind=realised_kind,
value=value,
origin_anomaly=anomaly,
estado_casilla_oficial=context.estados_casillas_oficiales[casilla.id],
legal_refs=tuple(casilla.legal_refs),
source_refs=tuple(casilla.source_refs),
formula_id=casilla.formula,
blocked_by=tuple(
_blocker_ref(finding) for finding in context.blocking_findings if finding.casilla_id == casilla.id
),
)
def _review_row_context(
*,
snapshot: RegistrySnapshot,
revision: CalculationRevision | None,
blocking_findings: tuple[ModeloVerificationFinding, ...],
operation: PinnedAuthorityOperation,
) -> _ReviewRowContext:
# Only embedded dictionary and XSD bytes are fetched; the rows still cite
# every casilla source reference, including record designs without bytes.
layouts = derive_export_layouts_from_bindings(snapshot.revision)
source_ids = layout_embedded_source_ids(layouts, sources=snapshot.sources)
source_payloads = {source_id: operation.source_evidence(source_id).payload for source_id in source_ids}
estados_casillas_oficiales = clasificar_casillas_oficiales(
snapshot.revision,
sources=snapshot.sources,
source_payloads=source_payloads,
)
consumption_index = relation_consumption_index(snapshot.revision)
return _ReviewRowContext(
revision=revision,
bindings_by_id={binding.id: binding for binding in snapshot.revision.bindings},
formulas_by_id={formula.id: formula for formula in snapshot.revision.formulas},
binding_to_casillas=casillas_by_binding(snapshot.revision),
fold_binding_ids=frozenset(
binding.id for binding, _ in relation_prefill_bindings_for_period(snapshot.revision)
),
relation_channels={
binding.id: relation_consumption_channels(binding.id, consumption_index)
for binding, _ in relation_prefill_bindings_for_period(snapshot.revision)
},
persisted_decimal_bindings=_persisted_decimal_bindings(snapshot=snapshot, revision=revision),
persisted_binding_ids=frozenset(() if revision is None else revision.binding_overrides),
estados_casillas_oficiales=estados_casillas_oficiales,
official_references=_official_references(
snapshot,
estados_casillas_oficiales,
source_payloads=source_payloads,
),
blocking_findings=blocking_findings,
)
def _review_casillas(
*,
snapshot: RegistrySnapshot,
revision: CalculationRevision | None,
blocking_findings: tuple[ModeloVerificationFinding, ...],
operation: PinnedAuthorityOperation,
) -> tuple[ModeloWorkReviewCasilla, ...]:
context = _review_row_context(
snapshot=snapshot,
revision=revision,
blocking_findings=blocking_findings,
operation=operation,
)
return tuple(_review_casilla(casilla, context) for casilla in snapshot.revision.casillas)
def _work_progress(
*,
snapshot: RegistrySnapshot,
rows: tuple[ModeloWorkReviewCasilla, ...],
verification: VerificationReport | None,
) -> ModeloWorkProgress:
manifest = snapshot.revision.completeness_manifest
if manifest is None:
return ModeloWorkProgress(state=ModeloWorkProgressState.UNDEFINED)
target_ids = frozenset(item.casilla_id for item in manifest.casillas)
materialised_count = sum(
row.casilla_id in target_ids and row.realised_kind is not ModeloValueKind.EMPTY for row in rows
)
if verification is not None and verification.completeness_status is VerificationCompletenessStatus.BLOCKED:
state = ModeloWorkProgressState.BLOCKED
elif (
verification is not None
and verification.completeness_status is VerificationCompletenessStatus.COMPLETE
and materialised_count == len(target_ids)
):
state = ModeloWorkProgressState.COMPLETE
else:
state = ModeloWorkProgressState.IN_PROGRESS
return ModeloWorkProgress(
state=state,
materialised_count=materialised_count,
target_count=len(target_ids),
denominator=ModeloWorkProgressDenominator(
registry_revision_id=snapshot.revision.id,
source_ref=manifest.source_ref,
),
)
[docs]
def assemble_modelo_work_review(
bucket_id: BucketId,
modelo: ModeloCode,
filing_year: int,
period: Period,
*,
operation: PinnedAuthorityOperation,
work_unit_repository: WorkUnitCatalogueRepositoryProtocol,
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol,
verification_repository: VerificationReportCatalogueRepositoryProtocol,
) -> ModeloWorkReview:
selected_revision = operation.revision_for_context(
modelo,
filing_year=filing_year,
period=period.registry_token,
)
snapshot = operation.snapshot(
modelo,
filing_year=filing_year,
period=period.registry_token,
grade=selected_revision.effective_authority_grade,
)
work_units = work_unit_repository.load()
calculation_repo = calculation_repository
verification_repo = verification_repository
work_unit = _work_unit_for_target(
bucket_id=bucket_id,
modelo=modelo,
filing_year=filing_year,
period=period,
registry_revision_id=snapshot.revision.id,
catalogue=work_units,
)
revision = _current_revision(work_unit, calculation_repo, operation=operation)
verification = _latest_verification(revision, verification_repo, operation=operation)
findings = () if verification is None else verification.findings
blocking_findings = tuple(
finding for finding in findings if finding.severity is ModeloVerificationFindingSeverity.BLOCKING
)
blockers = tuple(_blocker_ref(finding) for finding in blocking_findings)
rows = _review_casillas(
snapshot=snapshot,
revision=revision,
blocking_findings=blocking_findings,
operation=operation,
)
return ModeloWorkReview(
bucket_id=bucket_id,
modelo=modelo,
filing_year=filing_year,
period=period,
registry_revision_id=snapshot.revision.id,
work_unit_id=work_unit.work_unit_id,
calculation_revision_id=None if revision is None else revision.calculation_revision_id,
lifecycle_state=None if revision is None else revision.state,
verification_outcome=None if verification is None else verification.completeness_status,
progress=_work_progress(snapshot=snapshot, rows=rows, verification=verification),
casillas=rows,
findings=findings,
blockers=blockers,
row_source_fingerprints=revision_row_source_fingerprints_for_review(revision),
)