power: refactor component in order to dynamically import plugins

moonraker/extras/__init__.py

@@ -0,0 +1 @@
+# package definition for thirdparty package

moonraker/extras/power/__init__.py

@@ -0,0 +1,7 @@
+# package definition for thirdparty package
+import glob
+from os.path import dirname, basename, isfile, join
+
+modules = glob.glob(join(dirname(__file__), "*.py"))
+__all__ = [basename(f)[:-3]
+           for f in modules if isfile(f) and not f.endswith('__init__.py')]

moonraker/extras/power/gpio.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Optional
+
+from moonraker.components.power import PowerDevice, PrinterPower
+from moonraker.confighelper import ConfigHelper
+
+
+class GpioDevice(PowerDevice):
+    def __init__(self,
+                 config: ConfigHelper,
+                 initial_val: Optional[int] = None
+                 ) -> None:
+        super().__init__(config)
+        self.timer: Optional[float] = config.getfloat('timer', None)
+        if self.timer is not None and self.timer < 0.000001:
+            raise config.error(
+                f"Option 'timer' in section [{config.get_name()}] must "
+                "be above 0.0")
+        self.timer_handle: Optional[asyncio.TimerHandle] = None
+        if initial_val is None:
+            initial_val = int(self.initial_state or 0)
+        self.gpio_out = config.getgpioout('pin', initial_value=initial_val)
+
+    async def init_state(self) -> None:
+        if self.initial_state is None:
+            self.set_power("off")
+        else:
+            self.set_power("on" if self.initial_state else "off")
+            await self.process_bound_services()
+
+    def refresh_status(self) -> None:
+        pass
+
+    def set_power(self, state) -> None:
+        if self.timer_handle is not None:
+            self.timer_handle.cancel()
+            self.timer_handle = None
+        try:
+            self.gpio_out.write(int(state == "on"))
+        except Exception:
+            self.state = "error"
+            msg = f"Error Toggling Device Power: {self.name}"
+            logging.exception(msg)
+            raise self.server.error(msg) from None
+        self.state = state
+        self._check_timer()
+
+    def _check_timer(self) -> None:
+        if self.state == "on" and self.timer is not None:
+            event_loop = self.server.get_event_loop()
+            power: PrinterPower = self.server.lookup_component("power")
+            self.timer_handle = event_loop.delay_callback(
+                self.timer, power.set_device_power, self.name, "off")
+
+    def close(self) -> None:
+        if self.timer_handle is not None:
+            self.timer_handle.cancel()
+            self.timer_handle = None

moonraker/extras/power/homeassistant.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Dict, Any, Optional, cast
+from urllib.parse import quote
+
+from moonraker.components.power import HTTPDevice
+from moonraker.confighelper import ConfigHelper
+
+
+class HomeAssistant(HTTPDevice):
+    def __init__(self, config: ConfigHelper) -> None:
+        super().__init__(config, default_port=8123)
+        self.device: str = config.get("device")
+        self.token: str = config.gettemplate("token").render()
+        self.domain: str = config.get("domain", "switch")
+        self.status_delay: float = config.getfloat("status_delay", 1.)
+
+    async def _send_homeassistant_command(self, command: str) -> Dict[str, Any]:
+        body: Optional[Dict[str, Any]] = None
+        if command in ["on", "off"]:
+            out_cmd = f"api/services/{quote(self.domain)}/turn_{command}"
+            body = {"entity_id": self.device}
+            method = "POST"
+        elif command == "info":
+            out_cmd = f"api/states/{quote(self.device)}"
+            method = "GET"
+        else:
+            raise self.server.error(
+                f"Invalid homeassistant command: {command}")
+        url = f"{self.protocol}://{quote(self.addr)}:{self.port}/{out_cmd}"
+        headers = {
+            'Authorization': f'Bearer {self.token}'
+        }
+        data: Dict[str, Any] = {}
+        response = await self.client.request(
+            method, url, body=body, headers=headers,
+            attempts=3, enable_cache=False
+        )
+        msg = f"Error sending homeassistant command: {command}"
+        response.raise_for_status(msg)
+        if method == "GET":
+            data = cast(dict, response.json())
+        return data
+
+    async def _send_status_request(self) -> str:
+        res = await self._send_homeassistant_command("info")
+        return res["state"]
+
+    async def _send_power_request(self, state: str) -> str:
+        await self._send_homeassistant_command(state)
+        await asyncio.sleep(self.status_delay)
+        res = await self._send_status_request()
+        return res

moonraker/extras/power/homeseer.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import Dict, Any
+from urllib.parse import urlencode, quote
+
+from moonraker.components.power import HTTPDevice
+from moonraker.confighelper import ConfigHelper
+
+
+class HomeSeer(HTTPDevice):
+    def __init__(self, config: ConfigHelper) -> None:
+        super().__init__(config, default_user="admin", default_password="")
+        self.device = config.getint("device")
+
+    async def _send_homeseer(
+        self, request: str, state: str = ""
+    ) -> Dict[str, Any]:
+        query_args = {
+            "user": self.user,
+            "pass": self.password,
+            "request": request,
+            "ref": self.device,
+        }
+        if state:
+            query_args["label"] = state
+        query = urlencode(query_args)
+        url = (
+            f"{self.protocol}://{quote(self.user)}:{quote(self.password)}@"
+            f"{quote(self.addr)}/JSON?{query}"
+        )
+        return await self._send_http_command(url, request)
+
+    async def _send_status_request(self) -> str:
+        res = await self._send_homeseer("getstatus")
+        return res["Devices"][0]["status"].lower()
+
+    async def _send_power_request(self, state: str) -> str:
+        await self._send_homeseer(
+            "controldevicebylabel", state.capitalize()
+        )
+        return state

moonraker/extras/power/http.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Dict, Any
+
+from moonraker.components.power import HTTPDevice
+from moonraker.confighelper import ConfigHelper
+
+
+class GenericHTTP(HTTPDevice):
+    def __init__(self, config: ConfigHelper,) -> None:
+        super().__init__(config, is_generic=True)
+        self.urls: Dict[str, str] = {
+            "on": config.gettemplate("on_url").render(),
+            "off": config.gettemplate("off_url").render(),
+            "status": config.gettemplate("status_url").render()
+        }
+        self.request_template = config.gettemplate(
+            "request_template", None, is_async=True
+        )
+        self.response_template = config.gettemplate("response_template", is_async=True)
+
+    async def _send_generic_request(self, command: str) -> str:
+        request = self.client.wrap_request(
+            self.urls[command], request_timeout=20., attempts=3, retry_pause_time=1.
+        )
+        context: Dict[str, Any] = {
+            "command": command,
+            "http_request": request,
+            "async_sleep": asyncio.sleep,
+            "log_debug": logging.debug,
+            "urls": dict(self.urls)
+        }
+        if self.request_template is not None:
+            await self.request_template.render_async(context)
+            response = request.last_response()
+            if response is None:
+                raise self.server.error("Failed to receive a response")
+        else:
+            response = await request.send()
+        response.raise_for_status()
+        result = (await self.response_template.render_async(context)).lower()
+        if result not in ["on", "off"]:
+            raise self.server.error(f"Invalid result: {result}")
+        return result
+
+    async def _send_power_request(self, state: str) -> str:
+        return await self._send_generic_request(state)
+
+    async def _send_status_request(self) -> str:
+        return await self._send_generic_request("status")

moonraker/extras/power/hue.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from typing import cast, List, Dict, Any
+from urllib.parse import quote
+
+from moonraker.components.power import HTTPDevice
+from moonraker.confighelper import ConfigHelper
+
+
+class HueDevice(HTTPDevice):
+
+    def __init__(self, config: ConfigHelper) -> None:
+        super().__init__(config)
+        self.device_id = config.get("device_id")
+        self.device_type = config.get("device_type", "light")
+        if self.device_type == "group":
+            self.state_key = "action"
+            self.on_state = "all_on"
+        else:
+            self.state_key = "state"
+            self.on_state = "on"
+
+    async def _send_power_request(self, state: str) -> str:
+        new_state = True if state == "on" else False
+        url = (
+            f"{self.protocol}://{quote(self.addr)}/api/{quote(self.user)}"
+            f"/{self.device_type}s/{quote(self.device_id)}"
+            f"/{quote(self.state_key)}"
+        )
+        ret = await self.client.request("PUT", url, body={"on": new_state})
+        resp = cast(List[Dict[str, Dict[str, Any]]], ret.json())
+        state_url = (
+            f"/{self.device_type}s/{self.device_id}/{self.state_key}/on"
+        )
+        return (
+            "on" if resp[0]["success"][state_url]
+            else "off"
+        )
+
+    async def _send_status_request(self) -> str:
+        url = (
+            f"{self.protocol}://{quote(self.addr)}/api/{quote(self.user)}"
+            f"/{self.device_type}s/{quote(self.device_id)}"
+        )
+        ret = await self.client.request("GET", url)
+        resp = cast(Dict[str, Dict[str, Any]], ret.json())
+        return "on" if resp["state"][self.on_state] else "off"