Python Coeus/RTT module

# dimensional_core.py
 
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional
import uuid
 
 
@dataclass
class DimensionEvent:
    id: str
    timestamp: datetime
    entity_id: str
    state: str
    meta: Dict[str, Any] = field(default_factory=dict)
 
 
@dataclass
class DimensionalSnapshot:
    dimension_name: str
    total_events: int
    entities: List[str]
    states: List[str]
    last_event: Optional[DimensionEvent]
 
 
class DimensionalCore:
    """
    Dimensional core:
      - tracks entities and their state changes
      - provides structural snapshots for RTT-style reasoning
    """
 
    def __init__(self, dimension_name: str = "coeus"):
        self.dimension_name = dimension_name
        self._events: List[DimensionEvent] = []
 
    def observe(
        self,
        entity_id: str,
        state: str,
        meta: Optional[Dict[str, Any]] = None,
    ) -> DimensionEvent:
        event = DimensionEvent(
            id=str(uuid.uuid4()),
            timestamp=datetime.utcnow(),
            entity_id=entity_id,
            state=state,
            meta=meta or {},
        )
        self._events.append(event)
        return event
 
    def snapshot(self) -> DimensionalSnapshot:
        total = len(self._events)
        entities = sorted({e.entity_id for e in self._events})
        states = sorted({e.state for e in self._events})
        last_event = self._events[-1] if self._events else None
 
        return DimensionalSnapshot(
            dimension_name=self.dimension_name,
            total_events=total,
            entities=entities,
            states=states,
            last_event=last_event,
        )
 
    def debug_print(self) -> None:
        snap = self.snapshot()
        print(f"[DimensionalCore] dimension={snap.dimension_name}")
        print(f"  total_events: {snap.total_events}")
        print(f"  entities    : {', '.join(snap.entities) or '-'}")
        print(f"  states      : {', '.join(snap.states) or '-'}")
        if snap.last_event:
            le = snap.last_event
            print(f"  last_event  : {le.entity_id} -> {le.state} @ {le.timestamp.isoformat()}Z")

# resonance_core.py
 
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List
 
from .dimensional_core import DimensionalCore, DimensionalSnapshot
 
 
@dataclass
class ResonanceView:
    context: str
    dimensions: Dict[str, DimensionalSnapshot] = field(default_factory=dict)
 
 
class ResonanceCore:
    """
    Resonance structural awareness core:
      - manages multiple DimensionalCore instances
      - provides a unified structural view
    """
 
    def __init__(self, context: str = "coeus-protocol"):
        self.context = context
        self._dimensions: Dict[str, DimensionalCore] = {}
 
    def dimension(self, name: str) -> DimensionalCore:
        if name not in self._dimensions:
            self._dimensions[name] = DimensionalCore(dimension_name=name)
        return self._dimensions[name]
 
    def snapshot(self) -> ResonanceView:
        snaps = {
            name: dim.snapshot()
            for name, dim in self._dimensions.items()
        }
        return ResonanceView(context=self.context, dimensions=snaps)
 
    def debug_print(self) -> None:
        view = self.snapshot()
        print(f"[ResonanceCore] context={view.context}")
        if not view.dimensions:
            print("  (no dimensions)")
            return
        for name, snap in view.dimensions.items():
            print(f"  dimension: {name}")
            print(f"    total_events: {snap.total_events}")
            print(f"    entities    : {', '.join(snap.entities) or '-'}")
            print(f"    states      : {', '.join(snap.states) or '-'}")

# protocol_adapter.py
 
from __future__ import annotations
from typing import Any, Dict
 
from .resonance_core import ResonanceCore
 
 
class CoeusRTTAdapter:
    """
    Coeus + RTT integration layer.
 
    Intended usage:
      - attach to your Coeus protocol handler
      - call record_* methods at key lifecycle points
      - use resonance.debug_print() or snapshot() for structural introspection
    """
 
    def __init__(self, context: str = "coeus-protocol"):
        self.resonance = ResonanceCore(context=context)
 
    def record_connection(self, client_id: str, state: str, meta: Dict[str, Any] | None = None) -> None:
        dim = self.resonance.dimension("connections")
        dim.observe(entity_id=client_id, state=state, meta=meta)
 
    def record_message(self, message_id: str, state: str, meta: Dict[str, Any] | None = None) -> None:
        dim = self.resonance.dimension("messages")
        dim.observe(entity_id=message_id, state=state, meta=meta)
 
    def record_entity(self, entity_id: str, state: str, meta: Dict[str, Any] | None = None) -> None:
        dim = self.resonance.dimension("entities")
        dim.observe(entity_id=entity_id, state=state, meta=meta)
 
    def structural_view(self):
        """Return a full resonance snapshot for external tooling / UI."""
        return self.resonance.snapshot()
 
    def debug_print(self) -> None:
        self.resonance.debug_print()