Merge pull request #2 from ManticSic/auto_discover

Auto discover

Author: ManticSic

Dockerfile

@@ -1,7 +1,7 @@
 FROM python:3.7
 
-ENV DEVICE unknown
-ENV INTERVAL 5
+ENV FETCH_INTERVAL 5
+ENV DISCOVER_INTERVAL 30
 ENV LOGSTASH_HOST unknown
 ENV LOGSTASH_PORT 5000
 
@@ -10,8 +10,8 @@ WORKDIR /usr/src/app
 COPY requierments.txt .
 COPY main.py .
 COPY setup setup
+COPY services services
 COPY models models
-COPY collecting collecting
 
 COPY entry.sh .
 RUN chmod +x entry.sh

README.md

@@ -1,6 +1,6 @@
 # Kasa Reporter
 
-Small pyton program to send emeter data of kasa devices to logstash
+Small python program to send emeter data of kasa devices to logstash
 
 ## Build
 
@@ -12,16 +12,15 @@ $ docker build -t kasa-reporter .
 ## Run
 
 ```
-$ docker run --network host --env DEVICE=ip-or-hostname --env LOGSTASH_HOST=ip-or-hostname --name kasa-reporter kasa-reporter
+$ docker run --network host --env LOGSTASH_HOST=ip-or-hostname --name kasa-reporter kasa-reporter
 ```
 
 ## Arguments and environment variables
 
 Argument|Environment Variable|Type|Mandatory|Default|Description
 :---|---:|---:|---:|---:|---:
-`-d`, `--device`|`DEVICE`|String|Yes|n/a|Ip or hostname of the device.
-`-t`, `--type`|n/a|string|No|`plug`|Type of the device. Only the type `plug` is currently supported.
-`-i`, `--interval`|`INTERVAL`|Integer|No|`5`|Interval of requesting emeter data in seconds.
+`--fetch-interval`|`FETCH_INTERVAL`|Integer|No|`5`|Interval of requesting emeter data in seconds.
+`--discover-interval`|`DISCOVER_INTERVAL`|Integer|No|`30`|Interval of discovering devices in network 255.255.255.255.
 `--logstash-host`|`LOGSTASH_HOST`|String|Yes|n/a|Ip or hostname of the logstash server.
 `--logstash-port`|`LOGSTASH_PORT`|Integer|No|`5000`|Port of the logstash server.
 

collecting/kasa.py

@@ -5,8 +5,8 @@ services:
     image: kasa-reporter
     restart: always
     environment:
-      DEVICE: "unknown"
-      INTERVAL: 5
+      FETCH_INTERVAL: 5
+      DISCOVER_INTERVAL: 30
       LOGSTASH_HOST: "unknown"
       LOGSTASH_PORT: 5000
     network_mode: "host"

docker-compose.yml renamed to docker-compose.example.yml

@@ -2,23 +2,21 @@
 import sys
 import setup.log
 import setup.configuration
-
-from time import sleep
-from collecting.kasa import KasaCollector
+from services.discoveryservice import DiscoveryService
+from services.emeterservice import EmeterService
 
 configuration = setup.configuration.get_configuration()
-logger = setup.log.get_logger()
+logger = setup.log.get_logger('main')
+
+discovery_service = DiscoveryService()
+emeter_service = EmeterService(discovery_service)
 
 
 async def main() -> None:
     logger.info('Start fetching data.')
-    collector = KasaCollector(configuration)
-    await collector.setup()
 
-    while True:
-        data = await collector.fetch()
-        logger.info('Successfully fetched data.', data=data.__dict__)
-        sleep(configuration.interval)
+    discovery_service.start()
+    emeter_service.start()
 
 
 if __name__ == '__main__':

main.py

@@ -0,0 +1,8 @@
+from kasa import SmartDevice
+
+
+class SmartDeviceSpare:
+    def __init__(self, device: SmartDevice):
+        self.host = device.host
+        self.device_id = device.device_id
+        self.alias = device.alias

models/SmartDeviceSpare.py

@@ -1,8 +1,7 @@
 class Configuration:
 
     def __init__(self, args):
-        self.device = args.device
-        self.interval = args.interval
-        self.device_type = args.type
+        self.fetch_interval = args.fetch_interval
+        self.discover_interval = args.discover_interval
         self.logstash_host = args.logstash_host
         self.logstash_port = args.logstash_port

models/configuration.py

@@ -0,0 +1,41 @@
+import asyncio
+from threading import Thread
+from typing import Union
+
+
+class CancellationToken:
+    is_cancellation_requested: bool = False
+
+    def request_cancellation(self):
+        self.is_cancellation_requested = True
+
+
+class BackgroundService:
+    cancellation_token: Union[CancellationToken, None] = None
+    thread: Union[Thread, None] = None
+
+    def __init__(self, job):
+        self.job = job
+
+    def start(self):
+        if self.cancellation_token is not None:
+            raise Exception('Service is already started.')
+
+        self.cancellation_token = CancellationToken()
+        self.thread = Thread(target=self.__run)
+        self.thread.start()
+
+    def stop(self):
+        if self.cancellation_token is None:
+            raise Exception('Service is not running.')
+
+        self.cancellation_token.request_cancellation()
+
+    def __run(self):
+        asyncio.run(self.__job())
+
+    async def __job(self):
+        if self.job is None:
+            raise Exception('No job defined.')
+
+        await self.job()

collecting/__init__.py renamed to services/__init__.py

@@ -0,0 +1,34 @@
+from typing import List
+from kasa import SmartDevice
+from kasa import Discover
+from time import sleep
+
+from models.SmartDeviceSpare import SmartDeviceSpare
+from services.backgroundservice import BackgroundService
+from setup.log import get_logger
+
+logger = get_logger('DiscoveryService')
+
+
+class DiscoveryService(BackgroundService):
+    devices: List[SmartDevice] = []
+
+    def __init__(self):
+        super().__init__(self.__job_implementation)
+
+    async def __job_implementation(self):
+        while not self.cancellation_token.is_cancellation_requested:
+            result = await Discover.discover()
+
+            device_dict = dict(filter(lambda device: device[1].has_emeter, result.items()))
+            self.devices = list(map(lambda device: device[1], device_dict.items()))
+
+            devices_spare = list(map(lambda device: SmartDeviceSpare(device), self.devices))
+            logger.info(f'Devices discovered.', devices=devices_spare.__repr__())
+
+            sleep(10)
+
+        self.cancellation_token = None
+
+    def get_devices(self):
+        return self.devices

services/backgroundservice.py

@@ -0,0 +1,45 @@
+from time import sleep
+from kasa import SmartDevice
+from models.SmartDeviceSpare import SmartDeviceSpare
+from models.customeemterstatus import CustomEmeterStatus
+from services.backgroundservice import BackgroundService
+from services.discoveryservice import DiscoveryService
+from setup.log import get_logger
+
+logger = get_logger('EmeterService')
+
+
+class EmeterService(BackgroundService):
+    def __init__(self, discovery_service: DiscoveryService):
+        super().__init__(self.__job_implementation)
+        self.discovery_service = discovery_service
+
+    async def __job_implementation(self):
+        while not self.cancellation_token.is_cancellation_requested:
+            devices = self.discovery_service.devices
+            for device in devices:
+                try:
+                    data = await self.__fetch(device)
+                    logger.info('Fetched data.', data=data.__dict__, device=SmartDeviceSpare(device).__dict__)
+                except Exception:
+                    # todo log and format exception
+                    logger.error('Failed to fetch data.', device=SmartDeviceSpare(device).__dict__)
+            sleep(5)
+
+        self.cancellation_token = None
+
+    async def __fetch(self, device: SmartDevice) -> CustomEmeterStatus:
+        await device.update()
+
+        emeter_status = await device.get_emeter_realtime()
+        today = device.emeter_today
+        this_month = device.emeter_this_month
+
+        return CustomEmeterStatus(
+            emeter_status['voltage_mv'],
+            emeter_status['current_ma'],
+            emeter_status['power_mw'],
+            emeter_status['total_wh'],
+            today,
+            this_month
+        )

services/discoveryservice.py

@@ -4,9 +4,8 @@
 # configure argument parser
 parser = ArgumentParser()
 
-parser.add_argument('-d', '--device', type=str, required=True)  # device ip or hostname
-parser.add_argument('-t', '--type', type=str, choices=['plug'], default='plug')  # device type
-parser.add_argument('-i', '--interval', type=int, default=5, required=False)  # fetch interval in seconds
+parser.add_argument('--fetch-interval', type=int, default=5, required=False)  # fetch interval in seconds
+parser.add_argument('--discover-interval', type=int, default=30, required=False)  # fetch interval in seconds
 parser.add_argument('--logstash-host', type=str, required=True)  # logstash ip or hostname
 parser.add_argument('--logstash-port', type=int, default=5000, required=False)  # logstash port
 

services/emeterservice.py

@@ -83,5 +83,5 @@ def __add_configuration(logger: logging.Logger, method_name: str, event_dict: Ev
 )
 
 
-def get_logger() -> Any:
-    return structlog.get_logger('kasa-reporter')
+def get_logger(name: str) -> Any:
+    return structlog.get_logger(name)