From 35c86ffbd3ed03c427debabece4b0c89a0e82c0b Mon Sep 17 00:00:00 2001 From: martin Date: Mon, 16 Dec 2024 22:52:07 +0100 Subject: [PATCH] added custom config and started developing roon trigger --- config.py | 19 ++++++--- cpu_governor_auto_adjust.xml | 9 +++++ logger.py | 4 +- requirements.txt | 2 +- schedule.py | 2 +- triggers/roon.py | 74 +++++++++++++++++++++++++++++++++++- 6 files changed, 100 insertions(+), 10 deletions(-) diff --git a/config.py b/config.py index f4e5b63..a8eaa4e 100644 --- a/config.py +++ b/config.py @@ -4,16 +4,18 @@ from pathlib import Path from lxml import etree from functools import cached_property, lru_cache from arguments import ArgumentsParser -from typing import NamedTuple, Optional - +from typing import NamedTuple, Optional, Any +from collections import OrderedDict +import xmltodict class TriggerTuple(NamedTuple): name: str interval_in_seconds: int governor: str + custom_config: OrderedDict[str, Any] - def __repr__(self): - return f"{self.name}, interval: {self.interval_in_seconds}, governor: {self.governor}" + def __repr__(self) -> str: + return f"{self.name}, interval: {self.interval_in_seconds}, governor: {self.governor}, custom_config: {self.custom_config}" class Config: @@ -68,10 +70,17 @@ class Config: name = self._get_single_text_value_from_xpath('name', trigger_elem) interval_in_seconds = self._get_single_text_value_from_xpath('intervalInSeconds', trigger_elem) governor = self._get_single_text_value_from_xpath('governor', trigger_elem) + try: + custom_config_elem = self._get_single_element_by_xpath('customConfig', trigger_elem) + custom_config = xmltodict.parse(etree.tostring(custom_config_elem))['customConfig'] + except AssertionError: + self.log.debug("Trigger %s has no custom config", name) + custom_config = OrderedDict({}) _new_trigger = TriggerTuple( name=name, interval_in_seconds=int(interval_in_seconds), - governor=governor + governor=governor, + custom_config=custom_config ) ret_val.append(_new_trigger) self.log.info("Loaded trigger: %s", _new_trigger) diff --git a/cpu_governor_auto_adjust.xml b/cpu_governor_auto_adjust.xml index 72da2ba..2760b18 100644 --- a/cpu_governor_auto_adjust.xml +++ b/cpu_governor_auto_adjust.xml @@ -17,6 +17,15 @@ roon 2 performance + + cpu_governor_auto_adjust_for_roon + CPU Governor Auto Adjust for Roon + 1.0.0 + Martin Reurekas + martin@semrks.nl + /home/martin/dev/cpu_governor_auto_adjust/triggers/.data/core_id_file + /home/martin/dev/cpu_governor_auto_adjust/triggers/.data/token_file + diff --git a/logger.py b/logger.py index b883c27..a67cb45 100644 --- a/logger.py +++ b/logger.py @@ -14,10 +14,10 @@ def getLogger(name: str = 'root', loglevel: str = 'INFO') -> Logger: # if logger 'name' does not already exist, create it and attach handlers else: # set logLevel to loglevel or to INFO if requested level is incorrect - loglevel = getattr(logging, loglevel.upper(), "INFO") + loglevel = getattr(logging, loglevel.upper()) logger.setLevel(loglevel) if loglevel == logging.DEBUG: - fmt = '%(asctime)s %(levelname)-8s %(message)s (%(filename)s)' + fmt = '%(asctime)s %(levelname)-8s %(message)s (%(filename)s:%(lineno)d)' else: fmt = '%(asctime)s %(levelname)-8s %(message)s' fmt_date = '%Y-%m-%dT%T' diff --git a/requirements.txt b/requirements.txt index d3f7f33..a221ff7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ cpufreq==0.3.5 -lxml==5.3.0 +roonapi==0.1.6 diff --git a/schedule.py b/schedule.py index df9a882..9293683 100644 --- a/schedule.py +++ b/schedule.py @@ -13,7 +13,7 @@ class TriggerScheduler(AppClass): async def trigger(self, _trigger: Trigger) -> None: """Run a trigger with a specific name at a given interval.""" while True: - self.log.info("Trigger %s is running", _trigger.name) + self.log.debug("Trigger %s is running", _trigger.name) _trigger.run() await asyncio.sleep(_trigger.config.interval_in_seconds) diff --git a/triggers/roon.py b/triggers/roon.py index d0fc039..b88ba1e 100644 --- a/triggers/roon.py +++ b/triggers/roon.py @@ -1,11 +1,83 @@ from trigger import Trigger from config import Config +from roonapi import RoonApi, RoonDiscovery # type: ignore +from functools import cached_property +from typing import Any, NamedTuple +from pathlib import Path + + +class RoonServer(NamedTuple): + ip: str + port: str + + def __repr__(self) -> str: + return f"RoonServer: ip: {self.ip}, port: {self.port}" class RoonTrigger(Trigger): def __init__(self, _config: Config) -> None: super().__init__(_config) + @property + def playing(self) -> bool: + return False + + @cached_property + def appinfo(self) -> dict[str, Any]: + _appinfo = { + "extension_id": self.config.custom_config['extensionId'], + "display_name": self.config.custom_config['displayName'], + "display_version": self.config.custom_config['displayVersion'], + "publisher": self.config.custom_config['publisher'], + "email": self.config.custom_config['email'], + } + self.log.debug("appinfo: %s", _appinfo) + return _appinfo + + @cached_property + def core_id(self) -> str: + _core_id_path = Path(self.config.custom_config['coreIdFilePath']) + if not _core_id_path.exists(): + raise FileExistsError(f'unable to open {_core_id_path}') + _core_id = _core_id_path.read_text() + self.log.debug('roon core id: %s', _core_id) + return _core_id + + @cached_property + def token(self) -> str: + _token_path = Path(self.config.custom_config['tokenFilePath']) + if not _token_path.exists(): + raise FileExistsError(f'unable to open {_token_path}') + _token = _token_path.read_text() + self.log.debug('roon token: %s', _token) + return _token + + @cached_property + def discover(self) -> RoonDiscovery: + return RoonDiscovery(self.core_id) + + @cached_property + def server(self) -> RoonServer: + _server = self.discover.first() + assert isinstance(_server, tuple) and len(_server) == 2, "failed to discover roon server" + self.discover.stop() + roon_server = RoonServer(_server[0], _server[1]) + self.log.info("found %s", roon_server) + return roon_server + + @cached_property + def roonapi(self) -> RoonApi: + _roonapi = RoonApi(self.appinfo, self.token, self.server.ip, self.server.port, True) + return _roonapi + + def my_state_callback(self, event, changed_ids): + """Call when something changes in roon.""" + self.log.info("my_state_callback event:%s changed_ids: %s" % (event, changed_ids)) + for zone_id in changed_ids: + zone = self.roonapi.zones[zone_id] + self.log.info("zone_id:%s zone_info: %s" % (zone_id, zone)) + def run(self) -> bool: - self.log.info("run check code") + self.roonapi.register_state_callback(self.my_state_callback) + return False \ No newline at end of file