File size: 2,713 Bytes
7c5e40e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """Canonical, type-checked graph-program abstract syntax trees."""
from __future__ import annotations
from dataclasses import dataclass
from typing import TypeAlias
from strata.modeling.compose.types import GraphOperator, SemanticType, signature
class ProgramTypeError(TypeError):
pass
@dataclass(frozen=True, slots=True)
class AnchorRef:
"""Episode-local copied anchor; identities never enter the operator vocabulary."""
index: int
value_type: SemanticType
def __post_init__(self) -> None:
if self.index < 0:
raise ValueError("anchor index must be non-negative")
@property
def output_type(self) -> SemanticType:
return self.value_type
@property
def depth(self) -> int:
return 0
def canonical(self) -> tuple[str, int, str]:
return ("anchor", self.index, self.value_type.value)
@dataclass(frozen=True, slots=True)
class Apply:
operator: GraphOperator
argument: "ProgramNode"
def __post_init__(self) -> None:
operator = GraphOperator(self.operator)
object.__setattr__(self, "operator", operator)
expected = signature(operator).input_type
actual = self.argument.output_type
if actual is not expected:
raise ProgramTypeError(
f"{operator.value} requires {expected.value}, got {actual.value}"
)
@property
def output_type(self) -> SemanticType:
return signature(self.operator).output_type
@property
def depth(self) -> int:
return self.argument.depth + 1
def canonical(self) -> tuple[str, tuple]:
return (self.operator.value, self.argument.canonical())
ProgramNode: TypeAlias = AnchorRef | Apply
def apply_chain(anchor: AnchorRef, operators: tuple[GraphOperator, ...]) -> ProgramNode:
program: ProgramNode = anchor
for operator in operators:
program = Apply(operator, program)
return program
def operator_chain(program: ProgramNode) -> tuple[GraphOperator, ...]:
operators: list[GraphOperator] = []
node = program
while isinstance(node, Apply):
operators.append(node.operator)
node = node.argument
operators.reverse()
return tuple(operators)
def anchor_ref(program: ProgramNode) -> AnchorRef:
node = program
while isinstance(node, Apply):
node = node.argument
return node
def canonicalize(program: ProgramNode) -> tuple:
"""Return an identity-free structural key for equality and hashing."""
return program.canonical()
__all__ = [
"AnchorRef",
"Apply",
"ProgramNode",
"ProgramTypeError",
"anchor_ref",
"apply_chain",
"canonicalize",
"operator_chain",
]
|