Merge pull request #8 from die-bauerei/push_to_oh

pushing state-change of openhab plug instead of polling

oh_to_smartplug_energy_controller/rules/habapp_rules.py

@@ -1,9 +1,8 @@
 import HABApp
 #import HABApp.openhab.interface
 from HABApp.core.events import EventFilter
-from HABApp.openhab.events import ItemStateChangedEvent, ItemStateChangedEventFilter, ThingStatusInfoChangedEvent
+from HABApp.openhab.events import ItemStateChangedEvent, ItemStateChangedEventFilter, ItemStateUpdatedEvent, ItemStateUpdatedEventFilter, ThingStatusInfoChangedEvent
 from HABApp.openhab.items import NumberItem, SwitchItem, Thing
-from HABApp.openhab.definitions.values import OnOffValue
 
 from datetime import timedelta
 
@@ -28,11 +27,10 @@ def __init__(self, watt_obtained_from_provider_item : str, watt_produced_item :
         super().__init__()
         self._lock : asyncio.Lock = asyncio.Lock()
         self._url=req_url
-        self._send_latest_value_job=None
         self._watt_obtained_item=NumberItem.get_item(watt_obtained_from_provider_item)
-        self._watt_obtained_item.listen_event(self._item_state_changed, ItemStateChangedEventFilter())
+        self._watt_obtained_item.listen_event(self._watt_obtained_updated, ItemStateUpdatedEventFilter())
         self._watt_produced_item=NumberItem.get_item(watt_produced_item)
-        self._watt_produced_item.listen_event(self._item_state_changed, ItemStateChangedEventFilter())
+        self._watt_produced_item.listen_event(self._watt_produced_changed, ItemStateChangedEventFilter())
         self.run.soon(callback=self._init_oh_connection) # type: ignore
 
     async def _init_oh_connection(self):
@@ -41,26 +39,27 @@ async def _init_oh_connection(self):
                 log.error(f"Failed to init SmartMeterValueForwarder. Return code: {response.status}. Text: {await response.text()}")
             else:
                 data = await response.json()
-                force_request_time_in_sec=max(10, data['min_expected_freq_in_sec']-10)
-                self._send_latest_value_job=self.run.countdown(force_request_time_in_sec, self._send_latest_values) # type: ignore
+                force_request_time_in_sec=max(1, data['min_expected_freq_in_sec']-1)
+                self._watt_obtained_item.watch_update(force_request_time_in_sec).listen_event(self._send_latest_values)
                 log.info(f"SmartMeterValueForwarder successfully initialized with a force_request_time_in_sec of {force_request_time_in_sec}.")
 
-    async def _item_state_changed(self, event):
+    async def _watt_obtained_updated(self, event):
+        assert isinstance(event, ItemStateUpdatedEvent), type(event)
+        await self._send_values(str(self._watt_obtained_item.get_value()), str(self._watt_produced_item.get_value()))
+
+    async def _watt_produced_changed(self, event):
         assert isinstance(event, ItemStateChangedEvent), type(event)
         await self._send_values(str(self._watt_obtained_item.get_value()), str(self._watt_produced_item.get_value()))
 
-    async def _send_latest_values(self):
+    async def _send_latest_values(self, event):
+        log.warning("Forcing request to send latest values. This should not happen. Check your service which reads values from your electricity meter.")
         await self._send_values(str(self._watt_obtained_item.get_value()), str(self._watt_produced_item.get_value()))
 
     async def _send_values(self, watt_obtained_value : str, watt_produced_value : str):
         async with self.async_http.put(self._url, json={'watt_obtained_from_provider': watt_obtained_value, 
                                                         'watt_produced': watt_produced_value}) as response:
             if response.status != http.HTTPStatus.OK:
                 log.warning(f"Failed to forward smart meter values via put request to {self._url}. Return code: {response.status}. Text: {await response.text()}")
-        if self._send_latest_value_job:
-            async with self._lock:
-                self._send_latest_value_job.stop()
-                self._send_latest_value_job.reset()
 
 class SmartPlugSynchronizer(HABApp.Rule):
     def __init__(self, smartplug_uuid : str) -> None:
@@ -71,17 +70,9 @@ def __init__(self, smartplug_uuid : str) -> None:
         self._smartplug_uuid : str = smartplug_uuid
         self._info_url='http://localhost:8000/plug-info'
         self._state_url='http://localhost:8000/plug-state'
-        self._smart_meter_url='http://localhost:8000/smart-meter'
         self.run.soon(callback=self._init_oh_connection) # type: ignore
 
     async def _init_oh_connection(self):
-        async with self.async_http.get(f"{self._smart_meter_url}", headers={'Cache-Control': 'no-cache'}) as response:
-            if response.status != http.HTTPStatus.OK:
-                log.error(f"Failed to init sync_time_delta. Return code: {response.status}. Text: {await response.text()}")
-            else:
-                data = await response.json()
-                sync_time_delta=data['min_expected_freq_in_sec']/2
-
         async with self.async_http.get(f"{self._info_url}/{self._smartplug_uuid}", headers={'Cache-Control': 'no-cache'}) as response:
             if response.status != http.HTTPStatus.OK:
                 log.error(f"Failed to init SmartPlug with UUID {self._smartplug_uuid}. Return code: {response.status}. Text: {await response.text()}")
@@ -94,7 +85,7 @@ async def _init_oh_connection(self):
                 self._power_consumption_item=NumberItem.get_item(data['oh_power_consumption_item_name'])
                 self._power_consumption_item.listen_event(self._sync_values, ItemStateChangedEventFilter())
                 log.info(f"SmartPlug with UUID {self._smartplug_uuid} successfully initialized.")
-                self.run.every(start_time=timedelta(seconds=1), interval=timedelta(seconds=sync_time_delta), callback=self._sync_state) # type: ignore
+                self.run.every(start_time=timedelta(seconds=1), interval=timedelta(minutes=20), callback=self._check_state) # type: ignore
 
     async def _sync_values(self, event):
         power_consumption=self._power_consumption_item.get_value()
@@ -106,15 +97,21 @@ async def _sync_values(self, event):
             if response.status != http.HTTPStatus.OK:
                 log.warning(f"Failed to forward smartplug values via put request to {url}. Return code: {response.status}. Text: {await response.text()}")
 
-    async def _sync_state(self):
-        async with self.async_http.get(f"{self._state_url}/{self._smartplug_uuid}", headers={'Cache-Control': 'no-cache'}) as response:
-            if response.status != http.HTTPStatus.OK:
-                log.warning(f"Failed to sync state of SmartPlug with UUID {self._smartplug_uuid}. Return code: {response.status}. Text: {await response.text()}")
-            else:
-                data = await response.json()
-                value = OnOffValue.ON if data['proposed_state'] == 'On' else OnOffValue.OFF
-                log.info(f"About to sync state of {self._switch_item.name} to {value}")
-                self.openhab.send_command(self._switch_item.name, value)
+    async def _check_state(self):
+        # check if the proposed state has been set in an interval of ~10sec
+        check_count=0
+        while check_count < 10:
+            async with self.async_http.get(f"{self._state_url}/{self._smartplug_uuid}", headers={'Cache-Control': 'no-cache'}) as response:
+                if response.status != http.HTTPStatus.OK:
+                    log.warning(f"Failed to check state of SmartPlug with UUID {self._smartplug_uuid}. Return code: {response.status}. Text: {await response.text()}")
+                else:
+                    data = await response.json()
+                    proposed_to_be_on = True if data['proposed_state'] == 'On' else False
+                    if self._switch_item.is_on() == proposed_to_be_on:
+                        return # check successful
+            await asyncio.sleep(1)
+            check_count+=1
+        log.warning(f"Switch {self._switch_item.name} is not in the proposed state.")
 
 from pathlib import Path
 from dotenv import load_dotenv

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "smartplug_energy_controller"
-version = "0.1.1"
+version = "0.1.2"
 description = "Turning smartplugs on/off depending on current electricity consumption"
 authors = ["Heiko Bauer <heiko_bauer@icloud.com>"]
 repository = "https://github.com/die-bauerei/smartplug-energy-controller"

smartplug_energy_controller/__init__.py

@@ -0,0 +1,43 @@
+import logging
+from typing import Union
+from pathlib import Path
+from smartplug_energy_controller.config import ConfigParser, OpenHabConnectionConfig
+from smartplug_energy_controller.utils import OpenhabConnection
+
+try:
+    import importlib.metadata
+    __version__ = importlib.metadata.version('smartplug_energy_controller')
+except:
+    __version__ = 'development'
+
+_logger : Union[logging.Logger, None] = None
+def init_logger(file : Union[Path, None], level) -> None:
+    global _logger
+    if _logger is None:
+        _logger = logging.getLogger('smartplug-energy-controller')
+        log_handler : Union[logging.FileHandler, logging.StreamHandler] = logging.FileHandler(file) if file else logging.StreamHandler() 
+        formatter = logging.Formatter("%(levelname)s: %(asctime)s: %(message)s")
+        log_handler.setFormatter(formatter)
+        _logger.addHandler(log_handler)
+        _logger.setLevel(logging.INFO)
+        _logger.info(f"Starting smartplug-energy-controller version {__version__}")
+        _logger.setLevel(level)
+
+def get_logger() -> logging.Logger:
+    global _logger
+    if _logger is None:
+        raise RuntimeError(f"Unable to get logger. It has not been initialized yet.")
+    return _logger
+
+_oh_connection : Union[OpenhabConnection, None] = None
+def init_oh_connection(oh_con_cfg : Union[OpenHabConnectionConfig, None]) -> None:
+    global _oh_connection
+    _oh_connection = OpenhabConnection(oh_con_cfg, get_logger()) if oh_con_cfg else None
+
+def get_oh_connection() -> Union[OpenhabConnection, None]:
+    global _oh_connection
+    return _oh_connection
+
+def init(cfg_parser : ConfigParser) -> None:
+    init_logger(cfg_parser.general.log_file, cfg_parser.general.log_level)
+    init_oh_connection(cfg_parser.oh_connection)

smartplug_energy_controller/app.py

@@ -1,4 +1,3 @@
-import logging
 import uvicorn
 
 from pathlib import Path
@@ -8,36 +7,20 @@
 from typing import Union, cast
 from pydantic import BaseModel
 from pydantic_settings import BaseSettings
+from datetime import datetime
 
+from smartplug_energy_controller import init, get_logger
 from smartplug_energy_controller.plug_controller import *
 from smartplug_energy_controller.plug_manager import PlugManager
 from smartplug_energy_controller.config import ConfigParser
 
 class Settings(BaseSettings):
     config_path : Path
 
-def create_logger(file : Union[Path, None]) -> logging.Logger:
-    logger = logging.getLogger('smartplug-energy-controller')
-    log_handler : Union[logging.FileHandler, logging.StreamHandler] = logging.FileHandler(file) if file else logging.StreamHandler() 
-    formatter = logging.Formatter("%(levelname)s: %(asctime)s: %(message)s")
-    log_handler.setFormatter(formatter)
-    logger.addHandler(log_handler)
-    return logger
-
-try:
-    import importlib.metadata
-    __version__ = importlib.metadata.version('smartplug_energy_controller')
-except:
-    __version__ = 'development'
-
 settings = Settings() # type: ignore
 cfg_parser = ConfigParser(settings.config_path, Path(f"{root_path}/../oh_to_smartplug_energy_controller/config.yml"))
-logger=create_logger(cfg_parser.general.log_file)
-logger.setLevel(logging.INFO)
-logger.info(f"Starting smartplug-energy-controller version {__version__}")
-logger.setLevel(cfg_parser.general.log_level)
-manager=PlugManager.create(logger, cfg_parser)
-
+init(cfg_parser)
+manager=PlugManager.create(get_logger(), cfg_parser)
 app = FastAPI()
 
 class PlugValues(BaseModel):
@@ -48,6 +31,7 @@ class PlugValues(BaseModel):
 class SmartMeterValues(BaseModel):
     watt_obtained_from_provider: float
     watt_produced: Union[None, float] = None
+    timestamp : Union[None, datetime] = None
 
 @app.get("/")
 async def root(request: Request):
@@ -73,7 +57,7 @@ async def smart_meter_get():
 
 @app.put("/smart-meter")
 async def smart_meter_put(smart_meter_values: SmartMeterValues):
-    await manager.add_smart_meter_values(smart_meter_values.watt_obtained_from_provider, smart_meter_values.watt_produced)
+    await manager.add_smart_meter_values(smart_meter_values.watt_obtained_from_provider, smart_meter_values.watt_produced, smart_meter_values.timestamp)
 
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=8000)

smartplug_energy_controller/config.py

@@ -26,6 +26,12 @@ class OpenHabSmartPlugConfig(SmartPlugConfig):
     oh_switch_item_name : str = ''
     oh_power_consumption_item_name : str = ''
 
+@dataclass(frozen=True)
+class OpenHabConnectionConfig():
+    oh_url : str = ''
+    oh_user : str = ''
+    oh_password : str = ''
+
 @dataclass(frozen=True)
 class GeneralConfig():
     # Write logging to this file instead of to stdout
@@ -40,23 +46,31 @@ class GeneralConfig():
 class ConfigParser():
     def __init__(self, file : Path, habapp_config : Path) -> None:
         self._smart_plugs : Dict[str, SmartPlugConfig] = {}
+        self._oh_connection : Union[None, OpenHabConnectionConfig] = None
         yaml=YAML(typ='safe', pure=True)
         data=yaml.load(file)
         self._read_from_dict(data)
         if 'openhab_connection' in data:
+            self._oh_connection=OpenHabConnectionConfig(data['openhab_connection']['oh_url'], 
+                                                        data['openhab_connection']['oh_user'], 
+                                                        data['openhab_connection']['oh_password'])
             self._transfer_to_habapp(data['openhab_connection'], habapp_config)
 
     @property
     def general(self) -> GeneralConfig:
         return self._general
 
+    @property
+    def oh_connection(self) -> Union[None, OpenHabConnectionConfig]:
+        return self._oh_connection
+
     @cached_property
     def plug_uuids(self) -> List[str]:
         return list(self._smart_plugs.keys())
 
     def plug(self, plug_uuid : str) -> SmartPlugConfig:
         return self._smart_plugs[plug_uuid]
-        
+
     def _read_from_dict(self, data : dict):
         self._general=GeneralConfig(Path(data['log_file']), data['log_level'], data['eval_time_in_min'])
         for plug_uuid in data['smartplugs']:
@@ -71,7 +85,7 @@ def _read_from_dict(self, data : dict):
                 plug_cfg['oh_thing_name'], plug_cfg['oh_switch_item_name'], plug_cfg['oh_power_consumption_item_name'])
             else:
                 raise ValueError(f"Unknown Plug type: {plug_cfg['type']}")
-
+    
     def _transfer_to_habapp(self, data : dict, habapp_config_path : Path):
         # 1. fwd config to habapp config file
         yaml=YAML(typ='safe', pure=True)

smartplug_energy_controller/plug_controller.py

@@ -7,7 +7,8 @@
 from plugp100.new.device_factory import connect, DeviceConnectConfiguration
 from plugp100.new.tapoplug import TapoPlug
 
-from .config import *
+from smartplug_energy_controller.config import *
+from smartplug_energy_controller import get_oh_connection
 
 import asyncio
 
@@ -119,6 +120,7 @@ class OpenHabPlugController(PlugController):
 
     def __init__(self, logger : Logger, plug_cfg : OpenHabSmartPlugConfig) -> None:
         super().__init__(logger, plug_cfg)
+        assert get_oh_connection() is not None
         self._plug_cfg=plug_cfg
         assert self._plug_cfg.oh_thing_name != ''
         assert self._plug_cfg.oh_switch_item_name != ''
@@ -143,6 +145,30 @@ async def is_online(self) -> bool:
     async def is_on(self) -> bool:
         return self._is_on
 
+    async def turn_on(self) -> bool:
+        base_rc = await super().turn_on()
+        oh_connection = get_oh_connection()
+        if oh_connection is None:
+            self._logger.error("OpenHabConnection is not set. Cannot turn on plug")
+        elif base_rc:
+            success=await oh_connection.post_to_item(self._plug_cfg.oh_switch_item_name, 'ON')
+            if success:
+                self._logger.info("Turned OpenHabPlug Plug on")
+            return success
+        return False
+
+    async def turn_off(self) -> bool:
+        base_rc = await super().turn_off()
+        oh_connection = get_oh_connection()
+        if oh_connection is None:
+            self._logger.error("OpenHabConnection is not set. Cannot turn off plug")
+        elif base_rc:
+            success=await oh_connection.post_to_item(self._plug_cfg.oh_switch_item_name, 'OFF')
+            if success:
+                self._logger.info("Turned OpenHabPlug Plug off")
+            return success
+        return False
+
     async def update_values(self, watt_consumed_at_plug: float, online : bool, is_on : bool) -> None:
         async with self._lock:
             self._watt_consumed_at_plug=watt_consumed_at_plug

smartplug_energy_controller/plug_manager.py

@@ -5,9 +5,9 @@
 
 import asyncio
 
-from .utils import *
-from .config import *
-from .plug_controller import *
+from smartplug_energy_controller.utils import *
+from smartplug_energy_controller.config import *
+from smartplug_energy_controller.plug_controller import *
 
 efficiency_tolerance=0.05
 
@@ -50,6 +50,9 @@ def _current_break_even(self) -> Union[None, float]:
 
     def plug(self, plug_uuid : str) -> PlugController:
         return self._controllers[plug_uuid]
+
+    def plugs(self) -> List[PlugController]:
+        return list(self._controllers.values())
 
     async def _handle_turn_on_plug(self) -> None:
         assert self._having_overproduction
@@ -121,7 +124,8 @@ def _evaluate(self, watt_produced : Union[None, float] = None) -> bool:
         elif had_overprotection and self._having_overproduction and self._break_even is not None:
             # decrease break-even value when overproduction is still present
             self._break_even = max(self._base_load, self._break_even*0.975)
-            self._logger.info(f"Break-even value has been updated from {old_break_even} to {self._break_even}")
+            if old_break_even != self._break_even:
+                self._logger.info(f"Break-even value has been updated from {old_break_even} to {self._break_even}")
         self._watt_produced=watt_produced
         return True