55from asyncio import Task , create_task , gather
66from collections .abc import Awaitable , Callable
77from dataclasses import replace
8- from datetime import UTC , datetime
8+ from datetime import UTC , datetime , timedelta
99from functools import wraps
1010import logging
1111from math import ceil
@@ -606,14 +606,14 @@ def _check_timestamp_is_recent(
606606 return False
607607 return True
608608
609- async def _energy_log_records_load_from_cache (self ) -> bool :
609+ async def _energy_log_records_load_from_cache (self ) -> bool : # noqa: PLR0912
610610 """Load energy_log_record from cache."""
611611 if (cache_data := self ._get_cache (CACHE_ENERGY_COLLECTION )) is None :
612612 _LOGGER .warning (
613613 "Failed to restore energy log records from cache for node %s" , self .name
614614 )
615615 return False
616- restored_logs : dict [int , list [int ]] = {}
616+ restored_logs : dict [int , dict [int , tuple [ datetime , int ] ]] = {}
617617 if cache_data == "" :
618618 _LOGGER .debug ("Cache-record is empty" )
619619 return False
@@ -626,24 +626,41 @@ async def _energy_log_records_load_from_cache(self) -> bool:
626626 if len (timestamp_energy_log ) == 6 :
627627 address = int (log_fields [0 ])
628628 slot = int (log_fields [1 ])
629- self ._energy_counters .add_pulse_log (
630- address = address ,
631- slot = slot ,
632- timestamp = datetime (
633- year = int (timestamp_energy_log [0 ]),
634- month = int (timestamp_energy_log [1 ]),
635- day = int (timestamp_energy_log [2 ]),
636- hour = int (timestamp_energy_log [3 ]),
637- minute = int (timestamp_energy_log [4 ]),
638- second = int (timestamp_energy_log [5 ]),
639- tzinfo = UTC ,
640- ),
641- pulses = int (log_fields [3 ]),
642- import_only = True ,
629+ pulses = int (log_fields [3 ])
630+ timestamp = datetime (
631+ year = int (timestamp_energy_log [0 ]),
632+ month = int (timestamp_energy_log [1 ]),
633+ day = int (timestamp_energy_log [2 ]),
634+ hour = int (timestamp_energy_log [3 ]),
635+ minute = int (timestamp_energy_log [4 ]),
636+ second = int (timestamp_energy_log [5 ]),
637+ tzinfo = UTC ,
643638 )
644639 if restored_logs .get (address ) is None :
645- restored_logs [address ] = []
646- restored_logs [address ].append (slot )
640+ restored_logs [address ] = {}
641+ restored_logs [address ][slot ] = (timestamp , pulses )
642+
643+ # Sort and prune the records loaded from cache
644+ sorted_logs : dict [int , dict [int , tuple [datetime , int ]]] = {}
645+ skip_before = datetime .now (tz = UTC ) - timedelta (hours = DAY_IN_HOURS )
646+ sorted_addresses = sorted (restored_logs .keys (), reverse = True )
647+ for address in sorted_addresses :
648+ sorted_slots = sorted (restored_logs [address ].keys (), reverse = True )
649+ for slot in sorted_slots :
650+ if restored_logs [address ][slot ][0 ] > skip_before :
651+ if sorted_logs .get (address ) is None :
652+ sorted_logs [address ] = {}
653+ sorted_logs [address ][slot ] = restored_logs [address ][slot ]
654+
655+ for address , data in sorted_logs .items ():
656+ for slot , pulse_data in data .items ():
657+ self ._energy_counters .add_pulse_log (
658+ address = address ,
659+ slot = slot ,
660+ pulses = pulse_data [1 ],
661+ timestamp = pulse_data [0 ],
662+ import_only = True ,
663+ )
647664
648665 self ._energy_counters .update ()
649666
0 commit comments