Added live measurement tests

Author: ericbstie

tests/realtime/test_live_measurements.py

@@ -4,6 +4,7 @@
 from datetime import timedelta
 
 import tibber
+from tibber.types.live_measurement import LiveMeasurement
 
 
 @pytest.fixture
@@ -18,4 +19,51 @@ def test_adding_listener_with_unknown_event_raises_exception(home):
     with pytest.raises(ValueError):
         @home.event("invalid-event-name")
         def callback(data):
-            print(data)
+            print(data)
+
+def test_starting_live_feed_with_no_listeners_shows_warning(home, caplog):
+    # Return immediately after the first callback
+    home.start_live_feed(exit_condition = lambda data: True)
+    assert "The event that was broadcasted has no listeners / callbacks! Nothing was run." in caplog.text
+
+def test_retrieving_live_measurements(home):
+    global callback_was_run
+    callback_was_run = False
+    @home.event("live_measurement")
+    def callback(data):
+        global callback_was_run
+        callback_was_run = True
+        assert isinstance(data, LiveMeasurement)
+        timestamp = datetime.strptime(data.timestamp, "%Y-%m-%dT%H:%M:%S.%f%z")
+        timestamp = timestamp.replace(tzinfo=None)
+        now = datetime.now().replace(tzinfo=None)
+        assert timestamp > now - timedelta(seconds=30)
+        assert data.power > 0
+        assert data.last_meter_consumption > 0
+        assert data.accumulated_consumption > 0
+        assert isinstance(data.accumulated_production, float)
+        assert data.accumulated_consumption_last_hour > 0
+        assert isinstance(data.accumulated_production_last_hour, float)
+        assert data.accumulated_cost > 0
+        assert isinstance(data.accumulated_reward, float)
+        assert data.currency == "SEK"
+        assert data.min_power > 0
+        assert data.max_power > 0
+        assert data.average_power > 0
+        assert isinstance(data.power_production, float)
+        assert isinstance(data.power_reactive, float)
+        assert data.power_production_reactive > 0
+        assert isinstance(data.min_power_production, float)
+        assert isinstance(data.max_power_production, float)
+        assert data.last_meter_production > 0
+        assert data.power_factor > 0
+        assert data.voltage_phase_1 > 0
+        assert data.voltage_phase_2 > 0
+        assert data.voltage_phase_3 > 0
+        assert data.current_l1 > 0
+        assert data.current_l2 > 0
+        assert data.current_l3 > 0
+
+    # Return immediately after the first callback
+    home.start_live_feed(exit_condition = lambda data: True)
+    assert callback_was_run

tibber/networking/query_executor.py

@@ -19,8 +19,8 @@ def __init__(self, websession: Optional[aiohttp.ClientSession] = None):
         try:
             self.eventloop = asyncio.get_running_loop()
         except RuntimeError:
-            self.logger.debug("No running event loop was found. Creating a new one with asyncio.get_event_loop()")
-            self.eventloop = asyncio.get_event_loop()
+            self.logger.debug("No running event loop was found. Creating a new one with asyncio.new_event_loop()")
+            self.eventloop = asyncio.new_event_loop()
 
         self.eventloop.run_until_complete(self.__async_init__(websession))
 

tibber/types/home.py

@@ -180,8 +180,11 @@ class TibberHome(NonDecoratedTibberHome):
     """
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.loop = asyncio.get_event_loop()
-        self.websocket_client = None # Used to reference the websocket connection later, if it needs to be closed for example.
+        try:
+            self._loop = asyncio.get_running_loop()
+        except RuntimeError:
+            self._loop = asyncio.new_event_loop()
+        self._websocket_client = None # Used to reference the websocket connection later, if it needs to be closed for example.
         self.websocket_running = False
         self._callbacks = {}
 
@@ -222,17 +225,17 @@ def start_live_feed(self, exit_condition: Callable[[LiveMeasurement], bool] = No
             raise ValueError("The home does not have real time consumption enabled.")
 
         try:
-            self.tibber_client.eventloop.run_until_complete(self.run_websocket_loop(exit_condition, retries = retries, **kwargs))
-        except KeyboardInterrupt:
+            self._loop.run_until_complete(self.run_websocket_loop(exit_condition, retries = retries, **kwargs))
+        except KeyboardInterrupt:  # pragma: no cover
             print("Closing websocket...")
-            if self.websocket_client:
+            if self._websocket_client:
                 self.websocket_running = False
-                self.tibber_client.eventloop.run_until_complete(self.websocket_client.close_async())
-        except BaseException as e:
+                self._loop.run_until_complete(self._websocket_client.close_async())
+        except BaseException as e:  # pragma: no cover
             # Close the websocket, then re-raise the exception
-            if self.websocket_client:
+            if self._websocket_client:
                 self.websocket_running = False
-                self.tibber_client.eventloop.run_until_complete(self.websocket_client.close_async())
+                self._loop.run_until_complete(self._websocket_client.close_async())
             raise e
 
 
@@ -254,7 +257,7 @@ async def retrieve_from_websocket():
                 headers={"Authorization": self.tibber_client.token}
             )
 
-            self.websocket_client = gql.Client(
+            self._websocket_client = gql.Client(
                 transport=transport,
                 fetch_schema_from_transport=True,
             )
@@ -264,7 +267,7 @@ async def retrieve_from_websocket():
             self.logger.debug(f"Connecting to live measurement data endpoint with query: {' '.join(query.split())}")
             document_node_query = parse(query)
 
-            async for data in self.websocket_client.subscribe_async(document_node_query):
+            async for data in self._websocket_client.subscribe_async(document_node_query):
                 self.logger.debug("Real time data received!")
                 self.process_websocket_response(data, exit_condition=exit_condition)
                 if not self.websocket_running: break
@@ -288,7 +291,8 @@ async def retrieve_from_websocket():
                 await asyncio.sleep(retry_interval)
             finally:
                 self.websocket_running = False
-                await self.websocket_client.close_async()
+                if self._websocket_client:
+                    await self._websocket_client.close_async()
 
         if retry_attempts >= retries:
             self.logger.critical(f"Could not connect to the websocket, even after {retry_attempts} tries.")

tibber/types/live_measurement.py

@@ -80,17 +80,17 @@ def max_power(self) -> float:
     @property
     def power_production(self) -> float:
         """Net production (A-) at the moment (Watt)"""
-        return self.cache.get("powerProduction")  # pragma: no cover
+        return self.cache.get("powerProduction")
 
     @property
     def power_reactive(self) -> float:
         """Reactive consumption (Q+) at the moment (kVAr)"""
-        return self.cache.get("powerReactive")  # pragma: no cover
+        return self.cache.get("powerReactive")
 
     @property
     def power_production_reactive(self) -> float:
         """Net reactive production (Q-) at the moment (kVAr)"""
-        return self.cache.get("powerProductionReactive")  # pragma: no cover
+        return self.cache.get("powerProductionReactive")
 
     @property
     def min_power_production(self) -> float:
@@ -110,57 +110,57 @@ def last_meter_production(self) -> float:
     @property
     def power_factor(self) -> float:
         """Power factor (active power / apparent power)"""
-        return self.cache.get("powerFactor")  # pragma: no cover
+        return self.cache.get("powerFactor")
 
     @property
     def voltage_phase_1(self) -> float:
         """Voltage on phase 1; on Kaifa and Aidon meters the value is not part 
         of every HAN data frame therefore the value is null at timestamps with
         second value other than 0, 10, 20, 30, 40, 50. There can be other deviations
         based on concrete meter firmware."""
-        return self.cache.get("voltagePhase1")  # pragma: no cover
+        return self.cache.get("voltagePhase1")
 
     @property
     def voltage_phase_2(self) -> float:
         """Voltage on phase 2; on Kaifa and Aidon meters the value is not part 
         of every HAN data frame therefore the value is null at timestamps with
         second value other than 0, 10, 20, 30, 40, 50. There can be other deviations
         based on concrete meter firmware."""
-        return self.cache.get("voltagePhase2")  # pragma: no cover
+        return self.cache.get("voltagePhase2")
 
     @property
     def voltage_phase_3(self) -> float:
         """Voltage on phase 3; on Kaifa and Aidon meters the value is not part 
         of every HAN data frame therefore the value is null at timestamps with
         second value other than 0, 10, 20, 30, 40, 50. There can be other deviations
         based on concrete meter firmware."""
-        return self.cache.get("voltagePhase3")  # pragma: no cover
+        return self.cache.get("voltagePhase3")
 
     @property
     def currentL1(self) -> float:
         """Current on L1; on Kaifa and Aidon meters the value is not part of
         every HAN data frame therefore the value is null at timestamps with
         second value other than 0, 10, 20, 30, 40, 50. There can be other deviations
         based on concrete meter firmware."""
-        return self.cache.get("currentL1")  # pragma: no cover
+        return self.cache.get("currentL1")
 
     @property
     def currentL2(self) -> float:
         """Current on L2; on Kaifa and Aidon meters the value is not part of
         every HAN data frame therefore the value is null at timestamps with
         second value other than 0, 10, 20, 30, 40, 50. There can be other deviations
         based on concrete meter firmware."""
-        return self.cache.get("currentL2")  # pragma: no cover
+        return self.cache.get("currentL2")
 
     @property
     def currentL3(self) -> float:
         """Current on L3; on Kaifa and Aidon meters the value is not part of
         every HAN data frame therefore the value is null at timestamps with
         second value other than 0, 10, 20, 30, 40, 50. There can be other deviations
         based on concrete meter firmware."""
-        return self.cache.get("currentL3")  # pragma: no cover
+        return self.cache.get("currentL3")
 
     @property
-    def signal_strength(self) -> int:
+    def signal_strength(self) -> int:  # pragma: no cover
         """Device signal strength (Pulse - dB; Watty - percent)"""
-        return self.cache.get("signalStrength")  # pragma: no cover
+        return self.cache.get("signalStrength")