Skip to content

Commit

Permalink
Remove old caching system
Browse files Browse the repository at this point in the history
  • Loading branch information
@PimDoos
PimDoos committed Feb 22, 2025
1 parent 53e2f78 commit 789b8fc
Show file tree
Hide file tree
Showing 2 changed files with 2 additions and 231 deletions.
105 changes: 0 additions & 105 deletions custom_components/sessy/sessyentity.py
View file

This file was deleted.

128 changes: 2 additions & 126 deletions custom_components/sessy/util.py
View file
Original file line number Diff line number Diff line change
@@ -1,141 +1,17 @@
from __future__ import annotations
from datetime import datetime, time, timedelta
from datetime import datetime, time
from enum import Enum
from homeassistant.core import HomeAssistant
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_SCAN_INTERVAL

from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.event import async_track_time_interval, async_track_time_change
from homeassistant.helpers.dispatcher import async_dispatcher_send

from sessypy.const import SessyApiCommand
from sessypy.devices import SessyDevice, SessyBattery, SessyMeter, SessyP1Meter, SessyCTMeter

import logging
_LOGGER = logging.getLogger(__name__)

from .const import (
DEFAULT_SCAN_INTERVAL_POWER, DOMAIN, SCAN_INTERVAL_OTA_CHECK, SESSY_CACHE, SESSY_CACHE_INTERVAL, SESSY_CACHE_TRACKERS,
SESSY_CACHE_TRIGGERS, SESSY_DEVICE, UPDATE_TOPIC, DEFAULT_SCAN_INTERVAL, SCAN_INTERVAL_SCHEDULE
)

# START legacy caching system
async def setup_cache(hass: HomeAssistant, config_entry: ConfigEntry):
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE] = dict()
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRACKERS] = dict()
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRIGGERS] = dict()
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_INTERVAL] = dict()

async def setup_cache_commands(hass, config_entry: ConfigEntry, device: SessyDevice, setup=True):

# Skip static update intervals if updating from config flow handler
if setup:
# Sessy will not check for updates automatically, poll at intervals
await setup_cache_command(hass, config_entry, SessyApiCommand.OTA_CHECK, SCAN_INTERVAL_OTA_CHECK)

await setup_cache_command(hass, config_entry, SessyApiCommand.OTA_STATUS)
await setup_cache_command(hass, config_entry, SessyApiCommand.SYSTEM_INFO)
await setup_cache_command(hass, config_entry, SessyApiCommand.NETWORK_STATUS)

if isinstance(device, SessyBattery):
await setup_cache_command(hass, config_entry, SessyApiCommand.SYSTEM_SETTINGS)
await setup_cache_command(hass, config_entry, SessyApiCommand.POWER_STRATEGY)
await setup_cache_command(hass, config_entry, SessyApiCommand.DYNAMIC_SCHEDULE, SCAN_INTERVAL_SCHEDULE)


# Get power scan interval from options flow
scan_power_seconds = config_entry.options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL_POWER.seconds)
scan_interval_power = timedelta(seconds = scan_power_seconds)

if isinstance(device, SessyBattery):
await setup_cache_command(hass, config_entry, SessyApiCommand.POWER_STATUS, scan_interval_power)

elif isinstance(device, SessyP1Meter):
await setup_cache_command(hass, config_entry, SessyApiCommand.P1_DETAILS, scan_interval_power)

elif isinstance(device, SessyCTMeter):
await setup_cache_command(hass, config_entry, SessyApiCommand.CT_DETAILS, scan_interval_power)

if isinstance(device, SessyMeter):
await setup_cache_command(hass, config_entry, SessyApiCommand.METER_GRID_TARGET)

if isinstance(device, SessyBattery) or isinstance(device, SessyCTMeter):
await setup_cache_command(hass, config_entry, SessyApiCommand.ENERGY_STATUS)


async def setup_cache_command(hass: HomeAssistant, config_entry: ConfigEntry, command: SessyApiCommand, interval: timedelta | dict = DEFAULT_SCAN_INTERVAL):
update = set_cache_command(hass, config_entry, command, interval)
await update()


def set_cache_command(hass: HomeAssistant, config_entry: ConfigEntry, command: SessyApiCommand, interval: timedelta | dict = DEFAULT_SCAN_INTERVAL, skip_update: bool = False) -> function:
if not command in hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE]:
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE][command] = dict()

async def update(event_time_utc: datetime = None):
device: SessyDevice = hass.data[DOMAIN][config_entry.entry_id][SESSY_DEVICE]
cache: dict = hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE]

try:
result: dict = await device.api.get(command)
cache[command] = result
if "error" in result:
_LOGGER.debug(f"Sessy api returned an error while updating cache for {command}: {result.get("error")}")
except Exception as e:
result = None
cache[command] = None
_LOGGER.debug(f"Updating cache for {command} failed with error {e}")

async_dispatcher_send(hass, UPDATE_TOPIC.format(command))

if command in hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRACKERS]:
# Remove running tracker to avoid duplicates
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRACKERS][command]()

if type(interval) == timedelta:
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRACKERS][command] = async_track_time_interval(hass, update, interval)
elif type(interval) == dict:
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRACKERS][command] = async_track_time_change(hass, update, **interval)
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRIGGERS][command] = update
hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_INTERVAL][command] = interval

return update

async def clear_cache_command(hass: HomeAssistant, config_entry: ConfigEntry, command: SessyApiCommand = None):
trackers = hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRACKERS]
if command == None:

for tracker_command in trackers:
tracker = trackers[tracker_command]
tracker()
else:
tracker = trackers[command]
tracker()

def get_cache_command(hass: HomeAssistant, config_entry: ConfigEntry, command: SessyApiCommand, key: str = None):
cache: dict = hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE].get(command, None)
if cache == None:
return None
if key:
return cache.get(key)
else:
return cache

def get_cache_interval(hass: HomeAssistant, config_entry: ConfigEntry, command: SessyApiCommand):
return hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_INTERVAL][command]

def assert_cache_interval(hass: HomeAssistant, config_entry: ConfigEntry, command: SessyApiCommand, interval: timedelta = DEFAULT_SCAN_INTERVAL):
current_interval = get_cache_interval(hass, config_entry, command)
if current_interval != interval:
_LOGGER.debug(f"Updating cache update interval for {command}")
set_cache_command(hass, config_entry, command, interval, skip_update=True)

async def trigger_cache_update(hass: HomeAssistant, config_entry: ConfigEntry, command: SessyApiCommand):
update = hass.data[DOMAIN][config_entry.entry_id][SESSY_CACHE_TRIGGERS][command]
await update()

# END Legacy caching system
from .const import DOMAIN

# Transform functions

Expand Down

0 comments on commit 789b8fc

Please sign in to comment.