39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
"""Minimal EnergiBridge-like canonical bridge.
|
|
|
|
This module translates between venue payloads and a canonical,
|
|
off-chain TradeCipher representation using simple dataclass-based
|
|
contracts.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict
|
|
|
|
from .contracts import LocalTrade
|
|
|
|
|
|
class EnergiBridge:
|
|
"""A tiny translator between venue payloads and canonical objects."""
|
|
|
|
def to_canonical(self, obj: Any) -> Dict[str, Any]:
|
|
# For this MVP, we assume objects expose a to_json or __dict__.
|
|
if hasattr(obj, "to_json"):
|
|
payload = obj.to_json()
|
|
return {"type": obj.__class__.__name__, "payload": payload}
|
|
elif hasattr(obj, "__dict__"):
|
|
return {"type": obj.__class__.__name__, "payload": obj.__dict__}
|
|
else:
|
|
raise TypeError("Unsupported object for canonicalization")
|
|
|
|
def from_canonical(self, data: Dict[str, Any]) -> LocalTrade:
|
|
t = data.get("type")
|
|
payload = data.get("payload")
|
|
# Support LocalTrade payloads serialized as JSON strings
|
|
if t == "LocalTrade" and isinstance(payload, str):
|
|
return LocalTrade.from_json(payload)
|
|
if isinstance(payload, dict) and t == "LocalTrade":
|
|
return LocalTrade(**payload)
|
|
raise ValueError("Unsupported canonical payload for LocalTrade")
|
|
|
|
|
|
__all__ = ["EnergiBridge"]
|