Source code for pyisy.networking

"""ISY Network Resources Module."""

from __future__ import annotations

from asyncio import sleep
from typing import TYPE_CHECKING
from xml.dom import minidom

from .constants import (
    ATTR_ID,
    PROTO_NETWORK,
    TAG_NAME,
    TAG_NET_RULE,
    URL_NETWORK,
    URL_RESOURCES,
)
from .exceptions import XML_ERRORS, XML_PARSE_ERROR
from .helpers import value_from_xml
from .logging import _LOGGER

if TYPE_CHECKING:
    from .isy import ISY


[docs] class NetworkResources: """ Network Resources class cobject. DESCRIPTION: This class handles the ISY networking module. USAGE: This object may be used in a similar way as a dictionary with the either networking command names or ids being used as keys and the ISY networking command class will be returned. EXAMPLE: # a = networking['test function'] # a.run() ATTRIBUTES: isy: The ISY device class addresses: List of net command ids nnames: List of net command names nobjs: List of net command objects """
[docs] def __init__(self, isy: ISY, xml: str | None = None) -> None: """ Initialize the network resources class. isy: ISY class xml: String of xml data containing the configuration data """ self.isy = isy self.addresses: list[int] = [] self._address_index: dict[int, int] = {} self.nnames: list[str] = [] self._nnames_index: dict[str, int] = {} self.nobjs: list[NetworkCommand] = [] if xml is not None: self.parse(xml)
[docs] def parse(self, xml: str) -> None: """ Parse the xml data. xml: String of the xml data """ try: xmldoc = minidom.parseString(xml) except XML_ERRORS: _LOGGER.error("%s: NetworkResources, resources not loaded", XML_PARSE_ERROR) return features = xmldoc.getElementsByTagName(TAG_NET_RULE) for feature in features: address = int(value_from_xml(feature, ATTR_ID)) if address in self._address_index: continue nname = value_from_xml(feature, TAG_NAME) nobj = NetworkCommand(self, address, nname) self.addresses.append(address) self._address_index[address] = len(self.addresses) - 1 self.nnames.append(nname) self._nnames_index[nname] = len(self.nnames) - 1 self.nobjs.append(nobj) _LOGGER.info("ISY Loaded Network Resources Commands")
[docs] async def update(self, wait_time: int = 0) -> None: """ Update the contents of the networking class. wait_time: [optional] Amount of seconds to wait before updating """ await sleep(wait_time) xml = await self.isy.conn.get_network() self.parse(xml)
[docs] async def update_threaded(self, interval: int) -> None: """ Continually update the class until it is told to stop. Should be run in a thread. """ while self.isy.auto_update: await self.update(interval)
[docs] def __getitem__(self, val: str | int) -> NetworkCommand | None: """Return the item from the collection.""" try: val = int(val) return self.get_by_id(val) except (ValueError, KeyError): return self.get_by_name(val)
[docs] def __setitem__(self, val, value): """Set the item value.""" return
[docs] def get_by_id(self, val: int) -> NetworkCommand | None: """ Return command object being given a command id. val: Integer representing command id """ ind = self._address_index.get(val) return None if ind is None else self.get_by_index(ind)
[docs] def get_by_name(self, val: str) -> NetworkCommand | None: """ Return command object being given a command name. val: String representing command name """ ind = self._nnames_index.get(val) return None if ind is None else self.get_by_index(ind)
[docs] def get_by_index(self, val: int) -> NetworkCommand | None: """ Return command object being given a command index. val: Integer representing command index in List """ return self.nobjs[val]
[docs] class NetworkCommand: """ Network Command Class. DESCRIPTION: This class handles individual networking commands. ATTRIBUTES: network_resources: The networkin resources class """
[docs] def __init__(self, network_resources: NetworkResources, address: int, name: str) -> None: """Initialize network command class. network_resources: NetworkResources class address: Integer of the command id """ self._network_resources = network_resources self.isy = network_resources.isy self._id = address self._name = name
@property def address(self) -> str: """Return the Resource ID for the Network Resource.""" return str(self._id) @property def name(self) -> str: """Return the name of this entity.""" return self._name @property def protocol(self) -> str: """Return the Protocol for this node.""" return PROTO_NETWORK
[docs] async def run(self) -> None: """Execute the networking command.""" address = self.address req_url = self.isy.conn.compile_url([URL_NETWORK, URL_RESOURCES, address]) if not await self.isy.conn.request(req_url, ok404=True): _LOGGER.warning("ISY could not run networking command: %s", address) return _LOGGER.debug("ISY ran networking command: %s", address)