ipowerswitch/ipowerswitch.py

import dataclasses
import enum
import logging
import re
import typing

import bs4
import requests
import requests.auth
import rich.console
import rich.table
import typer
import yarl


class SwitchStatus(str, enum.Enum):
    on = "on"
    off = "off"


@dataclasses.dataclass
class SwitchConfiguration:
    outlet: int
    name: str
    power_resume_delay: int
    ring_on_reset: bool
    safe_shutdown: bool
    safe_reboot: bool
    shutdown_delay: int


@dataclasses.dataclass
class Switch:
    address: str
    user: str
    password: str

    def switch(
        self,
        switch_id: int,
        outlet: int,
        from_state: SwitchStatus | None = None,
    ) -> None:
        url = yarl.URL(f"http://{self.address}").joinpath(
            "cgi-bin",
            f"iswitch{switch_id:02d}",
            "irswitch.exe",
        )

        params = {
            "CURRENT": f"{switch_id:02d}",
            f"SW{outlet}.x": 1,
            f"SW{outlet}.y": 1,
        }
        if from_state:
            params[f"STATUS{outlet}"] = f"{from_state.value.upper():3s}"

        logging.info("Switching outlet: %s", params)
        response = requests.post(
            str(url),
            auth=requests.auth.HTTPBasicAuth(self.user, self.password),
            data=params,
            allow_redirects=False,
        )
        print(response)

    def switch_on(self, switch_id: int, outlet: int) -> None:
        return self.switch(switch_id, outlet, from_state=SwitchStatus.off)

    def switch_off(self, switch_id: int, outlet: int) -> None:
        return self.switch(switch_id, outlet, from_state=SwitchStatus.on)

    def switch_status(self, switch_id: int) -> dict[int, SwitchStatus]:
        url = yarl.URL(f"http://{self.address}").joinpath(
            f"iswitch{switch_id:02d}",
            "index.htm",
        )
        response = requests.get(
            str(url),
            auth=requests.auth.HTTPBasicAuth(self.user, self.password),
        )
        soup = bs4.BeautifulSoup(response.content, features="html.parser")
        inputs = {i["name"]: i["value"] for i in soup.find_all("input", type="hidden")}
        outlets = int(inputs["OUTLET"])
        return {
            outlet: SwitchStatus[inputs.get(f"STATUS{outlet}", "off").strip().lower()]
            for outlet in range(1, outlets + 1)
        }

    def switch_config(self, switch_id: int) -> dict[int, SwitchConfiguration]:
        url = yarl.URL(f"http://{self.address}").joinpath(
            f"iswitch{switch_id:02d}",
            "config.htm",
        )
        response = requests.get(
            str(url),
            auth=requests.auth.HTTPBasicAuth(self.user, self.password),
        )
        soup = bs4.BeautifulSoup(response.content, features="html.parser")
        inputs = {i["name"]: i for i in soup.find_all("input")}
        outlets = int(inputs["OUTLET"]["value"])
        configs = {}
        for outlet in range(1, outlets + 1):
            letter = chr(ord("a") + (outlets - 1))
            config = SwitchConfiguration(
                outlet=outlet,
                name=inputs[f"T{outlet}"]["value"].strip(),
                power_resume_delay=int(inputs[f"power_{letter}"]["value"]),
                ring_on_reset=inputs[f"act{outlet}"].has_attr("checked"),
                safe_shutdown=inputs[f"shut{outlet}"].has_attr("checked"),
                safe_reboot=inputs[f"rbt{outlet}"].has_attr("checked"),
                shutdown_delay=int(inputs[f"dlt{letter}"]["value"]),
            )
            configs[outlet] = config
        return configs


app = typer.Typer()
console = rich.console.Console()

SwitchIdArgument = typing.Annotated[int, typer.Argument(min=1, max=16)]
OutletArgument = typing.Annotated[int, typer.Argument(min=1, max=8)]


@app.command()
def switch(
    context: typer.Context,
    switch_id: SwitchIdArgument,
    outlet: OutletArgument,
    action: typing.Annotated[SwitchStatus, typer.Argument(case_sensitive=False)],
) -> None:
    switch: Switch = context.obj
    match action:
        case SwitchStatus.on:
            switch.switch_on(switch_id, outlet)
        case SwitchStatus.off:
            switch.switch_off(switch_id, outlet)


@app.command()
def status(context: typer.Context, switch_id: SwitchIdArgument) -> None:
    switch: Switch = context.obj
    statuses = switch.switch_status(switch_id)
    if not console.is_interactive:
        for key in sorted(statuses.keys()):
            console.print(f"{key}\t{statuses[key].value.upper()}")
    else:
        table = rich.table.Table("Outlet", "Status")
        for key in sorted(statuses.keys()):
            table.add_row(str(key), statuses[key].value.upper())
        console.print(table)


@app.command()
def config(context: typer.Context, switch_id: SwitchIdArgument) -> None:
    switch: Switch = context.obj
    configs = switch.switch_config(switch_id)

    def y_n(value: bool) -> str:
        return "Y" if value else "N"

    if not console.is_interactive:
        for key in sorted(configs.keys()):
            config = configs[key]
            console.print(
                "\t".join(
                    [
                        str(outlet),
                        config.name,
                        str(config.power_resume_delay),
                        y_n(config.ring_on_reset),
                        y_n(config.safe_shutdown),
                        y_n(config.safe_reboot),
                        str(config.shutdown_delay),
                    ]
                )
            )
    else:
        table = rich.table.Table(
            "Outlet",
            "Name",
            "Power Resume Delay",
            "Ring On/Reset",
            "Safe Shutdown/Reboot",
        )
        for key in sorted(configs.keys()):
            config = configs[key]
            table.add_row(
                str(key),
                config.name,
                f"{config.power_resume_delay} sec",
                y_n(config.ring_on_reset),
                "[{} / {}] {}".format(
                    y_n(config.safe_shutdown),
                    y_n(config.safe_reboot),
                    f"{config.shutdown_delay} sec",
                ),
            )
        console.print(table)


@app.callback()
def main(
    context: typer.Context,
    address: str,
    user: str = "admin",
    password: str = "admin",
) -> None:
    logging.basicConfig(level=logging.INFO)
    context.obj = Switch(
        address=address,
        user=user,
        password=password,
    )


if __name__ == "__main__":
    app()