From 77aa8b5b926e910cc6a1a4634836602293f15bdf Mon Sep 17 00:00:00 2001
From: "Edgar P. Burkhart" <git@edgarpierre.fr>
Date: Sun, 9 Mar 2025 12:26:57 +0100
Subject: [PATCH] Add user mode support with separate client classes for system
and user modes
---
hasspy/__main__.py | 18 +++++--
hasspy/mqtt.py | 116 ++++++++++++++++++++++++++++++++-------------
2 files changed, 97 insertions(+), 37 deletions(-)
diff --git a/hasspy/__main__.py b/hasspy/__main__.py
index a4e0c53..2d63178 100644
--- a/hasspy/__main__.py
+++ b/hasspy/__main__.py
@@ -3,7 +3,7 @@ import tomllib
from argparse import ArgumentParser
from pathlib import Path
-from hasspy.mqtt import HassClient
+from hasspy.mqtt import HassClient, HassSystemClient, HassUserClient
def main() -> None:
@@ -15,6 +15,7 @@ def main() -> None:
parser.add_argument(
"-v", "--verbose", help="Enable verbose logging", action="count", default=0
)
+ parser.add_argument("-u", "--user", help="User mode client", action="store_true")
args = parser.parse_args()
if args.config:
@@ -41,10 +42,17 @@ def main() -> None:
level=config.get("log_level", logging.INFO) - (args.verbose * 10)
)
- ha = HassClient(
- "orchomenos",
- config=config,
- )
+ ha: HassClient
+ if not args.user:
+ ha = HassSystemClient(
+ "orchomenos",
+ config=config,
+ )
+ else:
+ ha = HassUserClient(
+ "orchomenos",
+ config=config,
+ )
ha.loop_forever()
diff --git a/hasspy/mqtt.py b/hasspy/mqtt.py
index 4421384..dd23d48 100644
--- a/hasspy/mqtt.py
+++ b/hasspy/mqtt.py
@@ -1,3 +1,4 @@
+import getpass
import json
import logging
from subprocess import run
@@ -16,11 +17,6 @@ class HassClient(Client):
self.node_id = node_id
self.config = config
- self.state_topic = f"{self.node_id}/state"
- self.availability_topic = f"{self.node_id}/availability"
- self.command_topic = f"{self.node_id}/set"
- self.discovery_topic = f"homeassistant/device/{self.node_id}/config"
-
username = self.config.get("username")
if username:
self.username_pw_set(username, self.config.get("password"))
@@ -62,11 +58,20 @@ class HassClient(Client):
def publish_state(self) -> MQTTMessageInfo:
return self.publish_json(
topic=self.state_topic,
- payload={
- "power": "POWER_ON" if self.power_on else "POWER_OFF",
- },
+ payload=self.state_payload,
)
+ def on_command(self, client: Client, userdata: Any, message: MQTTMessage) -> None:
+ payload = message.payload.decode("utf-8")
+ log.debug(f"Received command: {payload}")
+
+ self.do_command(payload)
+
+ self.publish_state()
+
+ def do_command(self, payload: str) -> None:
+ log.debug(f"Executing command: {payload}")
+
def on_connect(self, *args: Any, **kwargs: Any) -> None:
log.info("Connected to MQTT broker")
self.publish_discovery()
@@ -75,9 +80,50 @@ class HassClient(Client):
self.publish_state()
- def on_command(self, client: Client, userdata: Any, message: MQTTMessage) -> None:
- payload = message.payload.decode("utf-8")
- log.debug(f"Received command: {payload}")
+ @property
+ def state_topic(self) -> str:
+ return f"{self.node_id}/state"
+
+ @property
+ def availability_topic(self) -> str:
+ return f"{self.node_id}/availability"
+
+ @property
+ def command_topic(self) -> str:
+ return f"{self.node_id}/set"
+
+ @property
+ def discovery_topic(self) -> str:
+ return f"homeassistant/device/{self.node_id}/config"
+
+ @property
+ def discovery_payload(self) -> dict[str, Any]:
+ return {
+ "dev": {
+ "ids": self.node_id,
+ "name": self.node_id,
+ },
+ "o": {
+ "name": "hasspy",
+ },
+ "cmps": self.components,
+ "availability_topic": self.availability_topic,
+ "command_topic": self.command_topic,
+ "state_topic": self.state_topic,
+ }
+
+ @property
+ def state_payload(self) -> dict[str, Any]:
+ return {}
+
+ @property
+ def components(self) -> dict[str, dict[str, str]]:
+ return {}
+
+
+class HassSystemClient(HassClient):
+ def do_command(self, payload: str) -> None:
+ super().do_command(payload)
match payload:
case "POWER_ON":
@@ -96,30 +142,36 @@ class HassClient(Client):
proc = run(["systemctl", "poweroff", "--when=+1m"])
if proc.returncode != 0:
log.error("Failed to schedule shutdown")
-
- self.publish_state()
+ case "LOCK":
+ log.info("Locking screen…")
+ run(["loginctl", "lock-sessions"])
@property
- def discovery_payload(self) -> dict[str, Any]:
+ def components(self) -> dict[str, dict[str, str]]:
return {
- "dev": {
- "ids": self.node_id,
- "name": self.node_id,
+ "power": {
+ "unique_id": f"{self.node_id}_power",
+ "p": "switch",
+ "name": "Power",
+ "payload_off": "POWER_OFF",
+ "payload_on": "POWER_ON",
+ "value_template": "{{ value_json.power }}",
},
- "o": {
- "name": "hasspy",
+ "lock": {
+ "unique_id": f"{self.node_id}_power",
+ "p": "button",
+ "name": "Lock",
+ "payload_press": "LOCK",
},
- "cmps": {
- "power": {
- "unique_id": f"{self.node_id}_power",
- "p": "switch",
- "name": "Power",
- "payload_off": "POWER_OFF",
- "payload_on": "POWER_ON",
- "value_template": "{{ value_json.power }}",
- },
- },
- "availability_topic": self.availability_topic,
- "command_topic": self.command_topic,
- "state_topic": self.state_topic,
}
+
+ @property
+ def state_payload(self) -> dict[str, Any]:
+ return {
+ "power": self.power_on,
+ }
+
+
+class HassUserClient(HassClient):
+ def __init__(self, node_id: str, config: Mapping[str, Any]) -> None:
+ super().__init__(f"{node_id}_{getpass.getuser()}", config)