"""Opaque OSS / IOSS regime tokens projected from the facts authority."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import date
from types import MappingProxyType
from typing import Final
from ...core.time.clock import today_madrid
from ..calculations.registry.errors import RegistryValidationError
from ..calculations.registry.facts.resolution import (
MappingFactQuery,
ResolvedMappingFact,
required_mapping_entry,
unique_mapping_tokens,
)
from ..calculations.registry.governed_fact_scope import GovernedFactSource, governed_facts_in_scope
from ..calculations.registry.schema_base import DateAxis
_ENTRY_SUBJECT: Final = "OSS/IOSS regime mapping"
_UNIQUE_TOKENS_REQUIREMENT: Final = "must declare unique tokens"
_FACT_ID = "modelo-369-exterior-oss-projection-catalogue"
_ORDER_KEY = "regime.order"
_COMMON_SEMANTICS_KEY = "regime.common_semantics"
_SELECTOR_KEY = "regime.selector"
[docs]
class OssIossRegime(str):
"""Opaque registry-projected OSS / IOSS regime token."""
__slots__ = ()
@classmethod
def __get_pydantic_core_schema__(cls, _source_type: object, _handler: object) -> object:
"""Expose the opaque token as a string to Pydantic without a catalogue."""
from pydantic_core import core_schema
return core_schema.no_info_after_validator_function(cls, core_schema.str_schema())
@property
def value(self) -> str:
"""Return the opaque token for string-oriented serialization."""
return str(self)
[docs]
@dataclass(frozen=True, slots=True)
class OssIossRegimeDefinition:
"""One registry-declared OSS / IOSS regime and its legal semantics."""
token: OssIossRegime
description: str
articles: str
filing_cadence: str
transaction_kinds: tuple[str, ...]
intrinsic_value_ceiling_eur: str | None = None
[docs]
@dataclass(frozen=True, slots=True)
class OssIossRegimeCatalogue:
"""Typed projection of the dated Modelo 369 OSS/IOSS mapping fact."""
definitions: tuple[OssIossRegimeDefinition, ...]
common_semantics: str
selector_key: str
@property
def all_regimes(self) -> frozenset[OssIossRegime]:
"""Return every regime token declared by the registry."""
return frozenset(definition.token for definition in self.definitions)
[docs]
def require(self, value: object) -> OssIossRegime:
"""Validate one opaque regime token against this projection."""
if isinstance(value, OssIossRegime):
token = value
elif isinstance(value, str):
token = OssIossRegime(value.strip())
else:
raise RegistryValidationError("OSS/IOSS regime must be a string token")
if not str(token):
raise RegistryValidationError("OSS/IOSS regime token must not be blank")
if token not in self.all_regimes:
raise RegistryValidationError(f"OSS/IOSS regime {str(token)!r} is not registry-declared")
return token
[docs]
def definition(self, value: object) -> OssIossRegimeDefinition:
"""Return the registry definition for one regime token."""
token = self.require(value)
for definition in self.definitions:
if definition.token == token:
return definition
raise RegistryValidationError(f"OSS/IOSS regime {str(token)!r} has no registry definition")
[docs]
def transaction_kinds_for(self, value: object) -> frozenset[str]:
"""Return transaction-kind values admitted by one registry regime."""
return frozenset(self.definition(value).transaction_kinds)
[docs]
def regime_for_transaction_kind(self, transaction_kind: str) -> OssIossRegime:
"""Return the unique registry regime admitting a transaction kind."""
matches = tuple(
definition.token for definition in self.definitions if transaction_kind in definition.transaction_kinds
)
if len(matches) != 1:
raise RegistryValidationError(
f"transaction kind {transaction_kind!r} must map to exactly one OSS/IOSS regime",
)
return matches[0]
def _mapping_entries(resolved: ResolvedMappingFact) -> Mapping[str, str]:
"""Narrow a resolved mapping payload to a unique string-to-string map."""
entries: dict[str, str] = {}
for entry in resolved.payload.entries:
if not isinstance(entry.key, str) or not isinstance(entry.value, str):
raise RegistryValidationError("OSS/IOSS regime entries must be string-to-string")
if entry.key in entries:
raise RegistryValidationError(f"duplicate OSS/IOSS regime key {entry.key!r}")
entries[entry.key] = entry.value
return MappingProxyType(entries)
[docs]
def resolve_oss_ioss_regime_catalogue(
*,
effective_date: date | None = None,
authority: GovernedFactSource | None = None,
) -> OssIossRegimeCatalogue:
"""Resolve the complete OSS / IOSS regime catalogue through facts authority."""
authority = authority or governed_facts_in_scope()
if authority is None:
raise RegistryValidationError(
"OSS/IOSS regime catalogue requires an explicit authority operation or scope",
)
resolved = authority.resolve_governed_fact(
MappingFactQuery(
fact_id=_FACT_ID,
date_axis=DateAxis.TRANSACTION_DATE,
effective_date=effective_date or today_madrid(),
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise RegistryValidationError("Modelo 369 OSS/IOSS projection must resolve as a mapping fact")
entries = _mapping_entries(resolved)
ordered_tokens = unique_mapping_tokens(
entries, _ORDER_KEY, subject=_ENTRY_SUBJECT, requirement=_UNIQUE_TOKENS_REQUIREMENT
)
definitions: list[OssIossRegimeDefinition] = []
for token in ordered_tokens:
prefix = f"regime.{token}"
ceiling_key = f"{prefix}.intrinsic_value_ceiling_eur"
definitions.append(
OssIossRegimeDefinition(
token=OssIossRegime(token),
description=required_mapping_entry(entries, f"{prefix}.description", subject=_ENTRY_SUBJECT),
articles=required_mapping_entry(entries, f"{prefix}.articles", subject=_ENTRY_SUBJECT),
filing_cadence=required_mapping_entry(entries, f"{prefix}.filing_cadence", subject=_ENTRY_SUBJECT),
transaction_kinds=unique_mapping_tokens(
entries,
f"{prefix}.transaction_kinds",
subject=_ENTRY_SUBJECT,
requirement=_UNIQUE_TOKENS_REQUIREMENT,
),
intrinsic_value_ceiling_eur=entries.get(ceiling_key),
),
)
return OssIossRegimeCatalogue(
definitions=tuple(definitions),
common_semantics=required_mapping_entry(entries, _COMMON_SEMANTICS_KEY, subject=_ENTRY_SUBJECT),
selector_key=required_mapping_entry(entries, _SELECTOR_KEY, subject=_ENTRY_SUBJECT),
)
[docs]
def require_oss_ioss_regime(
value: object,
*,
effective_date: date | None = None,
authority: GovernedFactSource | None = None,
) -> OssIossRegime:
"""Return one registry-declared OSS/IOSS token or refuse it."""
return resolve_oss_ioss_regime_catalogue(
effective_date=effective_date,
authority=authority,
).require(value)
__all__ = [
"OssIossRegime",
"OssIossRegimeCatalogue",
"OssIossRegimeDefinition",
"require_oss_ioss_regime",
"resolve_oss_ioss_regime_catalogue",
]