A first attempt at maybe a terrible idea?

Author: bookman-dius

mock/ElectricitySensor.py

@@ -0,0 +1,13 @@
+import random
+
+from MockSensor import MockSensor
+
+
+class ElectricitySensor(MockSensor):
+    """Simulates a magnetic sensor"""
+
+    def __init__(self, mac: str, role=None, update_interval: float = 30.0):
+        super().__init__(mac, role,  update_interval)
+
+    def get_unit(self):
+        return 'w'

mock/MockPlug.py

@@ -0,0 +1,195 @@
+import asyncio
+import datetime
+import json
+import random
+from typing import Optional, List
+
+import numpy as np
+
+from MockSensor import MockSensor
+import logging
+logger = logging.getLogger(__name__)
+
+
+def random_sample_duration():
+    if np.random.rand() < 0.5:
+        return np.random.normal(0.96, 0.00025)
+    else:
+        return np.random.normal(1.04, 0.00025)
+
+def random_sample_power():
+    if np.random.rand() < 0.25:
+        return np.random.normal(58.15, 0.26)
+    else:
+        return np.random.normal(59.8, 0.4)
+
+def random_sample_reactive_current():
+    if np.random.rand() < 0.25:
+        return np.random.normal(.305, 0.0002)
+    else:
+        return np.random.normal(.318, 0.002)
+
+
+class MockPlug(asyncio.DatagramProtocol):
+    """
+    Simulated Plug!
+    Generates both plug messages and relayed sensor messages.
+    """
+
+    def __init__(self, mac: str, gateway_id: str, sensors: Optional[List[MockSensor]] = None):
+        self.mac = mac
+        self.gateway_id = gateway_id
+        self.sensors = sensors or []
+        self.transport = None
+        self.sensor_tasks = []
+        self.subscribers = set()  # Track subscribed clients
+        self.broadcast_lock = asyncio.Lock()  # Prevent concurrent broadcasts
+
+    def connection_made(self, transport):
+        self.transport = transport
+        sock = transport.get_extra_info('socket')
+        logger.info(f"Mock Plug {self.gateway_id} UDP server started on {sock.getsockname()}")
+
+        # Start all sensor tasks
+        for sensor in self.sensors:
+            task = asyncio.create_task(sensor.run(self._handle_sensor_reading))
+            self.sensor_tasks.append(task)
+
+
+        asyncio.create_task(self._send_plug_data())
+
+    def datagram_received(self, data, addr):
+        """Handle incoming UDP packets"""
+        logger.info(f"Received {len(data)} bytes from {addr}")
+
+        # Try to decode as plain text first (for subscribe commands)
+        try:
+            message_str = data.decode('utf-8').strip()
+
+            # Handle subscribe command
+            if message_str.startswith('subscribe('):
+                self.handle_subscribe(message_str, addr)
+                return
+
+            # Try parsing as JSON
+            try:
+                # We need to handle this possibility...we could add better message handling if it matters...
+                message = json.loads(message_str)
+                logger.info(f"Parsed JSON message: {message}")
+                logger.info(f"Gateway {self.gateway_id} received: {message}")
+            except json.JSONDecodeError:
+                logger.warning(f"Received non-JSON text from {addr}: {message_str}")
+
+        except UnicodeDecodeError:
+            logger.warning(f"Could not decode message from {addr}: {data}")
+        except Exception as e:
+            logger.error(f"Error handling message: {e}")
+
+    def handle_subscribe(self, message_str: str, addr: tuple):
+        """
+        Handle subscription requests from clients.
+        Format: subscribe(num) where num is subscriber ID
+        subscribe(0) = disconnect/unsubscribe
+        """
+        logger.info(f"Client {addr} sent: {message_str}")
+
+        # Check if this is a disconnect (subscribe(0))
+        if 'subscribe(0)' in message_str:
+            if addr in self.subscribers:
+                self.subscribers.discard(addr)
+                logger.info(f"Client {addr} unsubscribed. Total subscribers: {len(self.subscribers)}")
+            else:
+                logger.warning(f"Client {addr} tried to unsubscribe but was not subscribed")
+        else:
+            # Add to subscribers list
+            self.subscribers.add(addr)
+            logger.info(f"Client {addr} subscribed. Total subscribers: {len(self.subscribers)}")
+
+
+    async def _handle_sensor_reading(self, reading: dict):
+        """
+        Internal callback for when sensors generate readings.
+        For now, it just replays it back, we could do more...
+        """
+        async with self.broadcast_lock:
+            self.broadcast_message(reading)
+            logger.debug(f"Relayed sensor reading: {reading['mac']}")
+
+    async def _send_plug_data(self):
+        """Send periodic plug status messages"""
+        while True:
+            duration = random_sample_duration()
+            await asyncio.sleep(duration)
+            t = datetime.datetime.now(datetime.timezone.utc).timestamp()
+            v = np.random.normal(240, 1.2)
+            p = random_sample_power()
+            active_current = p / v - random.random() / 1000
+            reactive_current = random_sample_reactive_current()
+            current = np.sqrt(active_current ** 2 + reactive_current ** 2) + 0.016 + (random.random() - 0.5) / 1000
+            message = {
+                'reactive_current': reactive_current,
+                'type': 'instant_power',
+                'summation_start': 1760615946.173936,
+                'count': random.randint(12, 13),
+                'duration': duration,
+                'role': 'appliance',
+                'power': p,
+                'unit': 'W',
+                'device': 'plug',
+                'source': 'BLE',
+                'active_current': active_current,
+                'mac': self.mac,
+                'voltage': v,
+                'starttime': t,
+                'current': current,
+                'borrowed_summation': 0,
+                'summation': 59.5649065 * t - 1.04883909e+11
+            }
+            async with self.broadcast_lock:
+                self.broadcast_message(message)
+                logger.info(f"Plug {self.mac} sent data!")
+
+    def send_message(self, message: dict, addr: tuple):
+        """Send a message to a specific address"""
+        if self.transport:
+            data = json.dumps(message).encode('utf-8')
+            self.transport.sendto(data, addr)
+            logger.debug(f"Sent message to {addr}: {message}")
+
+    def broadcast_message(self, message: dict, port: int = 49476):
+        """Broadcast a message to all subscribed clients (or network if no subscribers)"""
+        logger.debug(f"broadcast_message ENTRY - subscribers: {len(self.subscribers)}")
+        if self.transport:
+            data = json.dumps(message).encode('utf-8')
+
+            if self.subscribers:
+                # Send to each subscriber
+                dead_subscribers = set()
+                for addr in self.subscribers:
+                    try:
+                        logger.debug(f"Sending to {addr}...")
+                        self.transport.sendto(data, addr)
+                        logger.debug(f"Sent to {addr} OK")
+                    except Exception as e:
+                        logger.error(f"Failed to send to {addr}: {e}")
+                        dead_subscribers.add(addr)
+
+                # Clean up dead subscribers
+                if dead_subscribers:
+                    self.subscribers -= dead_subscribers
+                    logger.warning(f"Removed {len(dead_subscribers)} dead subscribers")
+
+                logger.debug(f"Sent message to {len(self.subscribers)} subscribers")
+            else:
+                # No subscribers yet, broadcast to network
+                logger.debug("Broadcasting to network (no subscribers)")
+                self.transport.sendto(data, ('<broadcast>', port))
+                logger.debug("Broadcast to network complete")
+        else:
+            logger.error("broadcast_message called but transport is None!")
+        logger.debug("broadcast_message EXIT")
+
+    def stop(self):
+        """Clean up sensor tasks"""
+        for task in self.sensor_tasks:
+            task.cancel()

mock/MockPlugUDPService.py

@@ -0,0 +1,137 @@
+import asyncio
+import logging
+import socket
+from typing import Optional, List
+
+from zeroconf import Zeroconf, ServiceInfo
+
+from MockPlug import MockPlug
+from MockSensor import MockSensor
+
+logger = logging.getLogger(__name__)
+
+class MockPlugUDPService:
+    """
+    Main service that manages zeroconf advertisement and UDP protocol.
+    """
+
+    def __init__(
+        self,
+        mac: str = "mock0plug123",
+        gateway_id: str = "Powersensor-gateway-mock0plug123-civet",
+        port: int = 49476,
+        sensors: Optional[List[MockSensor]] = None,
+        protocol_class: type = MockPlug,
+        properties: Optional[dict] = None
+    ):
+        self.mac = mac
+        self.gateway_id = gateway_id
+        self.port = port
+        self.sensors = sensors or []
+        self.protocol_class = protocol_class
+        self.properties = properties or {}
+
+        self.zeroconf = None
+        self.service_info = None
+        self.transport = None
+        self.protocol = None
+
+    def _get_local_ip(self) -> str:
+        """Get the local IP address"""
+        try:
+            # Create a socket to determine local IP
+            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            s.connect(("8.8.8.8", 80))
+            ip = s.getsockname()[0]
+            s.close()
+            return ip
+        except Exception:
+            return "127.0.0.1"
+
+    async def start(self):
+        """Start the mock gateway service"""
+        logger.info(f"Starting Mock Gateway Service: {self.gateway_id}")
+        logger.info(f"Configured with {len(self.sensors)} sensors")
+
+        # Start UDP server with sensors
+        loop = asyncio.get_running_loop()
+        self.transport, self.protocol = await loop.create_datagram_endpoint(
+            lambda: self.protocol_class(self.mac,self.gateway_id, self.sensors),
+            local_addr=('0.0.0.0', self.port)
+        )
+
+        # Set up Zeroconf advertisement
+        await self._setup_zeroconf()
+
+        logger.info(f"Mock Gateway {self.gateway_id} is now running and discoverable")
+
+    async def _setup_zeroconf(self):
+        """Set up zeroconf service advertisement (IPv4 only)"""
+        local_ip = self._get_local_ip()
+
+        # Prepare properties
+        props = {
+            'gateway_id': self.gateway_id,
+            'version': '1.0',
+            'sensor_count': str(len(self.sensors)),
+            **self.properties
+        }
+
+        # Create service info
+        service_name = f"{self.gateway_id}._powersensor._udp.local."
+
+        self.service_info = ServiceInfo(
+            "_powersensor._udp.local.",
+            service_name,
+            addresses=[socket.inet_aton(local_ip)],
+            port=self.port,
+            properties=props,
+            server=f"{self.gateway_id}.local."
+        )
+
+        # Register service in a thread (zeroconf is blocking) - IPv4 only
+        self.zeroconf = Zeroconf(interfaces=[local_ip])
+        await asyncio.get_running_loop().run_in_executor(
+            None,
+            self.zeroconf.register_service,
+            self.service_info
+        )
+
+        logger.info(f"Zeroconf service registered (IPv4): {service_name} at {local_ip}:{self.port}")
+
+    async def stop(self):
+        """Stop the mock gateway service"""
+        logger.info(f"Stopping Mock Gateway {self.gateway_id}")
+
+        # Stop protocol (cancels sensor tasks)
+        if self.protocol:
+            self.protocol.stop()
+
+        # Unregister zeroconf with proper timeout
+        if self.zeroconf and self.service_info:
+            try:
+                await asyncio.wait_for(
+                    asyncio.get_running_loop().run_in_executor(
+                        None,
+                        self._unregister_zeroconf
+                    ),
+                    timeout=2.0
+                )
+            except asyncio.TimeoutError:
+                logger.warning("Zeroconf unregister timed out")
+            except Exception as e:
+                logger.warning(f"Error unregistering zeroconf: {e}")
+
+        # Close UDP transport
+        if self.transport:
+            self.transport.close()
+
+        logger.info(f"Mock Gateway {self.gateway_id} stopped")
+
+    def _unregister_zeroconf(self):
+        """Synchronous helper to unregister zeroconf"""
+        try:
+            self.zeroconf.unregister_service(self.service_info)
+            self.zeroconf.close()
+        except Exception as e:
+            logger.warning(f"Error in zeroconf cleanup: {e}")