107 lines
2.6 KiB
Python
107 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
import time
|
|
import json
|
|
import hashlib
|
|
|
|
|
|
def _hash(data: dict) -> str:
|
|
payload = json.dumps(data, sort_keys=True).encode("utf-8")
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
@dataclass
|
|
class LocalPlan:
|
|
plan_id: str
|
|
neighborhood: str
|
|
contents: dict
|
|
version: int = 1
|
|
timestamp: float = field(default_factory=time.time)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"plan_id": self.plan_id,
|
|
"neighborhood": self.neighborhood,
|
|
"contents": self.contents,
|
|
"version": self.version,
|
|
"timestamp": self.timestamp,
|
|
}
|
|
|
|
@staticmethod
|
|
def from_dict(d: dict) -> "LocalPlan":
|
|
return LocalPlan(
|
|
plan_id=d["plan_id"],
|
|
neighborhood=d["neighborhood"],
|
|
contents=d["contents"],
|
|
version=d.get("version", 1),
|
|
timestamp=d.get("timestamp", time.time()),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class SharedSignals:
|
|
signal_id: str
|
|
origin: str
|
|
data: dict
|
|
timestamp: float = field(default_factory=time.time)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"signal_id": self.signal_id,
|
|
"origin": self.origin,
|
|
"data": self.data,
|
|
"timestamp": self.timestamp,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta_id: str
|
|
base_plan_id: str
|
|
updates: dict
|
|
author: str
|
|
timestamp: float = field(default_factory=time.time)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"delta_id": self.delta_id,
|
|
"base_plan_id": self.base_plan_id,
|
|
"updates": self.updates,
|
|
"author": self.author,
|
|
"timestamp": self.timestamp,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class GovernanceLog:
|
|
entries: list = field(default_factory=list)
|
|
|
|
def append(self, action: str, payload: dict, actor: str) -> dict:
|
|
entry = {
|
|
"action": action,
|
|
"payload": payload,
|
|
"actor": actor,
|
|
"timestamp": time.time(),
|
|
"prev_hash": self.entries[-1]["hash"] if self.entries else "0" * 64,
|
|
}
|
|
entry["hash"] = _hash(entry)
|
|
self.entries.append(entry)
|
|
return entry
|
|
|
|
def to_dict(self) -> dict:
|
|
return {"entries": self.entries}
|
|
|
|
|
|
def apply_delta(plan: LocalPlan, delta: PlanDelta) -> LocalPlan:
|
|
new_contents = json.loads(json.dumps(plan.contents)) # deep copy
|
|
for k, v in delta.updates.items():
|
|
new_contents[k] = v
|
|
return LocalPlan(
|
|
plan_id=plan.plan_id,
|
|
neighborhood=plan.neighborhood,
|
|
contents=new_contents,
|
|
version=plan.version + 1,
|
|
timestamp=time.time(),
|
|
)
|