edpibu/hasspy
edpibu/hasspy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

Refactor main entry point and command handling in HassSystemClient and HassUserClient

Browse source
This commit is contained in:
Edgar P. Burkhart 2025-03-09 12:59:30 +01:00
parent 9b3df6416e
commit fad234ad00
Signed by: edpibu
GPG key ID: 9833D3C5A25BD227
4 changed files with 104 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

56
hasspy/mqtt.py
View file

@ -1,4 +1,3 @@
import getpass
import json
import logging
from subprocess import run
@ -122,29 +121,26 @@ class HassClient(Client):
class HassSystemClient(HassClient):
commands = {
"POWER_ON": ["systemctl", "poweroff", "--when=cancel"],
"POWER_OFF": ["systemctl", "poweroff", "--when=+1m"],
"LOCK": ["loginctl", "lock-sessions"],
}
def do_command(self, payload: str) -> None:
if payload not in self.commands:
return
super().do_command(payload)
match payload:
case "POWER_ON":
if not self.power_on:
log.info("Cancelling shutdown…")
self.power_on = True
proc = run(self.commands[payload])
if proc.returncode != 0:
log.error(f"Failed to execute command: {payload}")
proc = run(["systemctl", "poweroff", "--when=cancel"])
if proc.returncode != 0:
log.error("Failed to cancel shutdown")
case "POWER_OFF":
if self.power_on:
log.info("Powering off…")
self.power_on = False
proc = run(["systemctl", "poweroff", "--when=+1m"])
if proc.returncode != 0:
log.error("Failed to schedule shutdown")
case "LOCK":
log.info("Locking screen…")
run(["loginctl", "lock-sessions"])
if payload == "POWER_ON":
self.power_on = True
elif payload == "POWER_OFF":
self.power_on = False
@property
def components(self) -> dict[str, dict[str, str]]:
@ -173,16 +169,26 @@ class HassSystemClient(HassClient):
class HassUserClient(HassClient):
commands = {
"PLAY_PAUSE": ["playerctl", "play-pause"],
}
def __init__(self, node_id: str, config: Mapping[str, Any]) -> None:
super().__init__(f"{node_id}_{getpass.getuser()}", config)
super().__init__(f"{node_id}", config)
def do_command(self, payload: str) -> None:
if payload not in self.commands:
return
super().do_command(payload)
match payload:
case "PLAY_PAUSE":
log.info("Toggling play/pause…")
run(["playerctl", "play-pause"])
proc = run(self.commands[payload])
if proc.returncode != 0:
log.error(f"Failed to execute command: {payload}")
@property
def availability_topic(self) -> str:
return f"{self.node_id}/user/availability"
@property
def components(self) -> dict[str, dict[str, str]]: