91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
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, Union
|
|
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)
|
|
self.set_status()
|
|
|
|
def set_status(self) -> None:
|
|
zones_state = list()
|
|
|
|
for zone_id in self.roonapi.zones:
|
|
self.log.debug(self.roonapi.zones[zone_id])
|
|
zone = self.roonapi.zones[zone_id]
|
|
self.log.debug("zone_id: %s zone_info: %s" , zone_id, zone)
|
|
zone_name = zone['display_name']
|
|
zone_state = zone['state']
|
|
self.log.info("zone_id: %s name: %s, state: %s" , zone_id, zone_name, zone_state)
|
|
zones_state.append(zone_state)
|
|
|
|
self.active = True if "playing" in zones_state else 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 roon_state_callback(self, event: str, changed_ids: Union[str, list[str]]) -> None:
|
|
"""Call when something changes in roon."""
|
|
self.log.debug(
|
|
"%s event: %s changed_ids: %s" , RoonTrigger.roon_state_callback.__qualname__, event, changed_ids
|
|
)
|
|
self.set_status()
|
|
|
|
async def async_run(self) -> None:
|
|
self.roonapi.register_state_callback(self.roon_state_callback, "zones_changed") |