2323from homeassistant .util import dt as dt_util
2424
2525from .const import (
26+ CONF_LIVE_REPORTING ,
2627 CONF_MODULE_BARCODE ,
2728 CONF_MODULE_NAME ,
2829 CONF_MODULE_PEAK_POWER ,
2930 CONF_MODULE_STRING ,
3031 CONF_MODULES ,
3132 CONF_WRITE_INTERVAL ,
33+ DEFAULT_LIVE_REPORTING ,
3234 DEFAULT_PEAK_POWER ,
3335 DEFAULT_PORT ,
3436 DEFAULT_WRITE_INTERVAL ,
@@ -110,12 +112,15 @@ def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
110112 self ._host : str = entry .data [CONF_HOST ]
111113 self ._port : int = entry .data .get (CONF_PORT , DEFAULT_PORT )
112114 self ._modules : list [dict [str , Any ]] = entry .data .get (CONF_MODULES , [])
113- raw_write_interval = entry .data .get (CONF_WRITE_INTERVAL , DEFAULT_WRITE_INTERVAL )
114- try :
115- self ._write_interval : float = float (raw_write_interval )
116- except (TypeError , ValueError ):
117- self ._write_interval = float (DEFAULT_WRITE_INTERVAL )
118- self ._write_interval = max (0.0 , self ._write_interval )
115+ self ._live_reporting : bool = entry .data .get (
116+ CONF_LIVE_REPORTING , DEFAULT_LIVE_REPORTING
117+ )
118+ raw_write_interval = entry .data .get (CONF_WRITE_INTERVAL , DEFAULT_WRITE_INTERVAL )
119+ try :
120+ self ._write_interval : float = float (raw_write_interval )
121+ except (TypeError , ValueError ):
122+ self ._write_interval = float (DEFAULT_WRITE_INTERVAL )
123+ self ._write_interval = max (0.0 , self ._write_interval )
119124
120125 # Build barcode allowlist from configured modules
121126 self ._configured_barcodes : set [str ] = {
@@ -152,11 +157,13 @@ def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
152157 self ._stop_event = threading .Event ()
153158
154159 # Write-interval throttle state (accessed only from the listener thread)
160+ # Only used when live_reporting is False.
155161 self ._last_ha_update : float = 0.0
156162 self ._ha_update_pending : bool = False
157- # Per-barcode buffer of raw numeric readings accumulated between HA
158- # writes. Flushed (averaged) each time the write interval fires.
159- self ._reading_buffers : dict [str , list [dict [str , Any ]]] = {}
163+ # Per-barcode buffer of timestamped numeric readings accumulated
164+ # between HA writes. Each entry is (monotonic_ts, field_dict).
165+ # Flushed (time-weighted averaged) each time the write interval fires.
166+ self ._reading_buffers : dict [str , list [tuple [float , dict [str , Any ]]]] = {}
160167
161168 # Midnight reset timer handle
162169 self ._midnight_reset_unsub : asyncio .TimerHandle | None = None
@@ -350,36 +357,50 @@ def _listen(self) -> None:
350357 if self ._process_event (event ):
351358 self ._ha_update_pending = True
352359
353- # Push to HA at most once per write interval, sending
354- # per-barcode averages over the buffered readings rather
355- # than the most-recent raw snapshot.
356- now = time .monotonic ()
357- if self ._ha_update_pending and (
358- now - self ._last_ha_update >= self ._write_interval
359- ):
360- self .data ["counters" ] = parser .counters
361- snapshot = self ._build_averaged_snapshot ()
362- per_node_counts = {
363- barcode : len (readings )
364- for barcode , readings in self ._reading_buffers .items ()
365- if readings and barcode in snapshot ["nodes" ]
366- }
367- _LOGGER .debug (
368- "HA update: %d node(s) — %s" ,
369- len (per_node_counts ),
370- ", " .join (
371- f"{ snapshot ['nodes' ][b ].get ('name' , b )} : "
372- f"{ n } reading(s)"
373- for b , n in per_node_counts .items ()
374- ),
375- )
376- self ._reading_buffers .clear ()
377- self .hass .loop .call_soon_threadsafe (
378- self .async_set_updated_data ,
379- snapshot ,
380- )
381- self ._last_ha_update = now
382- self ._ha_update_pending = False
360+ if self ._live_reporting :
361+ # Live mode — push every change immediately
362+ if self ._ha_update_pending :
363+ self .data ["counters" ] = parser .counters
364+ self .hass .loop .call_soon_threadsafe (
365+ self .async_set_updated_data ,
366+ dict (self .data ),
367+ )
368+ self ._ha_update_pending = False
369+ else :
370+ # Interval mode — buffer readings and push
371+ # time-weighted averages once per write_interval.
372+ now = time .monotonic ()
373+ if self ._ha_update_pending and (
374+ now - self ._last_ha_update
375+ >= self ._write_interval
376+ ):
377+ self .data ["counters" ] = parser .counters
378+ snapshot = self ._build_averaged_snapshot (
379+ now
380+ )
381+ per_node_counts = {
382+ barcode : len (readings )
383+ for barcode , readings
384+ in self ._reading_buffers .items ()
385+ if readings
386+ and barcode in snapshot ["nodes" ]
387+ }
388+ _LOGGER .debug (
389+ "HA update: %d node(s) — %s" ,
390+ len (per_node_counts ),
391+ ", " .join (
392+ f"{ snapshot ['nodes' ][b ].get ('name' , b )} : "
393+ f"{ n } reading(s)"
394+ for b , n in per_node_counts .items ()
395+ ),
396+ )
397+ self ._reading_buffers .clear ()
398+ self .hass .loop .call_soon_threadsafe (
399+ self .async_set_updated_data ,
400+ snapshot ,
401+ )
402+ self ._last_ha_update = now
403+ self ._ha_update_pending = False
383404 elif (
384405 RECONNECT_TIMEOUT > 0
385406 and (time .monotonic () - last_data_time ) > RECONNECT_TIMEOUT
@@ -395,16 +416,21 @@ def _listen(self) -> None:
395416 return
396417 _LOGGER .error ("Connection error: %s" , err )
397418 finally :
398- # Flush any data that was held back by the write interval.
399- # Skip updating counters here - parser may not have processed
400- # any data yet, and node/gateway data is what matters for sensors.
401- # On disconnect we discard the buffer and send the latest values.
419+ # Flush any data held back by the write interval.
402420 if self ._ha_update_pending :
403- self ._reading_buffers .clear ()
404- self .hass .loop .call_soon_threadsafe (
405- self .async_set_updated_data ,
406- dict (self .data ),
407- )
421+ if not self ._live_reporting and self ._reading_buffers :
422+ flush_now = time .monotonic ()
423+ snapshot = self ._build_averaged_snapshot (flush_now )
424+ self ._reading_buffers .clear ()
425+ self .hass .loop .call_soon_threadsafe (
426+ self .async_set_updated_data ,
427+ snapshot ,
428+ )
429+ else :
430+ self .hass .loop .call_soon_threadsafe (
431+ self .async_set_updated_data ,
432+ dict (self .data ),
433+ )
408434 self ._ha_update_pending = False
409435 with self ._source_lock :
410436 if self ._source is not None :
@@ -618,7 +644,10 @@ def _handle_power_report(self, event: PowerReportEvent) -> bool:
618644 "last_update" : now .isoformat (),
619645 }
620646 self ._reading_buffers .setdefault (barcode , []).append (
621- {field : self .data ["nodes" ][barcode ][field ] for field in _AVERAGED_FIELDS }
647+ (
648+ time .monotonic (),
649+ {field : self .data ["nodes" ][barcode ][field ] for field in _AVERAGED_FIELDS },
650+ )
622651 )
623652 self ._schedule_save ()
624653 return True
@@ -774,14 +803,21 @@ def _handle_topology(self, event: TopologyEvent) -> bool:
774803 # Persistence helpers
775804 # -------------------------------------------------------------------
776805
777- def _build_averaged_snapshot (self ) -> dict [str , Any ]:
806+ def _build_averaged_snapshot (self , flush_ts : float ) -> dict [str , Any ]:
778807 """Return the data snapshot to push to Home Assistant.
779808
780- For each barcode in ``_reading_buffers``, the buffered numeric
781- readings are averaged and written into a copy of the node dict.
782- ``self.data["nodes"]`` is **not** mutated — the raw latest values
783- are preserved there for persistence; only the returned snapshot
784- carries averaged values.
809+ For each barcode in ``_reading_buffers``, computes a **time-weighted
810+ average** of numeric readings and writes the result into a copy of
811+ the node dict. ``self.data["nodes"]`` is **not** mutated — the raw
812+ latest values are preserved there for persistence; only the returned
813+ snapshot carries averaged values.
814+
815+ Time-weighted averaging weights each reading by the duration it was
816+ "active" (time until the next reading, or until ``flush_ts`` for the
817+ last one). This gives a physically representative average when
818+ readings arrive at uneven intervals.
819+
820+ ``flush_ts`` should be ``time.monotonic()`` at the moment of flush.
785821
786822 Must be called before ``_reading_buffers`` is cleared.
787823 """
@@ -790,15 +826,41 @@ def _build_averaged_snapshot(self) -> dict[str, Any]:
790826 node = snapshot_nodes .get (barcode )
791827 if node is None or not readings :
792828 continue
793- avg_node = dict (node )
794- for field in _AVERAGED_FIELDS :
795- values = [r [field ] for r in readings if r [field ] is not None ]
796- avg_node [field ] = (
797- round (sum (values ) / len (values ), 3 ) if values else None
798- )
799- if avg_node ["power" ] is not None :
829+
830+ if len (readings ) == 1 :
831+ # Single reading — no averaging needed
832+ _ , fields = readings [0 ]
833+ avg_node = dict (node )
834+ for field in _AVERAGED_FIELDS :
835+ avg_node [field ] = fields [field ]
836+ else :
837+ avg_node = dict (node )
838+ for field in _AVERAGED_FIELDS :
839+ weighted_sum = 0.0
840+ total_weight = 0.0
841+ for i , (ts , fields ) in enumerate (readings ):
842+ value = fields [field ]
843+ if value is None :
844+ continue
845+ # Weight = duration until next reading (or flush)
846+ if i + 1 < len (readings ):
847+ weight = readings [i + 1 ][0 ] - ts
848+ else :
849+ weight = flush_ts - ts
850+ if weight <= 0 :
851+ weight = 1e-6 # avoid zero-weight edge case
852+ weighted_sum += value * weight
853+ total_weight += weight
854+ avg_node [field ] = (
855+ round (weighted_sum / total_weight , 3 )
856+ if total_weight > 0
857+ else None
858+ )
859+
860+ if avg_node .get ("power" ) is not None :
861+ peak = avg_node .get ("peak_power" , DEFAULT_PEAK_POWER )
800862 avg_node ["performance" ] = round (
801- (max (avg_node ["power" ], 0.0 ) / avg_node [ "peak_power" ] ) * 100.0 ,
863+ (max (avg_node ["power" ], 0.0 ) / peak ) * 100.0 ,
802864 2 ,
803865 )
804866 else :
0 commit comments