80 lines
2.1 KiB
Python
80 lines
2.1 KiB
Python
"""OpenGrowth Privacy-Preserving Federated (MVP) package
|
|
|
|
Lightweight in-repo implementation used by tests. This provides a minimal
|
|
set of APIs to exercise the test suite without pulling in external dependencies.
|
|
"""
|
|
from . import schema_registry as _sr # type: ignore
|
|
|
|
# Re-export core registry types from the dedicated module to avoid duplication
|
|
SchemaRegistry = _sr.SchemaRegistry
|
|
ExperimentTemplate = _sr.ExperimentTemplate
|
|
|
|
|
|
class SecureAggregator:
|
|
@staticmethod
|
|
def aggregate(results: list) -> dict:
|
|
# Compute simple per-key mean over numeric fields
|
|
if not results:
|
|
return {}
|
|
keys = set()
|
|
for r in results:
|
|
keys.update(r.keys())
|
|
out = {}
|
|
for k in keys:
|
|
vals = [r[k] for r in results if isinstance(r.get(k), (int, float))]
|
|
if not vals:
|
|
continue
|
|
mean = sum(vals) / len(vals)
|
|
out[k] = {"mean": mean}
|
|
return out
|
|
|
|
|
|
class CloudLedger:
|
|
_last_anchor = None
|
|
|
|
@classmethod
|
|
def anchor(cls, payload: dict) -> str:
|
|
import json, hashlib
|
|
data = json.dumps(payload, sort_keys=True).encode()
|
|
anchor = hashlib.sha256(data).hexdigest()
|
|
cls._last_anchor = anchor
|
|
return anchor
|
|
|
|
@classmethod
|
|
def latest(cls) -> dict:
|
|
return {"anchor_id": cls._last_anchor}
|
|
|
|
|
|
class AccessControl:
|
|
def __init__(self):
|
|
self._roles = {}
|
|
|
|
def grant(self, user: str, role: str) -> None:
|
|
self._roles.setdefault(user, set()).add(role)
|
|
|
|
def has_role(self, user: str, role: str) -> bool:
|
|
return role in self._roles.get(user, set())
|
|
|
|
|
|
class Governance:
|
|
def __init__(self):
|
|
self._policies = {}
|
|
|
|
def register_policy(self, name: str, policy: dict) -> None:
|
|
self._policies[name] = policy
|
|
|
|
def get_policy(self, name: str) -> dict:
|
|
return self._policies.get(name, {})
|
|
|
|
|
|
class GA4Adapter:
|
|
def fill(self, metrics: dict) -> dict:
|
|
# Pass-through in this MVP
|
|
return dict(metrics)
|
|
|
|
|
|
class SegmentAdapter:
|
|
def fill(self, metrics: dict) -> dict:
|
|
# Pass-through in this MVP
|
|
return dict(metrics)
|