From 137a594abe0d03999ad515f6c5c7fffbdacf8b17 Mon Sep 17 00:00:00 2001
From: fabiorafaelcoutada <fabiorafaelcoutada@fedora>
Date: Mon, 6 Jul 2026 02:27:00 +0100
Subject: [PATCH] feat(lab-gateway): PPK2 simulation mode for offline
 development/CI

---
 infra/lab-gateway/ppk2_manager.py | 71 ++++++++++++++++++++++++++++++-
 1 file changed, 69 insertions(+), 2 deletions(-)

diff --git a/infra/lab-gateway/ppk2_manager.py b/infra/lab-gateway/ppk2_manager.py
index bc268e2f..a3ecc20d 100644
--- a/infra/lab-gateway/ppk2_manager.py
+++ b/infra/lab-gateway/ppk2_manager.py
@@ -5,6 +5,9 @@ Uses ppk2-api for serial communication with the PPK2 device.
 Provides real-time power measurements via WebSocket-compatible async generator.
 """
 
+import math
+import os
+import random
 import time
 import threading
 import json
@@ -19,6 +22,63 @@ except ImportError:
     print("[PPK2] ppk2-api not installed — PPK2 features disabled")
 
 
+SIMULATION = os.environ.get("PPK2_SIMULATION", "0") == "1"
+
+
+class SimulatedPPK2:
+    """Software PPK2 stand-in for development and CI."""
+
+    def __init__(self):
+        self.mode = "AMPERE_MODE"
+        self.current_vdd = 3300
+        self._running = False
+        self._start_time = time.time()
+        self._lock = threading.Lock()
+
+    def use_ampere_meter(self):
+        self.mode = "AMPERE_MODE"
+
+    def use_source_meter(self):
+        self.mode = "SOURCE_MODE"
+
+    def set_source_voltage(self, mv: int):
+        with self._lock:
+            self.current_vdd = mv
+
+    def toggle_DUT_power(self, state: str):
+        pass
+
+    def start_measuring(self):
+        self._running = True
+
+    def stop_measuring(self):
+        self._running = False
+
+    def get_data(self) -> bytes:
+        if not self._running:
+            return b""
+        with self._lock:
+            vdd = self.current_vdd
+        t = time.time() - self._start_time
+        base = 25000 + (4200 - vdd) * 5 + random.uniform(-200, 200)
+        if int(t * 2) % 10 == 0:
+            base += 80000 + random.uniform(-1000, 1000)
+        analog = [int(base + math.sin(i * 0.1) * 50) for i in range(100)]
+        raw = b"".join(int((a & 0x3FFF) | (0 << 14)).to_bytes(4, "little") for a in analog)
+        return raw
+
+    def get_samples(self, raw: bytes):
+        """Parse packed samples into (analog_values, [digital_values])."""
+        analog = []
+        digital = []
+        for i in range(0, len(raw), 4):
+            sample = int.from_bytes(raw[i : i + 4], "little")
+            analog.append(float(sample & 0x3FFF))
+            digital.append((sample >> 24) & 0xFF)
+        return (analog, digital)
+
+
+
 class PPK2Manager:
     """Manages the Nordic PPK2 power profiler."""
 
@@ -33,9 +93,15 @@ class PPK2Manager:
         self._voltage_mv = 3300
         self._sample_buffer: deque = deque(maxlen=10000)
         self._stats = {"avg_ua": 0, "peak_ua": 0, "min_ua": 0, "total_samples": 0, "energy_uwh": 0}
+        self._simulation = SIMULATION or not HAS_PPK2
 
     def connect(self) -> bool:
-        """Connect to the PPK2 device."""
+        """Connect to the PPK2 device (or simulation if configured)."""
+        if self._simulation:
+            self.ppk2 = SimulatedPPK2()
+            print("[PPK2] Using simulation backend")
+            return True
+
         if not HAS_PPK2:
             return False
 
@@ -179,5 +245,6 @@ class PPK2Manager:
             "voltage_mv": self._voltage_mv,
             "measuring": self._running,
             "stats": self._stats,
-            "available": HAS_PPK2,
+            "available": HAS_PPK2 or self._simulation,
+            "simulation": self._simulation,
         }