"""Engine-ready AST generation service."""
from __future__ import annotations
import json
import logging
import re
import zlib
from dataclasses import dataclass, field
from datetime import date as date_cls
from typing import (
TYPE_CHECKING,
Any,
Dict,
Iterable,
List,
Optional,
Set,
Tuple,
Union,
cast,
)
from dpmcore.dpm_xl.utils.tokens import (
SEVERITY_WARNING,
VALID_SEVERITIES,
)
from dpmcore.errors import InternalError, SemanticError
from dpmcore.services._parameters import merge_parameters
from dpmcore.services._precondition_codes import (
extract_precondition_codes as _extract_precondition_codes,
)
from dpmcore.services.semantic import ParameterInfo, SemanticService
from dpmcore.services.syntax import SyntaxService
if TYPE_CHECKING:
from sqlalchemy.orm import Session
from dpmcore.services.calculations_export import CalculationsExport
from dpmcore.services.scope_calculator import (
ScopeCalculatorService,
ScopeResult,
)
logger = logging.getLogger(__name__)
_TABLE_CODE_NORMALIZER = re.compile(r"^([A-Z]+)_(\d+)_(\d+)$")
_DEFAULT_FROM_DATE = "2001-01-01"
_DEFAULT_NAMESPACE = "default_module"
_DATA_FIELDS_TO_STRIP = ("data_type", "cell_code", "table_code", "table_vid")
@dataclass(frozen=True)
class _OperandRefs:
"""What a single operation's operands reference.
``tables`` are the table codes of its ``VarID`` nodes; ``variables``
maps every operand datapoint id to its scalar type code, across all
modules the operation spans (home module included).
"""
tables: set[str] = field(default_factory=set)
variables: Dict[str, str] = field(default_factory=dict)
@dataclass(frozen=True)
class _PreparedExpression:
"""One expression readied for the script, or the reason it is not.
``error`` is set when the expression cannot become a declarable
operation — it failed semantic validation, it carries a shift whose
reference period cannot be declared (#326), or its precondition is
one the engine cannot evaluate (#338). All are reported the same
way, through ``failed_operations``, and all must be caught before
the caller accumulates anything for the operation: a skip afterwards
would leave its tables, parameters and operation entry in the
script.
"""
result: Any = None
ast: Any = None
ts: Dict[str, List[str]] = field(default_factory=dict)
error: Optional[str] = None
def _normalize_variable_code(code: str) -> str:
"""Normalise ``F_44_04`` → ``F_44.04`` (matches pydpm)."""
m = _TABLE_CODE_NORMALIZER.match(code)
if m:
return f"{m.group(1)}_{m.group(2)}.{m.group(3)}"
return code
def _sweep_order(
sort_orders: Dict[int, int],
row: Tuple[Optional[str], Optional[str], Optional[int]],
) -> Tuple[str, int]:
"""Sort key ordering the sweep by ``(code, start release date)``.
Applied in Python rather than SQL so ghost-fallback versions (#372),
which the release-filtered query cannot return, sort in with the
rows it did return. The second component is the start release's
date-based sort order, never the opaque ``start_release_id`` itself,
which is non-monotonic from DPM 4.2.1 onwards; a missing or orphan
start sorts first.
"""
code, _version_number, start_release_id = row
order = (
None if start_release_id is None else sort_orders.get(start_release_id)
)
return (code or "", -1 if order is None else order)
def _format_date(value: Any, fallback: Optional[str] = None) -> Optional[str]:
"""Format a ``date`` / ``datetime`` / string as ``YYYY-MM-DD``."""
if value is None:
return fallback
if isinstance(value, str):
return value
if isinstance(value, date_cls):
return value.strftime("%Y-%m-%d")
return str(value)
[docs]
class ASTGeneratorService:
"""Generate engine-ready validation scripts from DPM-XL expressions.
Args:
session: An open SQLAlchemy session (required for ``script``).
"""
[docs]
def __init__(self, session: Optional["Session"] = None) -> None:
"""Build the service, optionally bound to a SQLAlchemy ``session``."""
self.session = session
self._semantic: Optional[SemanticService] = None
self._scope_calc: Optional["ScopeCalculatorService"] = None
self._syntax = SyntaxService()
self._ghost_fallback_cache: Dict[int, Dict[int, List[int]]] = {}
if session is not None:
from dpmcore.services.scope_calculator import (
ScopeCalculatorService,
)
self._semantic = SemanticService(session)
self._scope_calc = ScopeCalculatorService(session)
[docs]
def script(
self,
expressions: List[Tuple[str, str]],
module_code: str,
module_version: str,
preconditions: Optional[
List[Union[Tuple[str, List[str]], Dict[str, Any]]]
] = None,
severity: Optional[str] = None,
severities: Optional[Dict[str, str]] = None,
release: Optional[str] = None,
) -> Dict[str, Any]:
"""Generate an engine-ready validations script.
Args:
expressions: ``[(expression, validation_code), ...]``.
module_code: Code of the primary module (e.g. ``"COREP_Con"``).
module_version: Version of the primary module
(e.g. ``"2.0.1"``).
preconditions: Optional list of precondition specs. Each
entry is either a tuple
``(precondition_expression, [validation_codes])`` or a
dict with keys ``expression`` and
``affected_operations`` (optional ``code`` and
``version_id`` are also accepted). A precondition can
guard many validation codes; a validation may have no
precondition. Codes that end up in
``failed_operations`` are stripped from the emitted
``affected_operations``, and a precondition left
gating nothing is dropped entirely.
severity: Optional global default severity tag
(``"error"``, ``"warning"``, ``"info"``). Defaults to
``"warning"``.
severities: Optional per-validation override map
``{validation_code: severity}``. Resolution per
validation is ``severities.get(code, severity_global)``.
Unknown codes (not present in ``expressions``) raise
``ValueError``.
release: Optional release code. When omitted, resolved to
the latest release whose window contains the requested
``ModuleVersion``. The resolved release is surfaced in
the ``dpm_release`` block and used for every
downstream DB filter.
Returns:
A dict with keys ``success`` (bool), ``enriched_ast`` (the
namespaced dict, or ``None`` on failure), ``error`` (str or
``None``), and ``failed_operations`` (a
``{validation_code: error_message}`` map of expressions
skipped due to semantic errors, or because their
precondition cannot be evaluated by the engine). The
namespaced dict mirrors the shape pydpm's
``generate_validations_script`` produces.
"""
session = self.session
if (
self._semantic is None
or self._scope_calc is None
or session is None
):
return {
"success": False,
"enriched_ast": None,
"error": "No database session — cannot generate script.",
"failed_operations": {},
}
try:
from dpmcore.dpm_xl.utils.serialization import serialize_ast
mv, release_row = self._resolve_release(
module_code, module_version, release
)
primary_module_vid: int = mv.module_vid
release_id: int = release_row.release_id
validation_codes = [code for _, code in expressions]
resolved_severities = self._resolve_severities(
severity, severities, validation_codes
)
try:
code_to_precondition_items = self._build_precondition_index(
preconditions or []
)
except ValueError as exc:
return {
"success": False,
"enriched_ast": None,
"error": str(exc),
"failed_operations": {},
}
# Operations whose gate the engine cannot evaluate leave the
# script here, before any accumulation, and are reported in
# ``failed_operations`` (see ``_unsupported_gate_operations``).
gate_failures = self._unsupported_gate_operations(
preconditions or []
)
from_reference_date = _format_date(
mv.from_reference_date, fallback=_DEFAULT_FROM_DATE
)
operations: Dict[str, Dict[str, Any]] = {}
failed_operations: Dict[str, str] = {}
scope_pairs: List[
Tuple[
Tuple[str, str],
"ScopeResult",
Dict[str, List[str]],
_OperandRefs,
]
] = []
referenced_table_codes: set[str] = set()
referenced_parameters: Dict[str, ParameterInfo] = {}
for item in expressions:
expr, code = item[0], item[1]
# Semantic validation plus the reference periods the
# expression needs; either can reject it, as can a gate
# the engine cannot evaluate. The _accumulate_parameters
# call below is complementary to the scope check inside —
# it catches conflicts between two expressions in this
# same script.
prepared = self._prepare_expression(
self._semantic,
expr,
release_id,
gate_failure=gate_failures.get(code),
)
if prepared.error is not None:
failed_operations[code] = prepared.error
continue
result, ast, ts = prepared.result, prepared.ast, prepared.ts
# serialize_ast() is typed for any node (a leaf can
# serialise to a bare scalar), but `ast` is always a
# root expression here, which the wire format always
# renders as an object.
ast_dict = cast(Dict[str, Any], serialize_ast(ast))
root_operator_id = self._resolve_root_operator_id(ast, session)
scope_error = self._process_operation(
item=item,
ast_dict=ast_dict,
parameters=result.parameters,
root_operator_id=root_operator_id,
ts=ts,
release_id=release_id,
resolved_severities=resolved_severities,
from_submission_date=from_reference_date,
code_to_precondition_items=code_to_precondition_items,
operations=operations,
scope_pairs=scope_pairs,
referenced_table_codes=referenced_table_codes,
referenced_parameters=referenced_parameters,
)
if scope_error is not None:
# A script whose dependency block is silently missing
# is structurally valid but semantically wrong (#122),
# so a scope failure fails the whole generation.
return {
"success": False,
"enriched_ast": None,
"error": scope_error,
"failed_operations": failed_operations,
}
return self._assemble_script(
mv=mv,
release_row=release_row,
primary_module_vid=primary_module_vid,
release_id=release_id,
preconditions=preconditions,
operations=operations,
failed_operations=failed_operations,
scope_pairs=scope_pairs,
referenced_table_codes=referenced_table_codes,
referenced_parameters=referenced_parameters,
)
except ValueError as exc:
return {
"success": False,
"enriched_ast": None,
"error": str(exc),
"failed_operations": {},
}
except Exception as exc:
return {
"success": False,
"enriched_ast": None,
"error": str(exc),
"failed_operations": {},
}
def _process_operation(
self,
*,
item: Tuple[str, str],
ast_dict: Dict[str, Any],
parameters: List[ParameterInfo],
root_operator_id: int,
ts: Dict[str, List[str]],
release_id: int,
resolved_severities: Dict[str, str],
from_submission_date: Optional[str],
code_to_precondition_items: Dict[str, List[Any]],
operations: Dict[str, Dict[str, Any]],
scope_pairs: List[
Tuple[
Tuple[str, str],
"ScopeResult",
Dict[str, List[str]],
_OperandRefs,
]
],
referenced_table_codes: set[str],
referenced_parameters: Dict[str, ParameterInfo],
operation_vid: Optional[int] = None,
scope_result: Optional["ScopeResult"] = None,
) -> Optional[str]:
"""Fold one operation's already-resolved AST into the script.
Adds to the script's accumulators (``operations``,
``scope_pairs``, ``referenced_table_codes``,
``referenced_parameters`` — all mutated in place).
Independent of how ``ast_dict``/``parameters``/``root_operator_id``
were obtained — re-parsed from text, or reshaped from the
persisted ``OperationNode`` tree — this is the same regardless.
``operation_vid`` is the real ``OperationVID`` this operation was
discovered under, when the caller has one (:meth:`script_from_db`
always does, from :meth:`_discover_module_validations`). ``None``
for :meth:`script`'s caller-supplied ``expressions`` — those
codes may not even be real DB operations (e.g. an
``--expressions`` file) — in which case
:meth:`_build_operation_entry` falls back to its deterministic
hash. See dpmcore#364.
``scope_result`` lets the caller supply an already-computed
``ScopeResult`` (:meth:`script_from_db` passes one built
straight from the persisted ``OperationScope`` rows, when
available) instead of having ``expr`` re-parsed here. ``None``
falls back to :meth:`ScopeCalculatorService.calculate_from_expression`
— :meth:`script`'s caller-supplied ``expressions`` always take
this path, since they have no ``operation_vid`` to read a
persisted scope from.
Returns the scope-calculation error message on failure (callers
decide whether that aborts the whole script or just excludes
this operation), or ``None`` on success. Nothing is added to
``operations``/``referenced_table_codes``/``referenced_parameters``/
``scope_pairs`` until the scope calculation has actually
succeeded — a failed operation must leave no trace in them,
since a script whose dependency block is silently missing is
structurally valid but semantically wrong (#122).
"""
expr, code = item[0], item[1]
if self._scope_calc is None:
raise InternalError(
"ScopeCalculatorService not initialised",
"_process_operation requires a live DB session.",
)
sr = scope_result or self._scope_calc.calculate_from_expression(
expression=expr,
release_id=release_id,
precondition_items=code_to_precondition_items.get(code, []),
)
if sr.has_error:
return (
f"Scope calculation failed for operation "
f"'{code}': {sr.error_message}"
)
# Operand refs come off the *raw* serialisation: cleaning
# strips the ``data_type`` each datapoint is typed by.
op_refs = _OperandRefs(
tables=self._extract_referenced_tables(ast_dict),
variables=self._extract_operand_datapoints(ast_dict),
)
self._clean_ast_data_entries(ast_dict)
referenced_table_codes.update(op_refs.tables)
self._accumulate_parameters(referenced_parameters, parameters)
operations[code] = self._build_operation_entry(
expression=expr,
code=code,
ast_dict=ast_dict,
severity=resolved_severities[code],
reference_date=from_submission_date,
root_operator_id=root_operator_id,
operation_vid=operation_vid,
)
scope_pairs.append((item, sr, ts, op_refs))
return None
def _assemble_script(
self,
*,
mv: Any,
release_row: Any,
primary_module_vid: int,
release_id: int,
preconditions: Optional[
List[Union[Tuple[str, List[str]], Dict[str, Any]]]
],
operations: Dict[str, Dict[str, Any]],
failed_operations: Dict[str, str],
scope_pairs: List[
Tuple[
Tuple[str, str],
"ScopeResult",
Dict[str, List[str]],
_OperandRefs,
]
],
referenced_table_codes: set[str],
referenced_parameters: Dict[str, ParameterInfo],
) -> Dict[str, Any]:
"""Assemble the final script dict from already-resolved operations.
Everything here is independent of how each operation's AST was
obtained (re-parsed from text, or reshaped from the persisted
``OperationNode`` tree) — it only reads what the per-operation
loop already produced. Shared by :meth:`script` (the
text-reparse path) and the DB-native path.
This is a verbatim extraction of what used to be the tail of
:meth:`script`'s body — no behaviour changed, only where the
code lives.
"""
if self._scope_calc is None or self.session is None:
raise InternalError(
"ScopeCalculatorService not initialised",
"_assemble_script requires a live DB session.",
)
primary_tables_full = self._scope_calc._get_module_tables(
primary_module_vid,
release_id=release_id,
ghost_vids=self._release_ghost_fallbacks(
self.session, release_id
).get(primary_module_vid, []),
)
# Seed from every module-composition table that carries
# variables — i.e. the non-abstract tables; abstract templates
# have no cells, and the engine schema forbids an empty
# variables map — then union in anything the expressions
# reference. MDPM lists all such tables even when no validation
# touches them (#158). The union keeps this additive: a
# referenced table is never dropped.
seed_codes = {
code
for code, data in primary_tables_full.items()
if data.get("variables")
}
seed_codes |= {
code
for code in referenced_table_codes
if code in primary_tables_full
}
tables_block: Dict[str, Any] = {
code: primary_tables_full[code] for code in sorted(seed_codes)
}
variables_block: Dict[str, str] = {}
for tbl in tables_block.values():
variables_block.update(tbl.get("variables", {}))
# ``emitted_operations`` is the script's own ``operations``,
# not the harvested list: anything that landed in
# ``failed_operations`` must not be left gated (#355).
preconditions_block, precondition_variables_block = (
self._build_preconditions_block(
preconditions or [],
release_id=release_id,
referenced_parameters=referenced_parameters,
emitted_operations=set(operations),
)
)
# Runtime-binding contract: the declared type of every parameter
# this script needs, keyed by code. This is the scope-wide
# invariant. ``is_set`` is recoverable from the ``set-`` prefix
# and ``default`` is a per-reference fallback the engine binds
# per scope, so neither belongs in this registry. Built after
# the gates so a parameter that only appears in a gate is
# declared alongside the ones the expressions reference.
parameters_block: Dict[str, str] = {
prm_code: prm.declared_type
for prm_code, prm in sorted(referenced_parameters.items())
}
dependency_info = self._build_dependency_info(
scope_pairs=scope_pairs,
primary_module_vid=primary_module_vid,
release_id=release_id,
)
dep_information: Dict[str, Any]
dep_modules: Dict[str, Any]
if dependency_info is not None:
dep_information = dependency_info["dependency_information"]
dep_modules = dependency_info["dependency_modules"]
else:
dep_information = {
"intra_instance_validations": [],
"cross_instance_dependencies": [],
"alternative_dependencies": [],
}
dep_modules = {}
namespace = (
self._scope_calc._get_module_uri(
module_vid=primary_module_vid,
mv=mv,
)
or _DEFAULT_NAMESPACE
)
module_info = self._build_module_info(mv)
ns_block: Dict[str, Any] = {
**module_info,
"dpm_release": self._build_release_info(release_row),
"dates": self._build_dates(mv),
"operations": operations,
"variables": variables_block,
"tables": tables_block,
"parameters": parameters_block,
"preconditions": preconditions_block,
"precondition_variables": precondition_variables_block,
"dependency_information": dep_information,
"dependency_modules": dep_modules,
}
return {
"success": True,
"enriched_ast": {namespace: ns_block},
"error": None,
"failed_operations": failed_operations,
}
def _resolve_db_operation_ast(
self,
*,
session: Any,
code: str,
expr: str,
operation_vid: Optional[int],
release_id: int,
) -> Tuple[
Optional[Dict[str, Any]],
List[ParameterInfo],
Dict[str, List[str]],
Optional[int],
Optional[str],
]:
"""Resolve one operation's ast_dict/parameters/ts/root_operator_id.
Tries the persisted ``OperationNode`` tree first
(:func:`db_ast.build_ast_from_db`), falling back to re-parsing
*expr* when that isn't possible. Used by :meth:`script_from_db`,
split out to keep its own branching within the complexity limit.
Returns ``(ast_dict, parameters, ts, root_operator_id, error)``
— a non-``None`` ``error`` means the operation failed and the
other fields should be ignored.
"""
if self._semantic is None:
raise InternalError(
"SemanticService not initialised",
"_resolve_db_operation_ast requires a live DB session.",
)
from dpmcore.dpm_xl.utils.db_ast import (
UnsupportedDbAst,
build_ast_from_db,
serialize_built_ast,
)
from dpmcore.dpm_xl.utils.serialization import serialize_ast
ast_dict: Optional[Dict[str, Any]] = None
built_ast: Optional[Any] = None
root_operator_id: Optional[int] = None
if operation_vid is not None:
try:
built_ast, root_operator_id = build_ast_from_db(
session, operation_vid, release_id
)
ast_dict = serialize_built_ast(built_ast)
except UnsupportedDbAst:
ast_dict = None
built_ast = None
if ast_dict is None:
# Same resolution script() uses for the text-reparse path —
# duplicated rather than shared, it's a handful of lines
# (see dpmcore#364).
prepared = self._prepare_expression(
self._semantic, expr, release_id, gate_failure=None
)
if prepared.error is not None:
return None, [], {}, None, prepared.error
result, ast, ts = prepared.result, prepared.ast, prepared.ts
# Same reasoning as script()'s own serialize_ast() call: a
# root expression always renders as an object.
ast_dict = cast(Dict[str, Any], serialize_ast(ast))
parameters = result.parameters
root_operator_id = self._resolve_root_operator_id(ast, session)
return ast_dict, parameters, ts, root_operator_id, None
# Same source script() uses — the AST itself — not whether it
# came from the DB or a re-parse: a parameter is self-contained
# in its own node (code/type/default are declared inline, never
# resolved from the DB), so it is read straight off the tree
# db_ast.py already built.
found_parameters: Dict[str, ParameterInfo] = {}
self._accumulate_ast_parameters(found_parameters, ast_dict)
parameters = list(found_parameters.values())
# Same reasoning as _prepare_expression's own call: a shift
# whose period cannot be declared must reject this one
# operation, not the whole module.
try:
ts = self._extract_time_shifts(built_ast)
except SemanticError as exc:
return None, [], {}, None, str(exc)
return ast_dict, parameters, ts, root_operator_id, None
def _process_db_expression_item(
self,
*,
item: Tuple[str, str],
session: Any,
gate_failures: Dict[str, str],
operation_vids: Dict[str, int],
primary_module_vid: int,
release_id: int,
resolved_severities: Dict[str, str],
from_submission_dates: Dict[str, str],
code_to_precondition_items: Dict[str, List[Any]],
operations: Dict[str, Dict[str, Any]],
scope_pairs: List[
Tuple[
Tuple[str, str],
"ScopeResult",
Dict[str, List[str]],
_OperandRefs,
]
],
referenced_table_codes: set[str],
referenced_parameters: Dict[str, ParameterInfo],
) -> Optional[str]:
"""Resolve and fold one :meth:`script_from_db` expression item.
Returns the failure reason to record in ``failed_operations``
for this item's code, or ``None`` on success. Split out to keep
:meth:`script_from_db`'s own branching within the complexity
limit.
"""
from dpmcore.services.scope_calculator import UnsupportedDbScope
expr, code = item[0], item[1]
# Checked once, up front, for both resolution paths below — a
# precondition the engine cannot evaluate must reject the
# operation regardless of whether its own AST would otherwise
# resolve from the DB or from re-parsed text.
gate_failure = gate_failures.get(code)
if gate_failure is not None:
return gate_failure
operation_vid = operation_vids.get(code)
(
ast_dict,
parameters,
ts,
root_operator_id,
resolve_error,
) = self._resolve_db_operation_ast(
session=session,
code=code,
expr=expr,
operation_vid=operation_vid,
release_id=release_id,
)
if resolve_error is not None:
return resolve_error
# A None error means _resolve_db_operation_ast populated every
# other field — its own contract, not re-checked field by field.
if ast_dict is None or root_operator_id is None:
raise InternalError(
"Inconsistent _resolve_db_operation_ast result",
f"No error for {code!r} but ast_dict/root_operator_id "
"is still None.",
)
scope_result: Optional["ScopeResult"] = None
if operation_vid is not None:
if self._scope_calc is None:
raise InternalError(
"ScopeCalculatorService not initialised",
"_process_db_expression_item requires a live DB session.",
)
# Same rationale as the AST: the scope is already persisted,
# so re-deriving it via re-parse+re-validate only risks a
# spurious rejection (e.g. grey cells).
try:
scope_result = self._scope_calc.build_scope_result_from_db(
operation_vid,
primary_module_vid,
release_id,
)
except UnsupportedDbScope:
scope_result = None
return self._process_operation(
item=item,
ast_dict=ast_dict,
parameters=parameters,
root_operator_id=root_operator_id,
ts=ts,
release_id=release_id,
resolved_severities=resolved_severities,
from_submission_date=from_submission_dates.get(code),
code_to_precondition_items=code_to_precondition_items,
operations=operations,
scope_pairs=scope_pairs,
referenced_table_codes=referenced_table_codes,
referenced_parameters=referenced_parameters,
operation_vid=operation_vid,
scope_result=scope_result,
)
[docs]
def script_from_db(
self,
expressions: List[Tuple[str, str]],
operation_vids: Dict[str, int],
mv: Any,
release_row: Any,
preconditions: Optional[
List[Union[Tuple[str, List[str]], Dict[str, Any]]]
] = None,
severity: Optional[str] = None,
severities: Optional[Dict[str, str]] = None,
from_submission_dates: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
"""Generate a DB-discovered module version's validations script.
For each operation, its persisted ``OperationNode`` tree is tried
first (:func:`db_ast.build_ast_from_db`) — falling back to
re-parsing its expression text only when that isn't possible
(an unsupported construct, or the reshaper for it not being
implemented yet — see dpmcore#364). Shares
:meth:`_process_operation`/:meth:`_assemble_script` with
:meth:`script`; the only thing that differs between the two is
how each operation's ``ast_dict``/``parameters``/``ts``/
``root_operator_id`` are obtained.
Args:
expressions: same as :meth:`script`.
operation_vids: ``{code: OperationVID}``, from
:meth:`_discover_module_validations` — which operations
have a persisted AST to try first.
mv: already resolved by the caller (:meth:`script_for_module`),
so this doesn't re-resolve it.
release_row: already resolved by the caller
(:meth:`script_for_module`), so this doesn't re-resolve
it.
preconditions: same as :meth:`script`.
severity: same as :meth:`script`.
severities: same as :meth:`script`.
from_submission_dates: ``{code: FromSubmissionDate}``, from
:meth:`_discover_module_validations` — the real
per-operation submission date. A code missing here (the
column was null) falls back to
:meth:`_build_operation_entry`'s own default, same as
:meth:`script`'s caller-supplied expressions.
Returns:
Same shape as :meth:`script`. Unlike :meth:`script`, a
per-operation scope-calculation error does not fail the
whole result: it is recorded in ``failed_operations`` and
that operation is left out, the same as a semantic error or
an ungateable precondition. :meth:`script` still aborts on
a scope error (#122) — this method discovers every active
validation for the module itself, so one bad operation
failing the whole discovered set would be a worse surprise
here than for a caller-supplied ``expressions`` list.
"""
session = self.session
if (
self._semantic is None
or self._scope_calc is None
or session is None
):
return {
"success": False,
"enriched_ast": None,
"error": "No database session — cannot generate script.",
"failed_operations": {},
}
try:
primary_module_vid: int = mv.module_vid
release_id: int = release_row.release_id
validation_codes = [code for _, code in expressions]
resolved_severities = self._resolve_severities(
severity, severities, validation_codes
)
try:
code_to_precondition_items = self._build_precondition_index(
preconditions or []
)
except ValueError as exc:
return {
"success": False,
"enriched_ast": None,
"error": str(exc),
"failed_operations": {},
}
gate_failures = self._unsupported_gate_operations(
preconditions or []
)
from_submission_dates = from_submission_dates or {}
operations: Dict[str, Dict[str, Any]] = {}
failed_operations: Dict[str, str] = {}
scope_pairs: List[
Tuple[
Tuple[str, str],
"ScopeResult",
Dict[str, List[str]],
_OperandRefs,
]
] = []
referenced_table_codes: set[str] = set()
referenced_parameters: Dict[str, ParameterInfo] = {}
for item in expressions:
code = item[1]
# Unlike script()'s text-reparse path (#122), a failure
# here excludes just this operation instead of aborting
# the whole module: export-script discovers every active
# validation for the module by itself, so one bad
# operation would otherwise take down a script the
# caller never explicitly asked for by name.
error = self._process_db_expression_item(
item=item,
session=session,
gate_failures=gate_failures,
operation_vids=operation_vids,
primary_module_vid=primary_module_vid,
release_id=release_id,
resolved_severities=resolved_severities,
from_submission_dates=from_submission_dates,
code_to_precondition_items=code_to_precondition_items,
operations=operations,
scope_pairs=scope_pairs,
referenced_table_codes=referenced_table_codes,
referenced_parameters=referenced_parameters,
)
if error is not None:
failed_operations[code] = error
return self._assemble_script(
mv=mv,
release_row=release_row,
primary_module_vid=primary_module_vid,
release_id=release_id,
preconditions=preconditions,
operations=operations,
failed_operations=failed_operations,
scope_pairs=scope_pairs,
referenced_table_codes=referenced_table_codes,
referenced_parameters=referenced_parameters,
)
except ValueError as exc:
return {
"success": False,
"enriched_ast": None,
"error": str(exc),
"failed_operations": {},
}
except Exception as exc:
return {
"success": False,
"enriched_ast": None,
"error": str(exc),
"failed_operations": {},
}
[docs]
def script_for_module(
self,
module_code: str,
module_version: str,
release: Optional[str] = None,
) -> Dict[str, Any]:
"""Auto-discover a module version's active validations and script them.
No ``expressions``, ``preconditions`` or ``severities`` need to be
supplied — they are looked up from ``OperationScope`` /
``OperationScopeComposition`` / ``OperationVersion`` for this module
version, then handed to :meth:`script_from_db`.
Args:
module_code: Code of the module (e.g. ``"COREP_Con"``).
module_version: Version of the module (e.g. ``"2.0.1"``).
release: Optional release code; resolved the same way as in
:meth:`script`.
Returns:
Same shape as :meth:`script`.
"""
if self.session is None:
return {
"success": False,
"enriched_ast": None,
"error": "No database session — cannot generate script.",
"failed_operations": {},
}
try:
mv, release_row = self._resolve_release(
module_code, module_version, release
)
(
expressions,
operation_vids,
preconditions,
severities,
from_submission_dates,
) = self._discover_module_validations(mv, release_row)
except ValueError as exc:
return {
"success": False,
"enriched_ast": None,
"error": str(exc),
"failed_operations": {},
}
return self.script_from_db(
expressions=expressions,
operation_vids=operation_vids,
mv=mv,
release_row=release_row,
preconditions=preconditions or None,
severities=severities or None,
from_submission_dates=from_submission_dates or None,
)
[docs]
def calculations_for_module(
self,
module_code: str,
reference_date: str,
publication_date: Optional[str] = None,
) -> Dict[str, Any]:
"""Export a module version's calculations set.
Where :meth:`script_for_module` scripts a module's *validations*,
this exports its *calculations*: the operations linked to the
module version through ``OperationOutput``, parsed as one script,
reordered so a calculation follows the ones it consumes, and
serialized with every operand resolved to its datapoints.
The companion datapoint map is
:meth:`calculations_datapoints`; the two are computed by the same
pass, so a caller that wants both should use
:meth:`calculations_export` and read the two fields off it.
Args:
module_code: Code of the module (e.g. ``"KRI"``).
reference_date: Reference date, ``YYYY-MM-DD``. The module
version whose reference-date window contains it is the
one exported; exactly one must.
publication_date: Publication date stamped into the
``dpm_release`` block; defaults to today.
Returns:
The export, keyed by the module's EBA taxonomy URI.
Raises:
ValueError: If the service has no database session.
DpmCoreError: If the module version, its calculations or the
``OperationOutput`` table cannot be resolved.
"""
return self.calculations_export(
module_code, reference_date, publication_date
).calculations
[docs]
def calculations_datapoints(
self,
module_code: str,
reference_date: str,
publication_date: Optional[str] = None,
) -> Dict[str, Dict[str, Any]]:
"""Map every datapoint a module's calculations touch to its cell.
The companion of :meth:`calculations_for_module`, with the same
arguments and the same failure modes.
Args:
module_code: Code of the module (e.g. ``"KRI"``).
reference_date: Reference date, ``YYYY-MM-DD``.
publication_date: Publication date stamped into the
``dpm_release`` block; defaults to today.
Returns:
``{variable_id: {"table", "row", "column", "sheet"}}``.
Raises:
ValueError: If the service has no database session.
DpmCoreError: If the module version, its calculations or the
``OperationOutput`` table cannot be resolved.
"""
return self.calculations_export(
module_code, reference_date, publication_date
).datapoints
[docs]
def calculations_export(
self,
module_code: str,
reference_date: str,
publication_date: Optional[str] = None,
) -> "CalculationsExport":
"""Export a module's calculations and datapoint map in one pass.
Args:
module_code: Code of the module (e.g. ``"KRI"``).
reference_date: Reference date, ``YYYY-MM-DD``.
publication_date: Publication date stamped into the
``dpm_release`` block; defaults to today.
Returns:
Both halves of the export.
Raises:
ValueError: If the service has no database session.
DpmCoreError: If the module version, its calculations or the
``OperationOutput`` table cannot be resolved.
"""
from dpmcore.services.calculations_export import CalculationsExporter
if self.session is None:
raise ValueError(
"No database session — cannot export calculations."
)
return CalculationsExporter(self.session).export(
module_code, reference_date, publication_date
)
[docs]
def list_module_versions(
self,
module_code: Optional[str] = None,
release: Optional[str] = None,
) -> List[Tuple[str, str]]:
"""Enumerate ``(module_code, version_number)`` pairs to sweep.
Used by the CLI's ``--all-modules``/``--all-versions`` sweep to
discover targets for :meth:`script_for_module`.
Args:
module_code: Restrict to this module's versions. ``None``
enumerates every module in the database.
release: When given, only ``ModuleVersion`` rows whose window
contains this release (via :func:`filter_by_release`) are
returned.
Returns:
``(module_code, version_number)`` pairs, ordered by
``(code, start_release_id)``, excluding ghost/phantom
``ModuleVersion`` rows (``from_reference_date ==
to_reference_date``, both non-null — the same test
:meth:`_walk_ghost_chain` uses). With ``release``, a module
whose only covering versions are ghosts is represented by
its ghost fallback instead of being dropped (#372, see
:meth:`_ghost_fallback_versions`).
Raises:
ValueError: If a database session is missing, or ``release``
doesn't match any ``Release.code``.
"""
from sqlalchemy import or_
from dpmcore.dpm_xl.utils.filters import (
filter_by_release,
resolve_release_id,
)
from dpmcore.orm.packaging import ModuleVersion
from dpmcore.orm.release_sort_order import load_release_sort_orders
if self.session is None:
raise ValueError("No database session — cannot list modules.")
session = self.session
query = session.query(
ModuleVersion.code,
ModuleVersion.version_number,
ModuleVersion.start_release_id,
).filter(
or_(
ModuleVersion.from_reference_date.is_(None),
ModuleVersion.to_reference_date.is_(None),
ModuleVersion.from_reference_date
!= ModuleVersion.to_reference_date,
)
)
if module_code is not None:
query = query.filter(ModuleVersion.code == module_code)
release_id: Optional[int] = None
if release is not None:
release_id = resolve_release_id(session, release_code=release)
query = filter_by_release(
query,
start_col=ModuleVersion.start_release_id,
end_col=ModuleVersion.end_release_id,
release_id=release_id,
)
# Unpacked rather than ``list(query.all())``: a SQLAlchemy ``Row``
# is not a ``tuple`` to mypy, and the fallback rows are appended
# to this same list.
rows: List[Tuple[Optional[str], Optional[str], Optional[int]]] = [
(code, version_number, start_release_id)
for code, version_number, start_release_id in query.all()
]
if release_id is not None:
rows.extend(
self._ghost_fallback_versions(session, release_id, module_code)
)
sort_orders = load_release_sort_orders(session)
seen: set[Tuple[str, str]] = set()
pairs: List[Tuple[str, str]] = []
for code, version_number, _start in sorted(
rows, key=lambda row: _sweep_order(sort_orders, row)
):
if code is None or version_number is None:
continue
pair = (code, version_number)
if pair not in seen:
seen.add(pair)
pairs.append(pair)
return pairs
def _release_ghost_fallbacks(
self, session: "Session", release_id: int
) -> Dict[int, List[int]]:
"""``{fallback_module_vid: [ghost_module_vid, ...]}``, memoised.
The map is release-wide and identical for every module of a
release, so an ``--all-modules`` sweep computes it once instead
of once per target: each miss costs two ``ModuleVersion``
queries plus a full ``Release`` scan, and the sweep asks for it
again in :meth:`_ghosts_represented_by` for every module it
scripts.
"""
from dpmcore.dpm_xl.model_queries import ModuleVersionQuery
cached = self._ghost_fallback_cache.get(release_id)
if cached is None:
cached = ModuleVersionQuery.ghost_fallbacks(session, release_id)
self._ghost_fallback_cache[release_id] = cached
return cached
def _ghost_fallback_versions(
self,
session: "Session",
release_id: int,
module_code: Optional[str],
) -> List[Tuple[Optional[str], Optional[str], Optional[int]]]:
"""Module versions standing in for a ghost at ``release_id``.
A module whose only release-covering versions are ghosts has no
sweep target of its own, since :meth:`list_module_versions`
excludes ghosts — and rightly so, a ghost scripts to a
single-day reporting window. Per the #182 fallback rule the
release is represented by the latest prior non-ghost version of
the module, so that version is the module's export target
(#372). Its own release window ends before ``release_id``;
:meth:`_resolve_explicit_release` accepts it anyway, because
:meth:`_effective_end_release_id` extends a fallback's end past
its ghost siblings (#221).
``module_code`` filters the *fallback's own* code, not the
module it belongs to: a module renamed across versions (six of
them in the 4.2.1 dictionary, e.g. ``REM_BM`` → ``REM_BM_CI``)
would otherwise answer a request for the new name with a version
carrying the old one, which is neither what the caller asked for
nor what an ``--all-modules`` sweep returns at that release.
Args:
session: SQLAlchemy session.
release_id: Target release id.
module_code: Restrict to this module code, ``None`` for all.
Returns:
``(code, version_number, start_release_id)`` triples, in the
shape :meth:`list_module_versions` sorts.
"""
from dpmcore.orm.packaging import ModuleVersion
from dpmcore.orm.query_utils import chunked_in
substitutions = self._release_ghost_fallbacks(session, release_id)
if not substitutions:
return []
query = session.query(
ModuleVersion.code,
ModuleVersion.version_number,
ModuleVersion.start_release_id,
)
if module_code is not None:
query = query.filter(ModuleVersion.code == module_code)
return list(chunked_in(query, ModuleVersion.module_vid, substitutions))
def _discover_module_validations(
self,
mv: Any,
release_row: Any,
) -> Tuple[
List[Tuple[str, str]],
Dict[str, int],
List[Union[Tuple[str, List[str]], Dict[str, Any]]],
Dict[str, str],
Dict[str, str],
]:
"""Look up the active ``(expression, code)`` pairs for a module.
Resolves the module version's active operations by joining
``OperationVersion`` to ``OperationScope`` /
``OperationScopeComposition``, restricted to the operations
composed into *mv* — but, unlike a single-release point lookup,
an ``OperationVersion`` is a candidate whenever its window
**overlaps** *mv*'s own ``[start_release_id, end_release_id)``
(mirrors mdpm's ``get_active_operations``, using
:mod:`dpmcore.orm.release_sort_order` instead of raw
``ReleaseID`` comparison so a non-monotonic release ID can't
misorder the overlap check).
When ``mv`` stands in for a ghost at ``release_row`` the ghost's
scopes are read through it as well — see
:meth:`_ghosts_represented_by`. Outside that substitution the
filter stays on ``mv``'s own ``ModuleVID``, so a module that has
both a genuine covering version and a ghost at the same release
still does not see the ghost's scopes; no module in the 4.2.1
dictionary has that shape, but it is the case to revisit if one
appears.
The *filter* above widens across *mv*'s whole ghost chain so a
row composed solely against an earlier ghost is fetchable at
all (``ghost_vids``, see :meth:`_ghosts_represented_by`), but
the *window* check only widens by the **one** ghost (if any)
whose own window actually covers ``release_row`` itself — see
:meth:`_release_ghost_fallbacks`. Using the whole chain there
instead would pull in a later, unrelated ghost's validations
for an earlier release entirely (Andrés's review on PR #396:
``COREP_FRTB-3.1.0`` at release ``3.5`` wrongly gained
``v23241_s``/``v23242_s``, which only exist under ghost
``3.3.0``'s own ``4.0``-``4.2`` window, not ``3.2.0``'s — the
one that actually covers ``3.5``). The phantom-scope exclusion
below treats a phantom "other" module as rescued rather than
void when it has its own #182 fallback at this same,
release-specific ghost — unconditionally, for any caller
(``--module-version``, ``--all-versions`` or an explicit
``--release`` alike), per issue #182 itself: "this fallback
must apply consistently on every path dpmcore uses to resolve
which module version applies". A ``--module-version``/
``--all-versions`` lookup resolves *mv* "as itself" toward
whatever release is currently applicable (``release=None``'s
own auto-pick, #221) — when that resolved release is one only
a ghost covers, that ghost's validations are exactly what #182
says must surface there too (``DORA-1.1.0`` needs 67 whether
reached via ``--module-version`` or an explicit ``--release
4.2``, since ``1.2.0`` covers both releases). A release *mv*
covers on its own resolves no ghost here at all (``COREP_FRTB-
3.1.0`` at its own release ``3.4`` stays at its own 2
operations). mdpm's own reference has no equivalent to #182 in
any mode, so it is not a valid comparison for a ghost-involved
release; it still is for one with none.
Also drops any candidate whose *only* ``OperationScopeID`` under
*mv* is shared exclusively with phantom module versions (see
:meth:`list_module_versions`'s docstring for what "phantom"
means) — mirrors mdpm's ``is_phantom_module_op``: a validation
reaching this module only through a cross-framework pairing
whose counterpart never really existed as its own module
version should not surface here. A previous version of this
docstring said this guard "only matters when sweeping every
module" — confirmed wrong: a single-module lookup needs it too.
A code can match more than one ``OperationVersion`` row: the
ghost substitution above can return one operation twice, once
per scope, and the two scope rows may carry different
severities — ``mv``'s own scope is ranked above the ghost's so
the winner does not depend on join order.
Returns:
``(expressions, operation_vids, preconditions, severities,
from_submission_dates)`` ready to pass to :meth:`script`.
``operation_vids`` maps each code to the ``OperationVID``
already resolved by this same query, so callers that want
the persisted AST for an operation don't need a second
lookup for it. ``from_submission_dates`` maps each code to
its own ``OperationScope.FromSubmissionDate`` (formatted,
or absent when the column is null) — the real per-operation
date, not the module version's own reference date every
operation used to share.
"""
from sqlalchemy import or_
from dpmcore.orm.operations import (
Operation,
OperationScope,
OperationScopeComposition,
OperationVersion,
)
from dpmcore.orm.release_sort_order import (
compute_sort_order,
load_release_sort_orders,
sort_order_from,
)
session = self.session
if session is None:
raise ValueError("No database session — cannot generate script.")
if self._scope_calc is None:
raise InternalError(
"ScopeCalculatorService not initialised",
"_discover_module_validations requires a live DB session.",
)
# A release mv covers in its own right needs no ghost at all —
# for it, ghost_vids must stay empty even in the *filter* below,
# not just the window/phantom-rescue steps: a ghost-composed
# row can have its own window happen to overlap mv's own window
# too (COREP_FRTB's v22730_s, [1,3) against mv's own [1,2)) purely
# by coincidence of how its own OperationVersion was recorded,
# which must not leak it into a query about a release that never
# needed any ghost involvement to begin with (COREP_FRTB-3.1.0 at
# its own release 3.4 must stay at its own 2 operations, not gain
# v22730_s and the many other ghost-composed rows whose window
# happens to touch release 1 too).
sort_orders = load_release_sort_orders(session)
target_sort = sort_order_from(sort_orders, release_row.release_id)
mv_start_sort = sort_order_from(sort_orders, mv.start_release_id)
mv_end_sort = (
compute_sort_order(None, None)
if mv.end_release_id is None
else sort_order_from(sort_orders, mv.end_release_id)
)
release_needs_ghosts = not (mv_start_sort <= target_sort < mv_end_sort)
ghost_vids = (
self._ghosts_represented_by(mv) if release_needs_ghosts else []
)
query = (
session.query(
OperationVersion.operation_vid,
Operation.code,
OperationVersion.expression,
OperationScope.severity,
OperationVersion.precondition_operation_vid,
OperationScopeComposition.module_vid,
OperationScope.operation_scope_id,
OperationVersion.start_release_id,
OperationVersion.end_release_id,
OperationScope.from_submission_date,
)
.join(
OperationScope,
OperationVersion.operation_vid == OperationScope.operation_vid,
)
.join(
OperationScopeComposition,
OperationScope.operation_scope_id
== OperationScopeComposition.operation_scope_id,
)
.join(
Operation,
OperationVersion.operation_id == Operation.operation_id,
)
.filter(
OperationScopeComposition.module_vid.in_(
[mv.module_vid, *ghost_vids]
)
)
# Access boolean convention: True is stored as -1, not 1.
.filter(OperationScope.is_active.in_([-1, 1, True]))
.filter(
or_(
Operation.code.startswith("v"),
Operation.code.startswith("e"),
)
)
)
rows = query.all()
# ghost_vids (mv's whole chain, already gated above to empty when
# this release needs no ghost at all) widens the *filter* — it
# decides which rows are even fetchable, e.g. a row composed
# solely against an earlier ghost with no row of its own for mv
# (COREP_FRTB's v22730_s, composed only against 3.2.0). Whether a
# fetched row actually *applies* to this release is a window
# question, and only the one ghost (if any) whose own window
# covers release_row itself is relevant there — using the whole
# chain here pulled in a later, unrelated ghost's validations for
# an earlier release entirely (Andrés's review on PR #396:
# COREP_FRTB 3.1.0 at release 3.5 wrongly gained v23241_s/
# v23242_s, which only exist under ghost 3.3.0's own release
# 4.0-4.2 window, not 3.2.0's — the one that actually covers 3.5).
release_ghost_vids = (
self._release_ghost_fallbacks(session, release_row.release_id).get(
mv.module_vid, []
)
if release_needs_ghosts
else []
)
windows = [(mv.start_release_id, mv.end_release_id)]
if release_ghost_vids:
windows += self._ghost_windows(session, release_ghost_vids)
overlapping = self._overlapping_operation_version_rows(
rows, sort_orders, windows
)
scope_ids = {row[6] for row in overlapping}
if scope_ids:
# Unlike the window check above, the phantom-rescue below
# asks a release-independent, structural question — "is
# this composition's other module a dead end, or does
# something current stand in for it at all" — so it folds
# in mv's *whole* chain (ghost_vids), not just the one
# ghost specific to this release: a row can be fetched (via
# the filter's own use of the whole chain) and pass the
# window check through mv's own window directly (not
# needing any ghost's window at all, e.g. v22730_s's own
# window already overlaps mv's), while still being composed
# *only* against an earlier ghost with no row of its own
# for mv — that composition is never a dead end as long as
# mv is its fallback, regardless of which release resolved.
ghost_fallback_map = {
**self._scope_calc._ghost_fallback_map(release_row.release_id),
**dict.fromkeys(ghost_vids, mv.module_vid),
}
phantom_scope_ids = self._scope_calc._phantom_paired_scope_ids(
session, scope_ids, mv.module_vid, ghost_fallback_map
)
else:
phantom_scope_ids = set()
latest_by_code = self._latest_operation_by_code(
overlapping, phantom_scope_ids, mv.module_vid
)
expressions: List[Tuple[str, str]] = []
operation_vids: Dict[str, int] = {}
severities: Dict[str, str] = {}
from_submission_dates: Dict[str, str] = {}
prec_vid_to_codes: Dict[int, List[str]] = {}
for code, (
rank,
expression,
severity,
prec_vid,
from_submission_date,
) in sorted(latest_by_code.items()):
expressions.append((expression, code))
operation_vids[code] = rank[0]
if severity:
severities[code] = severity
formatted_date = _format_date(from_submission_date)
if formatted_date:
from_submission_dates[code] = formatted_date
if prec_vid is not None:
prec_vid_to_codes.setdefault(prec_vid, []).append(code)
preconditions: List[Union[Tuple[str, List[str]], Dict[str, Any]]] = (
list(self._resolve_preconditions(prec_vid_to_codes))
)
return (
expressions,
operation_vids,
preconditions,
severities,
from_submission_dates,
)
@staticmethod
def _overlapping_operation_version_rows(
rows: List[Tuple[Any, ...]],
sort_orders: Dict[int, int],
windows: List[Tuple[Optional[int], Optional[int]]],
) -> List[Tuple[Any, ...]]:
"""Rows whose ``OperationVersion`` window overlaps any of *windows*.
Mirrors mdpm's ``get_active_operations``, using
:mod:`dpmcore.orm.release_sort_order` instead of raw
``ReleaseID`` comparison so a non-monotonic release ID can't
misorder the overlap check. *windows* is normally just *mv*'s
own ``(start_release_id, end_release_id)``; whenever *mv*
stands in for a #182 ghost at the resolved release, one window
per such ghost is added too (see :meth:`_ghost_windows`) — a
ghost's own window can start later than the fallback's, so an
``OperationVersion`` scoped through the ghost and checked only
against the fallback's window would be wrongly dropped
(confirmed against DORA's ghost 1.2.0, operation ``v903581_m``).
Split out of :meth:`_discover_module_validations` to keep its
own branching within the complexity limit.
"""
from dpmcore.orm.release_sort_order import (
compute_sort_order,
sort_order_from,
)
open_end_sort = compute_sort_order(None, None)
def _end_sort(release_id: Optional[int]) -> int:
if release_id is None:
return open_end_sort
return sort_order_from(sort_orders, release_id)
def _start_sort(release_id: Optional[int]) -> int:
if release_id is None:
return 0 # sorts before every real release's sort order
return sort_order_from(sort_orders, release_id)
sort_windows = [
(_start_sort(start), _end_sort(end)) for start, end in windows
]
return [
row
for row in rows
if any(
window_end > sort_order_from(sort_orders, row[7])
and _end_sort(row[8]) > window_start
for window_start, window_end in sort_windows
)
]
@staticmethod
def _latest_operation_by_code(
overlapping: List[Tuple[Any, ...]],
phantom_scope_ids: Set[int],
home_module_vid: int,
) -> Dict[
str, Tuple[Tuple[int, int], str, Optional[str], Optional[int], Any]
]:
"""Keep the latest (highest ``OperationVID``) row per code.
A code can match more than one ``OperationVersion`` row: the
ghost substitution in :meth:`_discover_module_validations` can
return one operation twice, once per scope, and the two scope
rows may carry different severities — ``mv``'s own scope is
ranked above the ghost's so the winner does not depend on join
order. Split out to keep the caller's own branching within the
complexity limit.
"""
latest_by_code: Dict[
str,
Tuple[Tuple[int, int], str, Optional[str], Optional[int], Any],
] = {}
for (
op_vid,
code,
expression,
severity,
prec_vid,
module_vid,
scope_id,
_start_rel,
_end_rel,
from_submission_date,
) in overlapping:
if scope_id in phantom_scope_ids:
continue
rank = (op_vid, 1 if module_vid == home_module_vid else 0)
existing = latest_by_code.get(code)
if existing is None or rank > existing[0]:
latest_by_code[code] = (
rank,
expression,
severity,
prec_vid,
from_submission_date,
)
return latest_by_code
def _ghosts_represented_by(self, mv: Any) -> List[int]:
"""Every #182 ghost sibling in ``mv``'s own contiguous chain.
When ``mv`` is the #182 fallback for a module whose own release
window is bridged by one or more ghost versions, the
validations active across that whole bridged span are the ones
scoped to those ghosts: their ``OperationScopeComposition``
rows carry the ghost's ``ModuleVID``, which a plain
``mv.module_vid`` filter never sees, so the fallback scripted
to 30 of DORA's 67 validations at 4.2 (#372).
Walks the exact same contiguous run of ghost siblings
:meth:`_effective_end_release_id` extends ``mv``'s own window
through — every ghost between ``mv`` and the next non-ghost
sibling (or indefinitely, if none follows) — rather than only
whichever one ghost happens to cover a single resolved release:
a chain can bridge more than one ghost (``COREP_FRTB``'s
``3.1.0`` stands in for both ``3.2.0`` and ``3.3.0`` before its
real ``3.4.0`` takes over), and a validation scoped only to an
earlier ghost in that chain (not the one covering whatever
release happened to be resolved) must not be dropped either.
The ghosts' VIDs widen that filter rather than replacing it: a
validation the fallback carries and the ghost dropped stays in
the script, which is what the module reported at the fallback's
own releases. In the 4.2.1 dictionary the two differ for one
module (``IF_CLASS3`` at 4.0 and 4.1, one validation).
Args:
mv: The module version being scripted.
Returns:
The ghost ``ModuleVID``s, empty when ``mv`` has no ghost
sibling bridging its own window.
"""
from dpmcore.orm.packaging import ModuleVersion
from dpmcore.orm.release_sort_order import resolve_sort_order
if (
mv.end_release_id is None
or self.session is None
or mv.module_id is None
):
return []
session = self.session
end_sort = resolve_sort_order(
session, mv.end_release_id, role="module version end release"
)
siblings = (
session.query(ModuleVersion)
.filter(ModuleVersion.module_id == mv.module_id)
.filter(ModuleVersion.module_vid != mv.module_vid)
.all()
)
candidates = self._siblings_past_end(siblings, end_sort)
ghost_vids: List[int] = []
boundary = end_sort
for start_sort, is_ghost, sibling in candidates:
if start_sort > boundary or not is_ghost:
break
ghost_vids.append(sibling.module_vid)
if sibling.end_release_id is None:
break
sibling_end_sort = resolve_sort_order(
session,
sibling.end_release_id,
role="sibling module version end release",
)
if sibling_end_sort > boundary:
boundary = sibling_end_sort
return ghost_vids
@staticmethod
def _ghost_windows(
session: "Session", ghost_vids: List[int]
) -> List[Tuple[Optional[int], Optional[int]]]:
"""``(start_release_id, end_release_id)`` for each of *ghost_vids*.
See :meth:`_overlapping_operation_version_rows` — a ghost's own
release window can start later than its fallback's, and an
operation genuinely scoped through the ghost is windowed
against the ghost, not the fallback.
"""
from dpmcore.orm.packaging import ModuleVersion
rows = (
session.query(
ModuleVersion.start_release_id, ModuleVersion.end_release_id
)
.filter(ModuleVersion.module_vid.in_(ghost_vids))
.all()
)
return [(row[0], row[1]) for row in rows]
def _resolve_preconditions(
self, prec_vid_to_codes: Dict[int, List[str]]
) -> List[Dict[str, Any]]:
"""Resolve discovered ``PreconditionOperationVID``s into entries.
Note: :meth:`_build_preconditions_block` parses the stored
``Expression`` and walks the resulting AST — it does not rebuild
the gate from the DB's ``OperationNode`` graph. What it can
express is therefore the engine's gate contract: filing-indicator
selections, run-time parameters and boolean literals combined
with ``and`` / ``or`` / ``xor`` / ``not`` (see its docstring and
``TestBuildPreconditionsBlock`` in
``tests/unit/services/test_ast_generator.py``). Two outcomes are
not the same and neither is specific to auto-discovery:
* a gate outside that contract is **not** silently dropped — the
operations it guards are reported in ``failed_operations`` and
left out of the script, because ungating them would change what
gets validated (#338);
* a gate whose ``{v_<variable_code>}`` codes do not resolve for
the release still drops silently, same as pydpm.
"""
if not prec_vid_to_codes:
return []
from dpmcore.orm.operations import Operation, OperationVersion
session = self.session
if session is None:
raise ValueError("No database session — cannot generate script.")
rows = (
session.query(
OperationVersion.operation_vid,
Operation.code,
OperationVersion.expression,
)
.join(
Operation,
OperationVersion.operation_id == Operation.operation_id,
)
.filter(
OperationVersion.operation_vid.in_(prec_vid_to_codes.keys())
)
.all()
)
preconditions: List[Dict[str, Any]] = []
for prec_vid, prec_code, prec_expression in rows:
preconditions.append(
{
"expression": prec_expression,
"affected_operations": sorted(prec_vid_to_codes[prec_vid]),
"code": prec_code,
"version_id": prec_vid,
}
)
return preconditions
# ------------------------------------------------------------------ #
# Resolution helpers
# ------------------------------------------------------------------ #
def _resolve_module_version(
self,
module_code: str,
module_version: str,
) -> Optional[Any]:
"""Look up a ``ModuleVersion`` by ``(code, version_number)``.
Returns the ORM row, or ``None`` if no match.
"""
from dpmcore.orm.packaging import ModuleVersion
if self.session is None:
return None
return (
self.session.query(ModuleVersion)
.filter(ModuleVersion.code == module_code)
.filter(ModuleVersion.version_number == module_version)
.first()
)
def _resolve_release(
self,
module_code: str,
module_version: str,
release: Optional[str],
) -> Tuple[Any, Any]:
"""Resolve ``(ModuleVersion, Release)`` for the request.
When ``release`` is omitted, falls back to the most recent
``Release`` whose window contains the requested
``ModuleVersion``.
"""
mv = self._resolve_module_version(module_code, module_version)
if mv is None:
raise ValueError(
f"ModuleVersion not found: {module_code} {module_version}"
)
if self.session is None:
raise ValueError("No database session — cannot resolve release.")
if release is not None:
release_row = self._resolve_explicit_release(
release, mv, module_code, module_version
)
return mv, release_row
latest = self._latest_release_in_window(mv)
if latest is None:
raise ValueError(
f"No Release matches module version {module_code} "
f"{module_version} window."
)
return mv, latest
def _resolve_explicit_release(
self,
release: str,
mv: Any,
module_code: str,
module_version: str,
) -> Any:
"""Resolve and window-check an explicit release.
Looks up ``Release.code == release`` and validates that the
release sits inside ``mv``'s window. Comparison runs against the
date-based sort order of each release (the DPM ``ReleaseID`` FK
is no longer monotonic — see
:mod:`dpmcore.orm.release_sort_order`), not the raw id. Raises
``ValueError`` if the release is unknown, predates
``start_release_id``, or is past ``end_release_id``.
Coordinates with the ghost-version fallback in
:func:`dpmcore.dpm_xl.model_queries._resolve_with_ghost_fallback`:
when ``mv`` is being used as the fallback for a ghost that
follows it, ``mv``'s effective end release is virtually extended
past the ghost(s) to the start of the next non-ghost sibling —
or ``None`` when only ghosts follow (issue #221). The
``predates`` branch is never relaxed:
``_latest_prior_non_collapsed_vids`` picks fallbacks strictly
backward, so a legitimate ghost-fallback can never produce an MV
whose start ``predates`` the requested release.
"""
from dpmcore.orm.infrastructure import Release
from dpmcore.orm.release_sort_order import (
compute_sort_order,
resolve_sort_order,
)
if self.session is None:
raise RuntimeError("session required")
release_row = (
self.session.query(Release).filter(Release.code == release).first()
)
if release_row is None:
raise ValueError(f"Release not found: {release}")
target = resolve_sort_order(
self.session, release_row.release_id, role=f"release {release}"
)
start = mv.start_release_id
end = mv.end_release_id
if start is not None and target < resolve_sort_order(
self.session, start, role="module version start release"
):
raise ValueError(
f"Release {release} predates module version "
f"{module_code} {module_version} "
f"(starts at release_id={start})."
)
if end is not None:
effective_end_id = self._effective_end_release_id(mv)
if effective_end_id is not None:
effective_end_sort = resolve_sort_order(
self.session,
effective_end_id,
role="module version effective end release",
)
# A row ending at an "always latest" release (undated or
# non-chronological) is still open even when queried at
# that release.
if (
effective_end_sort != compute_sort_order(None, None)
and target >= effective_end_sort
):
raise ValueError(
f"Release {release} is past the end of module "
f"version {module_code} {module_version} "
f"(ends at release_id={end})."
)
return release_row
def _effective_end_release_id(self, mv: Any) -> Optional[int]:
"""Compute ``mv``'s effective end, extended past ghost siblings.
When ``mv`` is used as ghost-fallback (its content substitutes
for a ghost of the same module covering releases past ``mv``'s
own end), the effective end virtually extends past those ghost
siblings. The extension follows a contiguous chain of ghost
siblings starting at (or overlapping) ``mv.end_release_id``:
- If the chain is followed by a non-ghost sibling, the effective
end is that sibling's ``start_release_id``.
- If only ghosts follow (open-ended ghost, or the last ghost's
end is null), the effective end is ``None`` (open).
- If no ghost adjoins ``mv``'s end, the effective end is
``mv.end_release_id`` unchanged.
Args:
mv: The ``ModuleVersion`` being window-checked.
Returns:
The release id to use as the effective upper bound of
``mv``'s window, or ``None`` if it is open-ended.
"""
from dpmcore.orm.packaging import ModuleVersion
from dpmcore.orm.release_sort_order import resolve_sort_order
if mv.end_release_id is None:
return None
if self.session is None or mv.module_id is None:
return mv.end_release_id
session = self.session
end_sort = resolve_sort_order(
session,
mv.end_release_id,
role="module version end release",
)
siblings = (
session.query(ModuleVersion)
.filter(ModuleVersion.module_id == mv.module_id)
.filter(ModuleVersion.module_vid != mv.module_vid)
.all()
)
candidates = self._siblings_past_end(siblings, end_sort)
return self._walk_ghost_chain(mv, candidates, end_sort)
def _siblings_past_end(
self, siblings: Iterable[Any], end_sort: int
) -> List[Tuple[int, bool, Any]]:
"""Filter and sort siblings whose window extends past ``end_sort``."""
from dpmcore.orm.release_sort_order import resolve_sort_order
session = self.session
if session is None:
raise RuntimeError("session required")
def sort_or_none(
release_id: Optional[int], role: str
) -> Optional[int]:
if release_id is None:
return None
return resolve_sort_order(session, release_id, role=role)
candidates: List[Tuple[int, bool, Any]] = []
for sibling in siblings:
sibling_end_sort = sort_or_none(
sibling.end_release_id, "sibling module version end release"
)
if sibling_end_sort is not None and sibling_end_sort <= end_sort:
continue
sibling_start_sort = sort_or_none(
sibling.start_release_id,
"sibling module version start release",
)
# Missing start acts as unbounded below; cap the sort key at
# ``end_sort`` so such siblings sit at the chain's head.
effective_start = (
end_sort
if sibling_start_sort is None
else max(sibling_start_sort, end_sort)
)
is_ghost = (
sibling.from_reference_date is not None
and sibling.to_reference_date is not None
and sibling.from_reference_date == sibling.to_reference_date
)
candidates.append((effective_start, is_ghost, sibling))
candidates.sort(key=lambda item: item[0])
return candidates
def _walk_ghost_chain(
self,
mv: Any,
candidates: List[Tuple[int, bool, Any]],
end_sort: int,
) -> Optional[int]:
"""Walk sibling candidates to find ``mv``'s effective end."""
from dpmcore.orm.release_sort_order import resolve_sort_order
session = self.session
if session is None:
raise RuntimeError("session required")
saw_ghost = False
boundary = end_sort
last_ghost_end_id: Optional[int] = None
for start_sort, ghost, sibling in candidates:
if start_sort > boundary:
break
if ghost:
saw_ghost = True
if sibling.end_release_id is None:
return None
sibling_end_sort = resolve_sort_order(
session,
sibling.end_release_id,
role="sibling module version end release",
)
if sibling_end_sort > boundary:
boundary = sibling_end_sort
last_ghost_end_id = sibling.end_release_id
continue
if saw_ghost:
# First non-ghost after a chain of ghosts terminates the
# virtual extension at its ``start_release_id``.
return sibling.start_release_id
# A non-ghost adjoins ``mv``'s end directly (no ghost bridge),
# so no extension applies — keep the schema value.
return mv.end_release_id
# Chain of ghosts with no non-ghost following it: the effective
# end is the last bounded ghost's ``end_release_id``. (An
# open-ended ghost would have returned ``None`` inside the loop.)
if saw_ghost:
return last_ghost_end_id
return mv.end_release_id
def _latest_release_in_window(self, mv: Any) -> Any:
"""Pick the latest Release covering *mv*'s window.
A release marked ``is_current`` in the DB is the DB's own signal
that this row represents "the current release" — pick that one
first (there can only be one). When no candidate is marked
current, fall back to the latest by date-based sort order
(``Release.date`` via :func:`compute_sort_order`), then to the
latest of any status if no released row matches.
The previous logic filtered candidates to ``status='released'``
before ordering by date, which silently downgraded a newer
release still in ``status='validation'`` — the exact shape that
made dpmcore emit ``4.2`` while the reference declared ``4.2.1``
for the same fixture DB.
The upper bound is ``mv``'s *effective* end
(:meth:`_effective_end_release_id`), the same bound
:meth:`_resolve_explicit_release` window-checks against. Reading
the raw ``end_release_id`` here instead made the two entry points
disagree: a release the explicit path accepts because ``mv``
stands in for a ghost there (#221) was one this path would never
pick, so ``script_for_module`` without a release silently
returned the narrower pre-#372 script.
"""
from dpmcore.orm.infrastructure import Release
from dpmcore.orm.release_sort_order import (
compute_sort_order,
resolve_sort_order,
)
if self.session is None:
raise RuntimeError("session required")
start_sort: Optional[int] = None
end_sort: Optional[int] = None
if mv.start_release_id is not None:
start_sort = resolve_sort_order(
self.session,
mv.start_release_id,
role="Module version window start release",
)
effective_end_id = self._effective_end_release_id(mv)
if effective_end_id is not None:
end_sort = resolve_sort_order(
self.session,
effective_end_id,
role="Module version window end release",
)
perpetual_sort_order = compute_sort_order(None, None)
rows = self.session.query(Release).all()
candidates: List[tuple[int, int, Any]] = []
for r in rows:
so = compute_sort_order(r.date, r.type)
if start_sort is not None and so < start_sort:
continue
# A row ending at an "always latest" release (undated or
# non-chronological) is still open even when queried at
# that release.
if (
end_sort is not None
and end_sort != perpetual_sort_order
and so >= end_sort
):
continue
candidates.append((so, r.release_id, r))
if not candidates:
return None
# release_id breaks ties among releases sharing a sort order.
current = [c for c in candidates if c[2].is_current]
if current:
return max(current, key=lambda c: (c[0], c[1]))[2]
released = [c for c in candidates if c[2].status == "released"]
pool = released or candidates
return max(pool, key=lambda c: (c[0], c[1]))[2]
def _resolve_severities(
self,
severity: Optional[str],
severities: Optional[Dict[str, str]],
validation_codes: List[str],
) -> Dict[str, str]:
"""Validate severity inputs and return ``{code: severity}``.
- Global default falls back to ``SEVERITY_WARNING`` when
``severity`` is ``None``.
- Each entry of ``severities`` is validated independently.
- Codes in ``severities`` that are not in ``validation_codes``
raise ``ValueError`` so callers learn at request time.
"""
def _normalise(value: str, label: str) -> str:
if not isinstance(value, str):
raise ValueError(
f"Invalid severity for {label}: must be a string"
)
lowered = value.lower()
if lowered not in VALID_SEVERITIES:
allowed = ", ".join(sorted(VALID_SEVERITIES))
raise ValueError(
f"Invalid severity {value!r} for {label}. "
f"Must be one of: {allowed}"
)
return lowered
global_value = (
_normalise(severity, "default")
if severity is not None
else SEVERITY_WARNING
)
per_code: Dict[str, str] = {}
if severities:
known_codes = set(validation_codes)
for raw_code, raw_severity in severities.items():
if raw_code not in known_codes:
raise ValueError(
f"Unknown validation_code in severities: {raw_code!r}"
)
per_code[raw_code] = _normalise(
raw_severity, f"validation {raw_code!r}"
)
return {
code: per_code.get(code, global_value) for code in validation_codes
}
@staticmethod
def _resolve_root_operator_id(ast: Any, session: "Session") -> int:
"""Resolve the OperatorID at the root of an expression AST.
Walks past structural wrappers (``Start``, ``WithExpression``,
``PersistentAssignment`` / ``TemporaryAssignment``) down to the
first node carrying an ``op`` attribute, then looks up
``Operator.OperatorID`` by ``Symbol`` via the same DataFrame
``MLGeneration.create_operation_node`` uses.
``ParExpr`` is treated as the operator itself (Symbol ``()``,
OperatorID 37 in the reference operator table), mirroring pydpm:
an expression whose body is wrapped in parentheses roots at the
paren operator, not at the operator inside. ``CondExpr`` and
``ParExpr`` carry no ``op`` attribute but map to fixed synthetic
symbols.
Raises ``RuntimeError`` if no operator is resolvable.
"""
from dpmcore.dpm_xl.model_queries import OperatorQuery
node: Any = ast
# Walk through wrappers down to the operator node.
for _ in range(64): # bounded to avoid runaway recursion
class_name = type(node).__name__
if class_name == "Start":
children = getattr(node, "children", None) or []
if not children:
break
node = children[0]
continue
if class_name == "WithExpression":
node = node.expression
continue
if class_name in ("PersistentAssignment", "TemporaryAssignment"):
# The assigned expression carries the comparison.
node = node.right
continue
break
class_name = type(node).__name__
op_symbol = getattr(node, "op", None)
if not op_symbol and class_name == "CondExpr":
# CondExpr carries no ``op`` attribute; its operator is fixed.
op_symbol = "if-then-else"
elif not op_symbol and class_name == "ParExpr":
# ParExpr carries no ``op`` attribute either; the root of
# a body wrapped in parentheses is the paren operator, not
# the operator inside. pydpm serialises this as OperatorID 37.
op_symbol = "()"
if not op_symbol:
raise RuntimeError(
f"Cannot resolve root operator: AST root "
f"{class_name!r} has no 'op' attribute."
)
df = OperatorQuery.get_operators(session)
matches = df[df["Symbol"] == op_symbol]["OperatorID"].values
if len(matches) == 0:
raise RuntimeError(
f"No OperatorID found for symbol {op_symbol!r}."
)
return int(matches[0])
# ------------------------------------------------------------------ #
# Section builders
# ------------------------------------------------------------------ #
@staticmethod
def _build_module_info(mv: Any) -> Dict[str, Any]:
"""Extract module identity fields from a ``ModuleVersion`` row."""
framework_code: Optional[str] = None
module = getattr(mv, "module", None)
if module is not None:
framework = getattr(module, "framework", None)
if framework is not None:
framework_code = framework.code
return {
"module_code": mv.code or "",
"module_version": mv.version_number or "",
"framework_code": framework_code or "",
}
@staticmethod
def _build_release_info(release_row: Any) -> Dict[str, Any]:
"""Build ``{"release", "publication_date"}`` from a ``Release`` row.
dpmcore exposes the publication date as ``Release.date``;
pydpm calls it ``publication_date``. We use pydpm's name on
the wire because the engine consumes that key.
"""
return {
"release": release_row.code or "",
"publication_date": _format_date(
release_row.date, fallback=_DEFAULT_FROM_DATE
),
}
@staticmethod
def _build_dates(mv: Any) -> Dict[str, Any]:
"""Build the ``{"from", "to"}`` block from a ``ModuleVersion``."""
return {
"from": _format_date(
mv.from_reference_date, fallback=_DEFAULT_FROM_DATE
),
"to": _format_date(mv.to_reference_date),
}
@staticmethod
def _build_operation_entry(
expression: str,
code: str,
ast_dict: Any,
severity: str,
reference_date: Optional[str],
root_operator_id: int,
operation_vid: Optional[int] = None,
) -> Dict[str, Any]:
"""Assemble a single ``operations[code]`` entry.
``version_id`` is the real ``OperationVID`` when the caller has
one (:meth:`script_from_db` always does — the wire format's own
spec, dpm-xl-docs' ``03-document-sections.md`` §3.10, defines
``version_id`` as exactly that: "OperationVID of this Operation
version in the source DPM content"). Only when ``operation_vid``
is unavailable (:meth:`script`'s caller-supplied ``expressions``,
which may not even name a real DB operation — e.g. an
``--expressions`` file) does this fall back to a deterministic
CRC32 of the expression truncated to four digits, which replaces
pydpm's non-deterministic ``hash(expression) % 10000``.
``reference_date`` is the real ``OperationScope.FromSubmissionDate``
when the caller has it (:meth:`script_from_db`, per operation).
Only when unavailable — the column was null, or :meth:`script`'s
caller-supplied ``expressions`` have no ``OperationScope`` to
read at all — does the wire ``from_submission_date`` fall back to
the default below. The wire key stays ``from_submission_date``
(the established script format contract); only the Python name
reflects what it's actually fed from.
"""
version_id = (
operation_vid
if operation_vid is not None
else zlib.crc32(expression.encode("utf-8")) % 10000
)
return {
"version_id": version_id,
"code": code,
"expression": expression,
"root_operator_id": root_operator_id,
"ast": ast_dict,
"from_submission_date": reference_date or _DEFAULT_FROM_DATE,
"severity": severity,
}
# The gate contract shared with the engine's precondition evaluator:
# filing indicators, run-time parameters and boolean literals combined
# with these operators (grouping parentheses are preserved as
# ParExpr, not unwrapped). A gate
# carrying anything else — a comparison, arithmetic, a cell reference,
# a non-boolean literal — has no runtime meaning on the engine side, so
# it stays out of the script and its operations are reported through
# ``failed_operations`` (see ``_unsupported_gate_operations``).
_GATE_BINARY_OPS = frozenset({"and", "or", "xor"})
_GATE_UNARY_OPS = frozenset({"not"})
def _build_preconditions_block(
self,
preconditions: List[Union[Tuple[str, List[str]], Dict[str, Any]]],
release_id: Optional[int],
referenced_parameters: Optional[Dict[str, ParameterInfo]] = None,
emitted_operations: Optional[Set[str]] = None,
) -> Tuple[Dict[str, Any], Dict[str, str]]:
"""Build the ``preconditions`` and ``precondition_variables`` blocks.
Each gate is parsed once and the resulting AST is walked to build
the entry: a ``{v_*}`` selection becomes a ``PreconditionItem``
node with the resolved ``variable_id``, a ``{p_*}`` reference is
emitted as the same ``ParameterRef`` node an expression produces
(``code``, canonical ``param_type``, ``default``), and every
logical operator that connects them (``and`` / ``or`` / ``xor`` /
``not``) survives into the emitted tree, so ``{v_A} or {v_B}``
reaches the engine as a disjunction instead of the ``and``-fold
the old regex path produced. Grouping parentheses are rebuilt as
a ``ParExpr`` node rather than unwrapped — the spec requires
preserving source parenthesisation for reconstruction (§4.10),
even though a ``ParExpr`` carries no precedence of its own; see
:meth:`_transform_precondition_ast`.
A parameter that appears in a gate is added to
``referenced_parameters`` in the same pass, so the script-level
``parameters`` block declares it alongside the ones the
expressions reference and the scope-wide type agreement in
:func:`~dpmcore.services._parameters.merge_parameters` runs
against a single registry (``3-8`` on a conflicting type).
Gates outside that contract (a comparison, a cell reference, a
non-boolean literal) are excluded here, and :meth:`script`
reports their operations in ``failed_operations`` instead of
shipping a gate the engine cannot evaluate. Codes that don't
resolve are still skipped silently (matches pydpm).
When *emitted_operations* is given, each entry's
``affected_operations`` is intersected with it, and an entry
left gating nothing at all is dropped before it is walked, so
the ``precondition_variables`` it alone would have contributed
go with it (#355). Preconditions are harvested from the database
before the expressions they gate are semantically validated, so
an operation rejected into ``failed_operations`` — including one
rejected for the gate contract above — would otherwise stay
listed here. That is not cosmetic: the engine reports the
``affected_operations`` of every precondition that does not hold
as validations skipped, so a code absent from ``operations``
gets reported as skipped without ever having been part of the
script.
"""
from dpmcore.dpm_xl.model_queries import VariableVersionQuery
from dpmcore.dpm_xl.utils.serialization import serialize_ast
preconditions_dict: Dict[str, Any] = {}
precondition_variables: Dict[str, str] = {}
if not preconditions or self.session is None:
return preconditions_dict, precondition_variables
# Parse every gate once — later passes read the AST, not the
# source text — and collect every ``{v_*}`` code the trees carry
# so the DB lookup batches them all.
parsed_gates, all_codes = self._parse_gates(preconditions)
resolved: Dict[str, Dict[str, int]] = (
VariableVersionQuery.get_variable_vids_by_codes(
self.session, all_codes, release_id=release_id
)
if all_codes
else {}
)
for (
ast,
validation_codes,
provided_code,
provided_version_id,
) in parsed_gates:
if emitted_operations is not None:
validation_codes = [
code
for code in validation_codes
if code in emitted_operations
]
# Filtered before the walk, not after: dropping the gate
# here is what keeps the variables it alone would have
# declared out of ``precondition_variables`` (#355).
if not validation_codes:
continue
# Variables are registered per gate and merged only once the
# gate survives the walk: a gate discarded below must not
# leave the variables of its resolved leaves declared behind
# it (#355).
gate_variables: Dict[str, str] = {}
gate_ast = self._transform_precondition_ast(
ast, resolved, gate_variables, serialize_ast
)
if gate_ast is None:
# Nothing left standing: either every ``{v_*}`` position
# was dropped for lack of resolution, or an unresolved
# one sat where dropping it would have tightened the
# gate (see ``_transform_precondition_ast``).
continue
precondition_variables.update(gate_variables)
if referenced_parameters is not None:
self._accumulate_ast_parameters(
referenced_parameters, gate_ast
)
key, entry = self._build_precondition_entry_from_ast(
gate_ast, validation_codes, provided_code, provided_version_id
)
self._merge_precondition_entry(preconditions_dict, key, entry)
return preconditions_dict, precondition_variables
@staticmethod
def _merge_precondition_entry(
preconditions_dict: Dict[str, Any],
key: str,
entry: Dict[str, Any],
) -> None:
"""Insert *entry* under *key*; merge ops on collision.
Two preconditions with the same variable-vid set produce the
same key. Without merging, the second occurrence used to
clobber the first and lose its ``affected_operations``.
"""
existing = preconditions_dict.get(key)
if existing is None:
preconditions_dict[key] = entry
return
merged_ops = list(existing.get("affected_operations", []))
for op in entry.get("affected_operations", []):
if op not in merged_ops:
merged_ops.append(op)
existing["affected_operations"] = merged_ops
@staticmethod
def _gate_spec_fields(
precond_spec: Union[Tuple[str, List[str]], Dict[str, Any]],
) -> Tuple[str, List[str], Optional[str], Optional[int]]:
"""Normalise a gate spec to ``(expression, codes, code, vid)``.
Both input shapes are accepted: the ``(expression,
affected_operations)`` tuple, and the dict form that may also
carry an explicit ``code`` / ``version_id`` override.
"""
if isinstance(precond_spec, dict):
return (
precond_spec["expression"],
precond_spec["affected_operations"],
precond_spec.get("code"),
precond_spec.get("version_id"),
)
precond_expr, validation_codes = precond_spec
return precond_expr, validation_codes, None, None
def _unsupported_gate_operations(
self,
preconditions: List[Union[Tuple[str, List[str]], Dict[str, Any]]],
) -> Dict[str, str]:
"""Map each operation gated by an engine-unsupported gate to a reason.
The gate contract is filing indicators, run-time parameters and
boolean literals combined with ``and`` / ``or`` / ``xor`` /
``not``. The engine's precondition evaluator has no evaluation
for anything else — a comparison, arithmetic, a cell reference —
and such a node in a gate fails the run. A gate outside the
contract must therefore not be emitted, and silently dropping it
would ungate the operations it guards. This is the third option:
the operations are reported in :meth:`script`'s
``failed_operations`` with the reason, and left out of the
script. Gates that fail to parse are ignored here —
``_build_precondition_index`` already turns them into a
script-level error.
"""
failures: Dict[str, str] = {}
for precond_spec in preconditions:
precond_expr, validation_codes, _, _ = self._gate_spec_fields(
precond_spec
)
if not precond_expr:
continue
try:
ast = self._syntax.parse(precond_expr)
except Exception: # noqa: S112 — see docstring above
continue
reason = self._gate_unsupported_reason(ast)
if reason is None:
continue
message = (
f"Precondition {precond_expr!r} cannot be evaluated by the "
f"engine: {reason}. A precondition may only combine "
"filing indicators, parameters and boolean literals with "
"and/or/xor/not."
)
for validation_code in validation_codes:
failures.setdefault(validation_code, message)
return failures
@classmethod
def _gate_unsupported_reason(cls, node: Any) -> Optional[str]:
"""Why the engine could not evaluate this gate, or ``None`` if it can.
Walks the parsed gate and returns a short reason for the first
node outside the contract: filing-indicator selections
(``VarRef`` / ``PreconditionItem``), ``ParameterRef`` and boolean
``Constant`` leaves joined by ``_GATE_BINARY_OPS`` /
``_GATE_UNARY_OPS``, optionally grouped.
"""
from dpmcore.dpm_xl.ast import nodes as ast_nodes
node = cls._unwrap_start(node, ast_nodes)
if isinstance(node, ast_nodes.Start):
return "it is not a single expression"
if isinstance(node, ast_nodes.ParExpr):
return cls._gate_unsupported_reason(node.expression)
if isinstance(node, (ast_nodes.BinOp, ast_nodes.UnaryOp)):
return cls._gate_operator_unsupported_reason(node, ast_nodes)
if isinstance(
node,
(
ast_nodes.VarRef,
ast_nodes.PreconditionItem,
ast_nodes.ParameterRef,
),
):
return None
if isinstance(node, ast_nodes.Constant):
if node.type == "Boolean":
return None
return f"it contains the non-boolean literal {node.value!r}"
if isinstance(node, ast_nodes.VarID):
return "it references a cell"
return f"it contains a {node.__class__.__name__} node"
@classmethod
def _gate_operator_unsupported_reason(
cls, node: Any, ast_nodes: Any
) -> Optional[str]:
"""The operator half of :meth:`_gate_unsupported_reason`."""
if isinstance(node, ast_nodes.BinOp):
if node.op not in cls._GATE_BINARY_OPS:
return f"it uses the operator {node.op!r}"
return cls._gate_unsupported_reason(
node.left
) or cls._gate_unsupported_reason(node.right)
if node.op not in cls._GATE_UNARY_OPS:
return f"it uses the operator {node.op!r}"
return cls._gate_unsupported_reason(node.operand)
def _parse_gates(
self,
preconditions: List[Union[Tuple[str, List[str]], Dict[str, Any]]],
) -> Tuple[
List[Tuple[Any, List[str], Optional[str], Optional[int]]],
List[str],
]:
"""Parse every emittable gate once and index its ``{v_*}`` codes.
Returns ``(parsed_gates, all_codes)`` where ``parsed_gates`` is
the list of ``(ast, validation_codes, provided_code,
provided_version_id)`` tuples the caller iterates over, and
``all_codes`` is the deduplicated list of variable codes to feed
the batched DB resolution. Two kinds of gate are left out: one
that fails to parse — ``_build_precondition_index`` already
surfaces that as a script-level error — and one the engine
cannot evaluate, whose operations
``_unsupported_gate_operations`` has already reported.
"""
parsed_gates: List[
Tuple[Any, List[str], Optional[str], Optional[int]]
] = []
all_codes: List[str] = []
for precond_spec in preconditions:
(
precond_expr,
validation_codes,
provided_code,
provided_version_id,
) = self._gate_spec_fields(precond_spec)
if not precond_expr:
continue
try:
ast = self._syntax.parse(precond_expr)
except Exception: # noqa: S112 — see docstring above
continue
if self._gate_unsupported_reason(ast) is not None:
continue
parsed_gates.append(
(ast, validation_codes, provided_code, provided_version_id)
)
for var_code in self._extract_precondition_codes(ast):
if var_code not in all_codes:
all_codes.append(var_code)
return parsed_gates, all_codes
@classmethod
def _transform_precondition_ast(
cls,
node: Any,
resolved: Dict[str, Dict[str, int]],
precondition_variables: Dict[str, str],
serialize_ast: Any,
negated: bool = False,
) -> Optional[Dict[str, Any]]:
"""Walk the parsed gate AST and emit the engine's dict shape.
``VarRef`` (the parser's node for ``{v_*}``) and ``PreconditionItem``
collapse to a ``PreconditionItem`` dict with the resolved
``variable_id``, registering the ``variable_vid`` in
``precondition_variables`` on the way. A selection that does not
resolve becomes ``None`` — *unknown* — and unknown is only ever
allowed to relax the gate, never to tighten it: the surrounding
operator is rewired around it where dropping it widens the gate
(a conjunct of ``and``, or of ``or`` under a ``not``), and
otherwise the unknown propagates outwards until the whole gate
returns ``None`` and the operations it guarded run ungated.
Rewiring an ``or`` — or either side of an ``xor`` — the way an
``and`` is rewired would leave a *stricter* gate behind and skip
validations that should have run, so those cases propagate
instead. ``ParExpr`` wrappers are rebuilt around their transformed
``expression`` (dropped only when that inner result is itself
``None``) — the spec requires preserving source parenthesisation
for reconstruction even though it carries no precedence of its
own. ``BinOp``/``UnaryOp`` keep their operator, and a ``not``
flips the polarity the rewiring rule is read at. ``ParameterRef``
and boolean ``Constant`` leaves go through the standard
``serialize_ast`` path,
so the gate carries exactly the node shape an expression would
(``code`` / ``param_type`` / ``default``; ``type_`` / ``value``).
Any other node is a programming error here: ``_parse_gates`` keeps
gates carrying one out, so the emitter never has to guess a wire
shape the engine lacks.
The walker takes the parsed AST (not the serialised dict) so the
``variable`` attribute of a ``VarRef`` reaches the resolver intact.
"""
from dpmcore.dpm_xl.ast import nodes as ast_nodes
# Unwrap the top-level ``Start`` wrapper the grammar puts around
# every parsed expression. The engine's precondition entry
# matches the inside of that wrapper, not the wrapper itself.
node = cls._unwrap_start(node, ast_nodes)
if isinstance(node, ast_nodes.ParExpr):
inner = cls._transform_precondition_ast(
node.expression,
resolved,
precondition_variables,
serialize_ast,
negated,
)
if inner is None:
return None
return {"class_name": "ParExpr", "expression": inner}
if isinstance(node, (ast_nodes.VarRef, ast_nodes.PreconditionItem)):
variable = getattr(node, "variable", None) or getattr(
node, "variable_code", None
)
return cls._precondition_item(
variable, resolved, precondition_variables
)
if isinstance(node, (ast_nodes.ParameterRef, ast_nodes.Constant)):
serialized: Dict[str, Any] = serialize_ast(node)
return serialized
if isinstance(node, ast_nodes.BinOp):
return cls._transform_gate_binop(
node, resolved, precondition_variables, serialize_ast, negated
)
if isinstance(node, ast_nodes.UnaryOp):
operand = cls._transform_precondition_ast(
node.operand,
resolved,
precondition_variables,
serialize_ast,
# ``not`` is the only unary operator the gate contract
# allows (``_GATE_UNARY_OPS``), so descending here always
# flips the polarity.
not negated,
)
if operand is None:
return None
return {
"class_name": "UnaryOp",
"op": node.op,
"operand": operand,
}
raise InternalError(
"Unsupported node in a precondition gate",
f"{node.__class__.__name__} reached the gate emitter; "
"_parse_gates should have excluded this gate.",
)
@classmethod
def _transform_gate_binop(
cls,
node: Any,
resolved: Dict[str, Dict[str, int]],
precondition_variables: Dict[str, str],
serialize_ast: Any,
negated: bool,
) -> Optional[Dict[str, Any]]:
"""The binary-operator half of :meth:`_transform_precondition_ast`.
Both sides are walked first; an operand that came back ``None``
is an unresolved filing indicator, and whether the node may be
rewired around it depends on the operator and the polarity it
sits at — see :meth:`_gate_drop_relaxes`.
"""
left = cls._transform_precondition_ast(
node.left, resolved, precondition_variables, serialize_ast, negated
)
right = cls._transform_precondition_ast(
node.right,
resolved,
precondition_variables,
serialize_ast,
negated,
)
if left is None or right is None:
survivor = right if left is None else left
if survivor is None or not cls._gate_drop_relaxes(
node.op, negated
):
return None
return survivor
return {
"class_name": "BinOp",
"op": node.op,
"left": left,
"right": right,
}
@staticmethod
def _gate_drop_relaxes(op: str, negated: bool) -> bool:
"""Whether dropping an unknown operand of *op* widens the gate.
``unknown and X`` weakens to ``X``; ``unknown or X`` does not —
it would tighten to ``X`` and skip validations the gate was never
meant to skip. Under a ``not`` the two swap by De Morgan, so the
rule is read at the current polarity. ``xor`` relaxes at neither
polarity: an unknown operand of an exclusive or leaves nothing
honest to emit.
"""
return op == ("or" if negated else "and")
@staticmethod
def _unwrap_start(node: Any, ast_nodes: Any) -> Any:
"""Strip the ``Start`` wrapper the parser adds to every AST.
``SyntaxService.parse`` returns a ``Start`` node whose only child
is the actual expression. The engine's precondition entry is the
expression itself.
"""
if isinstance(node, ast_nodes.Start):
children = getattr(node, "children", None) or []
if len(children) == 1:
return children[0]
return node
@staticmethod
def _precondition_item(
variable: Any,
resolved: Dict[str, Dict[str, int]],
precondition_variables: Dict[str, str],
) -> Optional[Dict[str, Any]]:
"""Emit the ``PreconditionItem`` dict for a filing-indicator code.
Codes that don't resolve return ``None`` so the caller can drop
the position — the same non-fatal outcome the previous regex
path produced when a code was unknown.
"""
if not isinstance(variable, str):
return None
var_code = _normalize_variable_code(variable)
info = resolved.get(var_code)
if info is None:
return None
precondition_variables[str(info["variable_vid"])] = "b"
return {
"class_name": "PreconditionItem",
"variable_id": info["variable_id"],
"variable_code": var_code,
# Kept for key derivation in
# ``_build_precondition_entry_from_ast`` — the pre-existing
# ``p_<vid>`` scheme groups precondition entries by
# ``variable_vid``, not ``variable_id``. Stripped before the
# entry is emitted; the engine's node has no such field.
"variable_vid": info["variable_vid"],
}
@classmethod
def _accumulate_ast_parameters(
cls,
referenced_parameters: Dict[str, ParameterInfo],
node: Any,
) -> None:
"""Add every ``ParameterRef`` in a serialised AST to the registry.
Reads the type off the serialised node so the shared registry
holds the same canonical name the engine binds against. Used
both for a precondition gate's emitted AST and for an
operation's ``ast_dict`` reconstructed straight from the DB
(:meth:`script_from_db`) — a parameter can never survive
``build_ast_from_db`` (it raises ``UnsupportedDbAst``
instead), so this walk costs nothing there today, but ties
``parameters`` extraction to the AST itself rather than to
which path produced it. A parameter redeclared with another
type — in a gate, an expression, or both — raises
``SemanticError`` ``3-8`` through
:func:`~dpmcore.services._parameters.merge_parameters`.
``default`` is a per-reference fallback and stays on the node
only: the script-level registry is type-only.
"""
if isinstance(node, dict):
if node.get("class_name") == "ParameterRef":
code = node.get("code")
declared_type = node.get("param_type")
if isinstance(code, str) and isinstance(declared_type, str):
merge_parameters(
referenced_parameters,
(
ParameterInfo(
code=code,
declared_type=declared_type,
default=None,
),
),
)
for value in node.values():
if isinstance(value, (dict, list)):
cls._accumulate_ast_parameters(
referenced_parameters, value
)
elif isinstance(node, list):
for item in node:
if isinstance(item, (dict, list)):
cls._accumulate_ast_parameters(referenced_parameters, item)
@classmethod
def _is_conjunction_of_items(cls, node: Any) -> bool:
"""``True`` for the shape the regex path used to emit.
That is a ``PreconditionItem`` or an ``and`` tree whose leaves
are all ``PreconditionItem`` — the only shape for which the
``p_<sorted vids>`` key alone identifies the gate. Called on
an already ``ParExpr``-stripped tree (see
:meth:`_strip_par_expr`), so grouping parentheses never affect
this check.
"""
if not isinstance(node, dict):
return False
class_name = node.get("class_name")
if class_name == "PreconditionItem":
return True
return (
class_name == "BinOp"
and node.get("op") == "and"
and cls._is_conjunction_of_items(node.get("left"))
and cls._is_conjunction_of_items(node.get("right"))
)
@classmethod
def _strip_par_expr(cls, node: Any) -> Any:
"""Return a copy of *node* with every ``ParExpr`` unwrapped.
Grouping parentheses change how a gate is reconstructed
(#379), not what it means, so they must not change its
identity (key/version_id) either — otherwise a validation
whose source text happens to add non-meaningful parentheses
around an otherwise-identical gate would stop merging with
one that doesn't. Only used to compute that identity; the
emitted ``entry["ast"]`` keeps the real ``ParExpr`` nodes.
"""
if isinstance(node, dict):
if node.get("class_name") == "ParExpr":
return cls._strip_par_expr(node.get("expression"))
return {k: cls._strip_par_expr(v) for k, v in node.items()}
if isinstance(node, list):
return [cls._strip_par_expr(item) for item in node]
return node
@classmethod
def _strip_precondition_item_vids(cls, node: Any) -> None:
"""Drop the internal ``variable_vid`` field from every item node.
``variable_vid`` is added by :meth:`_precondition_item` so key
derivation groups entries the same way ``p_<vid>`` used to. The
engine's ``PreconditionItem`` only carries ``variable_id`` and
``variable_code``, so the field is stripped once the key has
been built.
"""
if isinstance(node, dict):
if node.get("class_name") == "PreconditionItem":
node.pop("variable_vid", None)
for value in node.values():
if isinstance(value, (dict, list)):
cls._strip_precondition_item_vids(value)
elif isinstance(node, list):
for item in node:
if isinstance(item, (dict, list)):
cls._strip_precondition_item_vids(item)
@classmethod
def _build_precondition_entry_from_ast(
cls,
gate_ast: Dict[str, Any],
validation_codes: List[str],
provided_code: Optional[str] = None,
provided_version_id: Optional[int] = None,
) -> Tuple[str, Dict[str, Any]]:
"""Assemble a ``preconditions[key]`` entry from an emitted gate AST.
The key must identify the gate's meaning, because
:meth:`_merge_precondition_entry` unions the ``affected_operations``
of entries that share a key under the first entry's ``ast``. For
the shape the regex path used to emit — filing indicators joined
by ``and`` — the key stays ``p_<sorted vids>``, so those gates
(all of them on the dictionaries shipped so far) are unchanged
and keep merging as before. Any other shape (``or`` / ``xor`` /
``not``, a parameter, a literal) appends a CRC-32 of the canonical
AST — ``p_<sorted vids>_<crc>``, or ``p_<crc>`` when the gate has
no filing indicator — so ``{v_A} or {v_B}`` can never be merged
into ``{v_A} and {v_B}``. ``version_id`` follows the same rule as
an operation without one: the first vid, or the CRC folded to four
digits. ``provided_code`` / ``provided_version_id`` override both.
Both the shape check and the CRC are computed on a
``ParExpr``-stripped copy (:meth:`_strip_par_expr`): grouping
parentheses change how the gate is reconstructed, not what it
means, so ``({v_A} and {v_B})`` must key and merge exactly like
``{v_A} and {v_B}``. The returned ``ast`` keeps the real
``ParExpr`` nodes.
"""
item_vids = sorted(cls._collect_precondition_item_vids(gate_ast))
# Strip the internal ``variable_vid`` field before hashing and
# emitting; the engine's ``PreconditionItem`` does not carry it.
cls._strip_precondition_item_vids(gate_ast)
vids_part = "_".join(str(v) for v in item_vids)
identity_ast = cls._strip_par_expr(gate_ast)
if cls._is_conjunction_of_items(identity_ast):
default_key = f"p_{vids_part}"
default_version_id = item_vids[0]
else:
digest = zlib.crc32(
json.dumps(identity_ast, sort_keys=True).encode("utf-8")
)
default_key = (
f"p_{vids_part}_{digest:08x}"
if item_vids
else f"p_{digest:08x}"
)
default_version_id = item_vids[0] if item_vids else digest % 10000
code = provided_code if provided_code is not None else default_key
version_id = (
provided_version_id
if provided_version_id is not None
else default_version_id
)
return code, {
"ast": gate_ast,
"affected_operations": list(validation_codes),
"version_id": version_id,
"code": code,
}
@classmethod
def _collect_precondition_item_vids(cls, node: Any) -> List[int]:
"""Return the ``variable_vid`` of every ``PreconditionItem`` node.
The pre-existing ``p_<vid>`` naming keys entries by
``variable_vid``, so gates that name the same filing indicator
collapse to the same entry.
"""
vids: List[int] = []
if isinstance(node, dict):
if node.get("class_name") == "PreconditionItem":
vid = node.get("variable_vid")
if isinstance(vid, int):
vids.append(vid)
for value in node.values():
if isinstance(value, (dict, list)):
vids.extend(cls._collect_precondition_item_vids(value))
elif isinstance(node, list):
for item in node:
if isinstance(item, (dict, list)):
vids.extend(cls._collect_precondition_item_vids(item))
return vids
# ------------------------------------------------------------------ #
# AST helpers
# ------------------------------------------------------------------ #
@staticmethod
def _clean_ast_data_entries(ast_dict: Any) -> Any:
"""Strip engine-internal debug fields from ``VarID`` data entries.
Mirrors pydpm's ``_clean_ast_data_entries``: drops
``data_type``, ``cell_code``, ``table_code``, ``table_vid``
from each entry of every ``VarID`` node's ``data`` array.
Operates in-place.
"""
if isinstance(ast_dict, dict):
ASTGeneratorService._strip_varid_data(ast_dict)
for value in ast_dict.values():
if isinstance(value, (dict, list)):
ASTGeneratorService._clean_ast_data_entries(value)
elif isinstance(ast_dict, list):
for item in ast_dict:
if isinstance(item, (dict, list)):
ASTGeneratorService._clean_ast_data_entries(item)
return ast_dict
@staticmethod
def _strip_varid_data(node: Dict[str, Any]) -> None:
"""Drop debug fields from a single ``VarID`` node's ``data`` list."""
if node.get("class_name") != "VarID":
return
data = node.get("data")
if not isinstance(data, list):
return
for entry in data:
if not isinstance(entry, dict):
continue
for name in _DATA_FIELDS_TO_STRIP:
entry.pop(name, None)
@staticmethod
def _extract_referenced_tables(ast_dict: Any) -> set[str]:
"""Walk a serialised AST and return referenced table codes."""
codes: set[str] = set()
def _walk(node: Any) -> None:
if isinstance(node, dict):
if node.get("class_name") == "VarID":
table = node.get("table")
if isinstance(table, str) and table:
codes.add(table)
for value in node.values():
if isinstance(value, (dict, list)):
_walk(value)
elif isinstance(node, list):
for item in node:
if isinstance(item, (dict, list)):
_walk(item)
_walk(ast_dict)
return codes
@staticmethod
def _extract_operand_datapoints(ast_dict: Any) -> Dict[str, str]:
"""Walk a serialised AST and return ``{datapoint: data_type}``.
Every ``VarID`` node carries one ``data`` entry per data point its
cell reference resolves to, each holding the datapoint id (the
variable id) and its scalar type code. Must run *before*
:meth:`_clean_ast_data_entries`, which strips ``data_type``.
A datapoint with no type resolves to ``""`` — the same fallback
:meth:`ScopeCalculatorService._get_module_tables` uses when a
property carries no data type.
"""
datapoints: Dict[str, str] = {}
def _walk(node: Any) -> None:
if isinstance(node, dict):
ASTGeneratorService._collect_varid_datapoints(node, datapoints)
children: Iterable[Any] = node.values()
elif isinstance(node, list):
children = node
else:
return
for child in children:
if isinstance(child, (dict, list)):
_walk(child)
_walk(ast_dict)
return datapoints
@staticmethod
def _collect_varid_datapoints(
node: Dict[str, Any],
into: Dict[str, str],
) -> None:
"""Record one ``VarID`` node's datapoints and types into *into*."""
if node.get("class_name") != "VarID":
return
data = node.get("data")
if not isinstance(data, list):
return
for entry in data:
if not isinstance(entry, dict):
continue
dp = entry.get("datapoint")
if dp is None:
continue
into[str(dp)] = entry.get("data_type") or ""
def _accumulate_parameters(
self,
accumulated: Dict[str, ParameterInfo],
parameters: Iterable[ParameterInfo],
) -> None:
"""Merge one expression's parameters into ``accumulated``, by code.
Consumes the already-deduped ``SemanticResult.parameters`` produced by
the semantic pass rather than re-walking the serialised AST, so there
is a single source of truth for which parameters an expression
references. A parameter binds to a single value across every operation
it co-executes with, so its declared type is intrinsic and must stay
consistent script-wide — the flat registry holds one type per code.
Raises ``SemanticError`` ``3-8`` on a conflicting redeclaration rather
than silently letting one reference win. The merge itself lives in
:func:`~dpmcore.services._parameters.merge_parameters`, shared with
:meth:`~dpmcore.services.semantic.SemanticService.validate`, which
applies the same rule across an expression and its precondition.
"""
merge_parameters(accumulated, parameters)
def _build_precondition_index(
self,
preconditions: List[Union[Tuple[str, List[str]], Dict[str, Any]]],
) -> Dict[str, List[str]]:
"""Map each validation code → unioned precondition variable codes.
Parses each precondition expression once and extracts variable
codes that act as precondition items. Raises ``ValueError`` if
a precondition expression cannot be parsed.
Supports both tuple and dict formats.
"""
index: Dict[str, List[str]] = {}
for precond_spec in preconditions:
precond_expr, validation_codes, _, _ = self._gate_spec_fields(
precond_spec
)
try:
ast = self._syntax.parse(precond_expr)
except Exception as exc:
raise ValueError(
f"Invalid precondition expression {precond_expr!r}: {exc}"
) from exc
codes = self._extract_precondition_codes(ast)
for vc in validation_codes:
merged = index.setdefault(vc, [])
for c in codes:
if c not in merged:
merged.append(c)
return index
@staticmethod
def _extract_precondition_codes(ast: Any) -> List[str]:
"""Return the variable codes referenced by a precondition AST.
Delegates to :func:`~dpmcore.services._precondition_codes.\
extract_precondition_codes`, shared with
:class:`~dpmcore.services.scope_calculator.ScopeCalculatorService`.
Kept as a method so it stays an overridable seam.
"""
return _extract_precondition_codes(ast)
def _build_dependency_info(
self,
scope_pairs: List[
Tuple[
Tuple[str, str],
"ScopeResult",
Dict[str, List[str]],
_OperandRefs,
]
],
primary_module_vid: Optional[int],
release_id: Optional[int],
) -> Optional[Dict[str, Any]]:
"""Build dependency_info from collected scope results.
Aggregates across all expressions: merges
``intra_instance_validations`` and deduplicates
``cross_instance_dependencies`` by module URI set,
appending new ``affected_operations`` to existing entries.
"""
if (
not self._scope_calc
or primary_module_vid is None
or not scope_pairs
):
return None
all_intra: List[str] = []
all_cross: List[Dict[str, Any]] = []
all_dep_modules: Dict[str, Any] = {}
all_scope_results: List["ScopeResult"] = []
# The home-module table set is a per-script constant (the
# primary module never changes across this loop), and computing
# it inside ``detect_cross_module_dependencies`` would repeat a
# per-table variable/open-key fetch on every iteration. Compute
# it once here and thread it through.
home_module_tables: Set[str] = set(
self._scope_calc._get_module_tables(
primary_module_vid, release_id=release_id
).keys()
)
for item, sr, ts, refs in scope_pairs:
all_scope_results.append(sr)
op_code = item[1]
current = self._scope_calc.detect_cross_module_dependencies(
scope_result=sr,
primary_module_vid=primary_module_vid,
operation_code=op_code,
release_id=release_id,
time_shifts=ts,
compute_alternative_deps=False,
referenced_variables=refs.variables,
referenced_tables=refs.tables,
home_module_tables=home_module_tables,
)
all_intra.extend(current.get("intra_instance_validations", []))
self._merge_cross_deps(
all_cross,
current.get("cross_instance_dependencies", []),
)
self._merge_dep_modules(
all_dep_modules,
current.get("dependency_modules", {}),
)
# Restrict alternatives to the script's genuine dependency modules
# so the groups can never name a module absent from
# ``dependency_modules`` (#202 dangling references).
alt_deps = self._scope_calc.detect_alternative_dependencies(
scope_results=all_scope_results,
primary_module_vid=primary_module_vid,
release_id=release_id,
valid_module_uris=set(all_dep_modules),
)
deduped_intra: List[str] = list(dict.fromkeys(all_intra))
return {
"dependency_information": {
"intra_instance_validations": deduped_intra,
"cross_instance_dependencies": all_cross,
"alternative_dependencies": alt_deps,
},
"dependency_modules": all_dep_modules,
}
@staticmethod
def _merge_cross_deps(
existing: List[Dict[str, Any]],
new: List[Dict[str, Any]],
) -> None:
"""Merge *new* cross-instance deps into *existing*.
Deduplicates by the set of ``(module URI, reference period)``
pairs. When a duplicate is found, its ``affected_operations``
are merged instead.
The reference period is part of the key because two operations
can need the *same* module at *different* instances — one at
``T``, another at ``T-1Q`` (#325 makes a non-default period
reachable for the home module). Keying on the URI alone merged
them into a single entry whose period was whichever operation
came first, silently sending the other to the wrong instance.
"""
def _uri_key(dep: Dict[str, Any]) -> Tuple[Tuple[str, str], ...]:
modules = dep.get("modules", [])
return tuple(
sorted(
(m.get("URI", ""), m.get("ref_period", ""))
if isinstance(m, dict)
else (str(m), "")
for m in modules
)
)
seen = {_uri_key(d) for d in existing}
for dep in new:
key = _uri_key(dep)
if key not in seen:
existing.append(dep)
seen.add(key)
else:
for ex in existing:
if _uri_key(ex) == key:
ops = ex.setdefault("affected_operations", [])
for op in dep.get("affected_operations", []):
if op not in ops:
ops.append(op)
break
@staticmethod
def _merge_dep_modules(
existing: Dict[str, Any],
new: Dict[str, Any],
) -> None:
"""Merge *new* dependency_modules into *existing*.
Avoids table duplicates within each module URI. Two operations
can reference different cells of the same dependency table, and
each declares only the datapoints it uses (#250), so a repeated
table unions its ``variables`` rather than keeping the first.
"""
for uri, data in new.items():
if uri not in existing:
existing[uri] = data
continue
tables = existing[uri].setdefault("tables", {})
for tbl, tbl_data in data.get("tables", {}).items():
if tbl not in tables:
tables[tbl] = tbl_data
continue
merged_vars = dict(tables[tbl].get("variables", {}))
merged_vars.update(tbl_data.get("variables", {}))
tables[tbl] = {**tables[tbl], "variables": merged_vars}
existing[uri].setdefault("variables", {}).update(
data.get("variables", {})
)
@staticmethod
def _to_ref_period(internal: str) -> str:
"""Render a ``t(+|-)<indicator><n>`` marker as a reference period.
The declared period is the *opposite* of the shift the expression
asks for. ``time_shift`` moves the operand's reference period by
``shift_number``, so the instance whose shifted operand lines up
with the one being reported is the one at ``T - shift_number``:
``time_shift(x, A, 1, refPeriod)`` needs ``T-1A`` and
``time_shift(x, Q, -1, refPeriod)`` needs ``T+1Q``. The marker is
built with that inversion already applied (see
:meth:`_extract_time_shifts`), so this only reorders it.
"""
if internal.startswith(("t+", "t-")):
sign = internal[1]
ind = internal[2]
num = internal[3:]
return f"T{sign}{num}{ind}"
return "T"
@staticmethod
def _literal_shift(node: Any) -> Tuple[int, Optional[int]]:
"""Resolve a ``shift_number`` expression to ``(sign, magnitude)``.
``sign`` is the direction the expression asks for (``1`` or
``-1``) and ``magnitude`` its size, or ``None`` when the shift
number is not an integer literal — a reference or a computed
expression, whose size no reference period can name.
The same written literal reaches this in several shapes, and
reading only the two obvious ones mis-declared the rest: ``+1``
is a ``UnaryOp``, so it lost its size and rendered ``T-nQ``;
``( -1 )`` is a ``ParExpr`` around one, so it lost its direction
too; and ``(-1)`` without the spaces is lexed as a *negative*
``Constant``, which rendered the sign twice (``T--1Q``).
"""
from dpmcore.dpm_xl.ast.nodes import Constant, ParExpr, UnaryOp
sign = 1
while True:
if isinstance(node, ParExpr):
node = node.expression
elif isinstance(node, UnaryOp) and node.op in ("+", "-"):
if node.op == "-":
sign = -sign
node = node.operand
else:
break
if not isinstance(node, Constant):
return sign, None
try:
shift = sign * int(node.value)
except (TypeError, ValueError):
return sign, None
return (-1 if shift < 0 else 1), abs(shift)
def _prepare_expression(
self,
semantic: SemanticService,
expr: str,
release_id: int,
gate_failure: Optional[str] = None,
) -> _PreparedExpression:
"""Validate *expr* and read the reference periods it needs.
``validate()`` runs the per-expression scope check: a parameter
referenced here must not clash with the declared type of a
co-scoped operation already persisted in the DB (raises 3-8).
``_accumulate_parameters`` in the caller is complementary — it
catches conflicts between two expressions in the same script.
``gate_failure`` is the reason the operation's precondition
cannot be evaluated by the engine, when there is one (see
``_unsupported_gate_operations``); such an operation is skipped
before validation, with that reason as its error.
"""
if gate_failure is not None:
return _PreparedExpression(error=gate_failure)
result = semantic.validate(expr, release_id=release_id)
if not result.is_valid:
return _PreparedExpression(
error=result.error_message or "",
)
ast = semantic.ast
try:
return _PreparedExpression(
result=result,
ast=ast,
ts=self._extract_time_shifts(ast),
)
except SemanticError as exc:
return _PreparedExpression(error=str(exc))
@staticmethod
def _shift_marker(node: Any) -> str:
"""Return the ``t(+|-)<indicator><n>`` marker for a time shift.
The marker encodes the *declared* period, which inverts the
shift the expression asks for: ``time_shift(x, A, 1, refPeriod)``
reads the instance at ``T-1A`` and
``time_shift(x, Q, -1, refPeriod)`` the one at ``T+1Q``. A shift
of ``0`` is no shift at all, so it keeps the plain marker.
Raises:
SemanticError: 4-7-5, when the shift number is not an
integer literal. Such a shift used
to render as ``T-nQ``, which names no resolvable
instance and so silently sent the operation to an
instance the engine cannot load (#326).
"""
sign, magnitude = ASTGeneratorService._literal_shift(node.shift_number)
if magnitude is None:
raise SemanticError("4-7-5")
if magnitude == 0:
return "t"
# ``magnitude`` is unsigned, so the sign is carried by the marker
# alone — a ``Constant`` holding a negative value would otherwise
# render two signs (``t-Q-1``).
marker = "-" if sign > 0 else "+"
return f"t{marker}{node.period_indicator}{magnitude}"
@staticmethod
def _extract_time_shifts(ast: Any) -> Dict[str, List[str]]:
"""Extract the reference periods each table is referenced at.
Returns a mapping of table codes to the sorted, de-duplicated
reference periods that table is read at, e.g.
``{"C_01.00": ["T", "T-1Q"]}``.
One period per table is not enough (#326): a table read both
plain and shifted, or shifted twice by different amounts, needs
one instance per period, and keeping a single period collapsed
the rest into whichever the visitor reached last. Unshifted
references are recorded as ``"T"`` too, so a module read at both
the current and a shifted instance is visible to the caller
instead of being declared at the shifted one alone.
Raises:
SemanticError: 4-7-5, propagated from
:meth:`_shift_marker` for a non-literal shift number.
"""
from dpmcore.dpm_xl.ast.template import ASTTemplate
time_shifts: Dict[str, Set[str]] = {}
current_period = ["t"]
class _Extractor(ASTTemplate):
def visit_AnnualiseOp(self, node: Any) -> None:
self.visit(node.operand)
def visit_TimeShiftOp(self, node: Any) -> None:
prev = current_period[0]
current_period[0] = ASTGeneratorService._shift_marker(node)
self.visit(node.operand)
current_period[0] = prev
def visit_VarID(self, node: Any) -> None:
if node.table:
time_shifts.setdefault(node.table, set()).add(
current_period[0]
)
def visit__DbSubClauseOp(self, node: Any) -> None:
# db_ast.py's stand-in for a ``sub`` clause (dpmcore#364)
# — not one of ASTTemplate's known node types, so its
# generic_visit would raise NotImplementedError instead
# of walking past it.
self.visit(node.operand)
self.visit(node.condition)
try:
_Extractor().visit(ast)
except SemanticError:
# A shift whose period cannot be declared is the caller's
# decision to act on, not something to swallow into an
# empty mapping.
raise
except Exception:
logger.exception(
"Failed to extract time shifts; returning an empty mapping.",
)
return {}
return {
table: sorted(
{ASTGeneratorService._to_ref_period(p) for p in periods}
)
for table, periods in time_shifts.items()
}
__all__ = ["ASTGeneratorService"]