Skip to content

Commit

Permalink
fix: minor api cleanup
Browse files Browse the repository at this point in the history
  • Loading branch information
joshuagruenstein committed Sep 22, 2024
1 parent e60e19c commit 1983874
Show file tree
Hide file tree
Showing 2 changed files with 21 additions and 62 deletions.
3 changes: 2 additions & 1 deletion examples/example.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from aiohttp import BasicAuth

from intellinet_pdu_ctrl.api import IPU
from intellinet_pdu_ctrl.types import OutletCommand


async def main() -> None:
Expand All @@ -13,7 +14,7 @@ async def main() -> None:
os.environ.get("PDU_USER", "admin"), os.environ.get("PDU_PASS", "admin")
),
) as ipu:
print(await ipu.get_status())
await ipu.set_outlets(OutletCommand.ON, 0)


if __name__ == "__main__":
Expand Down
80 changes: 19 additions & 61 deletions intellinet_pdu_ctrl/api.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
from dataclasses import asdict
from enum import Enum
from types import TracebackType
from typing import ClassVar, Type
from typing import Any, ClassVar, Type

import aiohttp
from lxml import etree as et
Expand All @@ -26,16 +25,6 @@ class PDUEndpoints(Enum):


class IPU:

"""This class is represents a api wrapper for the Intellinet IP smart PDU API [163682].
It provides all the functionality of the web interface it is based on.
Class-Attributes:
DEFAULT_CREDS (:obj:`tuple` of :obj:`str`): default username/password of pdu
DEFAULT_ENDCODING (str): default encoding of pdu
DEFAULT_SCHEMA (str): default schema of pdu
"""

DEFAULT_CREDS: ClassVar[aiohttp.BasicAuth] = aiohttp.BasicAuth("admin", "admin")

def __init__(
Expand All @@ -62,82 +51,51 @@ async def __aexit__(
async def _get_request(
self, page: PDUEndpoints, params: dict[str, str] | None = None
) -> et._Element:
"""Internal wrapper around aiohttp get method and the pdus available endpoints.
Args:
page (str): endpoint / page that is requested
params (dict, optional): get parameters to be sent along with request. Used for updating settings.
Returns:
:obj:`lxml.etree._Element`: The parsed XML/HTML element.
"""
async with self.session.get(page.value, params=params) as resp:
raw_resp_content = await resp.text()

parser = et.HTML if "html" in raw_resp_content.lower() else et.XML
return parser(raw_resp_content) # type: ignore

async def _post_request(
self, page: PDUEndpoints, data: dict[str, str]
self, page: PDUEndpoints, data: dict[str, Any]
) -> aiohttp.ClientResponse:
"""Internal wrapper around aiohttp post method and the pdus available endpoints.
Args:
page (str): See: self._get_request()
data (dict): post data
"""
headers = {"Content-type": "application/x-www-form-urlencoded"}
return await self.session.post(page.value, data=data, headers=headers)
return await self.session.post(
page.value,
data=data,
headers={"Content-type": "application/x-www-form-urlencoded"},
)

async def get_status(self) -> PDUStatus:
"""gives you basic status/health of the device.
Values: deg. C, outlet states [on/off], status [read: are there warnings?], humidity in perc, amps.
Returns:
dict: containing the aforementioned stats.
e.g. {'degree_celcius': '26', 'outlet_states': ['on', 'on', 'off', 'on', 'on', 'on', 'on', 'on'],
'stat': 'normal', 'humidity_percent': '27', 'current_amperes': '0.5'}
"""
e = await self._get_request(PDUEndpoints.status)
return PDUStatus.from_xml(e)

async def set_config_pdu(self, outlet_configs: AllOutletsConfig) -> None:
"""Setter for self.pdu_config()
Args:
outlet_configs (dict): dict that is formatted like the output of self._get_config_pdu()
"""
translation_table = {
"turn_on_delay": "ondly",
"turn_off_delay": "ofdly",
"name": "otlt",
}

settings = {}
for otl_nr, v in enumerate(outlet_configs.outlets):
for _k, _v in asdict(v).items():
new_key = f"{translation_table[_k]}{otl_nr}"
settings[new_key] = _v
async def set_outlets_config(self, outlet_configs: AllOutletsConfig) -> None:
settings = dict[str, Any]()
for o, v in enumerate(outlet_configs.outlets):
settings[f"otlt{o}"] = v.name
settings[f"ofdly{o}"] = v.turn_off_delay
settings[f"ondly{o}"] = v.turn_on_delay

await self._post_request(PDUEndpoints.config_pdu, data=settings)

async def get_config_pdu(self) -> AllOutletsConfig:
async def get_outlets_config(self) -> AllOutletsConfig:
etree = await self._get_request(PDUEndpoints.config_pdu)

return AllOutletsConfig.from_xml(etree)

async def get_config_threshold(self) -> ThresholdsConfig:
async def get_thresholds_config(self) -> ThresholdsConfig:
etree = await self._get_request(PDUEndpoints.thresholds)
return ThresholdsConfig.from_xml(etree)

async def set_config_threshold(self, threshold_config: ThresholdsConfig) -> None:
async def set_thresholds_config(self, threshold_config: ThresholdsConfig) -> None:
await self._post_request(
PDUEndpoints.thresholds, data=threshold_config.to_dict()
)

async def set_outlets_state(
self, state: OutletCommand, *list_of_outlet_ids: int
) -> None:
outlet_states = {"outlet{}".format(k): str(1) for k in list_of_outlet_ids}
async def set_outlets(self, state: OutletCommand, *list_of_outlet_ids: int) -> None:
outlet_states = {f"outlet{k}": str(1) for k in list_of_outlet_ids}
outlet_states["op"] = str(state.value)
outlet_states["submit"] = "Anwenden"

await self._get_request(PDUEndpoints.outlet, params=outlet_states)

0 comments on commit 1983874

Please sign in to comment.