| |
| |
|
|
| import os |
| import json |
| import threading |
| from datetime import datetime, timezone |
|
|
| class AsyncEventBuffer: |
| def __init__(self, buffer_path=r"C:\Users\leonk\Documents\AI_Projects\ai_workspace\telemetry\event_buffer.json"): |
| self.buffer_path = buffer_path |
| self.memory_buffer = [] |
| self._lock = threading.Lock() |
| os.makedirs(os.path.dirname(self.buffer_path), exist_ok=True) |
|
|
| def log_event(self, event_type: str, payload: dict): |
| event = { |
| "type": event_type, |
| "payload": payload, |
| "timestamp": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") |
| } |
| |
| with self._lock: |
| self.memory_buffer.append(event) |
| |
| |
| if event_type in ["ANOMALY_DETECTED", "SECURITY_BREACH", "CRITICAL_SHUTDOWN"]: |
| self._flush_locked([event], mode="a") |
| print(f"[AsyncBuffer]: WAL Triggered -> Critical event '{event_type}' instantly synced to disk [LK].") |
| else: |
| print(f"[AsyncBuffer]: Event '{event_type}' buffered instantly at edge with zero lag.") |
|
|
| def flush_to_persistence(self): |
| with self._lock: |
| if not self.memory_buffer: |
| return |
| events_to_flush = list(self.memory_buffer) |
| self.memory_buffer.clear() |
| self._flush_locked(events_to_flush, mode="w") |
| print(f"[AsyncBuffer]: Flushed {len(events_to_flush)} events to persistence layer asynchronously [LK].") |
|
|
| def _flush_locked(self, events, mode="w"): |
| existing = [] |
| if mode == "a" and os.path.exists(self.buffer_path): |
| try: |
| with open(self.buffer_path, "r", encoding="utf-8") as f: |
| content = f.read().strip() |
| if content: |
| existing = json.loads(content) |
| except Exception: |
| existing = [] |
| |
| if mode == "w": |
| |
| combined = events |
| else: |
| |
| combined = existing + events |
|
|
| with open(self.buffer_path, "w", encoding="utf-8") as f: |
| json.dump(combined, f, indent=4) |
|
|
| if __name__ == "__main__": |
| buf = AsyncEventBuffer() |
| buf.log_event("ANOMALY_DETECTED", {"source": "edge_sensor_01", "threat": "unauthorized_probe"}) |
| buf.log_event("TELEMETRY_SYNC", {"status": "optimal", "temp_c": 32}) |
| buf.flush_to_persistence() |
|
|
