"""ISY Network Resources Module."""
from asyncio import sleep
from xml.dom import minidom
from .constants import (
_LOGGER,
ATTR_ID,
TAG_NAME,
TAG_NET_RULE,
URL_NETWORK,
URL_RESOURCES,
)
from .exceptions import XML_ERRORS, XML_PARSE_ERROR, ISYResponseParseError
from .helpers import value_from_xml
[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, xml=None):
"""
Initialize the network resources class.
isy: ISY class
xml: String of xml data containing the configuration data
"""
self.isy = isy
self.addresses = []
self.nnames = []
self.nobjs = []
if xml is not None:
self.parse(xml)
[docs] def parse(self, xml):
"""
Parse the xml data.
xml: String of the xml data
"""
try:
xmldoc = minidom.parseString(xml)
except XML_ERRORS:
_LOGGER.error("%s: NetworkResources", XML_PARSE_ERROR)
raise ISYResponseParseError(XML_PARSE_ERROR)
features = xmldoc.getElementsByTagName(TAG_NET_RULE)
for feature in features:
address = int(value_from_xml(feature, ATTR_ID))
if address not in self.addresses:
nname = value_from_xml(feature, TAG_NAME)
nobj = NetworkCommand(self, address)
self.addresses.append(address)
self.nnames.append(nname)
self.nobjs.append(nobj)
_LOGGER.info("ISY Loaded Network Resources Commands")
[docs] async def update(self, wait_time=0):
"""
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):
"""
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):
"""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 None
[docs] def get_by_id(self, val):
"""
Return command object being given a command id.
val: Integer representing command id
"""
try:
ind = self.addresses.index(val)
return self.get_by_index(ind)
except (ValueError, KeyError):
return None
[docs] def get_by_name(self, val):
"""
Return command object being given a command name.
val: String representing command name
"""
try:
ind = self.nnames.index(val)
return self.get_by_index(ind)
except (ValueError, KeyError):
return None
[docs] def get_by_index(self, val):
"""
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, address):
"""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
@property
def address(self):
"""Return the Resource ID for the Network Resource."""
return self._id
[docs] async def run(self):
"""Execute the networking command."""
req_url = self.isy.conn.compile_url([URL_NETWORK, URL_RESOURCES, str(self._id)])
if not await self.isy.conn.request(req_url, ok404=True):
_LOGGER.warning("ISY could not run networking command: %s", str(self._id))
return
_LOGGER.debug("ISY ran networking command: %s", str(self._id))