Compare commits
No commits in common. "9261f8af49798f6378621b09a9ef42bf43f745b5" and "e224438cf343b0c0ace69474ffbac7d5eccc9714" have entirely different histories.
9261f8af49
...
e224438cf3
1 changed files with 18 additions and 44 deletions
|
@ -1,21 +1,17 @@
|
|||
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"
|
||||
ON = "ON"
|
||||
OFF = "OFF"
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
|
@ -42,7 +38,7 @@ class Switch:
|
|||
f"SW{outlet}.y": 1,
|
||||
}
|
||||
if from_state:
|
||||
params[f"STATUS{outlet}"] = f"{from_state.value.upper():3s}"
|
||||
params[f"STATUS{outlet}"] = f"{from_state.value:3s}"
|
||||
|
||||
logging.info("Switching outlet: %s", params)
|
||||
response = requests.post(
|
||||
|
@ -54,63 +50,41 @@ class Switch:
|
|||
print(response)
|
||||
|
||||
def switch_on(self, switch_id: int, outlet: int) -> None:
|
||||
return self.switch(switch_id, outlet, from_state=SwitchStatus.off)
|
||||
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)
|
||||
}
|
||||
return self.switch(switch_id, outlet, from_state=SwitchStatus.ON)
|
||||
|
||||
|
||||
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)]
|
||||
|
||||
|
||||
class SwitchAction(str, enum.Enum):
|
||||
on = "on"
|
||||
off = "off"
|
||||
toggle = "toggle"
|
||||
|
||||
|
||||
@app.command()
|
||||
def switch(
|
||||
context: typer.Context,
|
||||
switch_id: SwitchIdArgument,
|
||||
outlet: OutletArgument,
|
||||
action: typing.Annotated[SwitchStatus, typer.Argument(case_sensitive=False)],
|
||||
action: typing.Annotated[
|
||||
SwitchAction, typer.Argument(case_sensitive=False)
|
||||
] = SwitchAction.toggle,
|
||||
) -> None:
|
||||
switch: Switch = context.obj
|
||||
match action:
|
||||
case SwitchStatus.on:
|
||||
case SwitchAction.on:
|
||||
switch.switch_on(switch_id, outlet)
|
||||
case SwitchStatus.off:
|
||||
case SwitchAction.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)
|
||||
case SwitchAction.toggle:
|
||||
switch.switch(switch_id, outlet)
|
||||
|
||||
|
||||
@app.callback()
|
||||
|
|
Loading…
Reference in a new issue