Get rid of Chump dependency for Pushover

paradox/interfaces/text/pushover.py

@@ -1,7 +1,7 @@
+import http.client
 import logging
 import re
-
-import chump
+import urllib
 
 from paradox.config import config as cfg
 from paradox.event import EventLevel
@@ -10,12 +10,12 @@
 logger = logging.getLogger("PAI").getChild(__name__)
 
 _level_2_priority = {
-    EventLevel.NOTSET: chump.LOWEST,
-    EventLevel.DEBUG: chump.LOWEST,
-    EventLevel.INFO: chump.LOW,
-    EventLevel.WARN: chump.NORMAL,
-    EventLevel.ERROR: chump.HIGH,
-    EventLevel.CRITICAL: chump.EMERGENCY,
+    EventLevel.NOTSET: -2,
+    EventLevel.DEBUG: -2,
+    EventLevel.INFO: -1,
+    EventLevel.WARN: 0,
+    EventLevel.ERROR: 1,
+    EventLevel.CRITICAL: 2,
 }
 
 
@@ -31,66 +31,45 @@ def __init__(self, alarm):
             cfg.PUSHOVER_MIN_EVENT_LEVEL,
         )
 
-        self.app = None
         self.users = {}
 
-    def _run(self):
-        super()._run()
-
-        self.app = chump.Application(cfg.PUSHOVER_KEY)
-        if not self.app.is_authenticated:
-            raise Exception(
-                "Failed to authenticate with Pushover. Please check PUSHOVER_APPLICATION_KEY"
-            )
-
     def send_message(self, message: str, level: EventLevel):
         for settings in cfg.PUSHOVER_BROADCAST_KEYS:
             user_key = settings["user_key"]
             devices_raw = settings["devices"]
 
-            user = self.users.get(user_key)
-
-            if user is None:
-                user = self.users[user_key] = self.app.get_user(user_key)
-
-            if not user.is_authenticated:
-                raise Exception(
-                    "Failed to check user key with Pushover. Please check PUSHOVER_BROADCAST_KEYS[%s]"
-                    % user_key
-                )
-
             if devices_raw == "*" or devices_raw is None:
-                try:
-                    user.send_message(
-                        message,
-                        title="Alarm",
-                        priority=_level_2_priority.get(level, chump.NORMAL),
-                    )
-                    logger.info(f"Notification sent: {message}, level={level}")
-                except Exception:
-                    logger.exception("Pushover send message")
-
+                self._send_pushover_message(user_key, message, level)
             else:
                 devices = list(filter(bool, re.split(r"[\s]*,[\s]*", devices_raw)))
 
-                for elem in (elem for elem in devices if elem not in user.devices):
-                    logger.warning(
-                        "%s is not in the Pushover device list for the user %s"
-                        % (elem, user_key)
-                    )
-
                 for device in devices:
-                    try:
-                        user.send_message(
-                            message,
-                            title="PAI",
-                            device=device,
-                            priority=_level_2_priority.get(level, chump.NORMAL),
-                        )
-                        logger.info(
-                            f"Notification sent: {message}, level={level}, device={device}"
-                        )
-                    except Exception:
-                        logger.exception("Pushover send message")
+                    self._send_pushover_message(user_key, message, level, device)
+
+    def _send_pushover_message(self, user_key, message, level, device=None):
+        conn = http.client.HTTPSConnection("api.pushover.net:443")
+        params = {
+            "token": cfg.PUSHOVER_KEY,
+            "user": user_key,
+            "message": message,
+            "priority": _level_2_priority.get(level, 0),
+            "title": "Alarm",
+        }
+        if device:
+            params["device"] = device
+
+        conn.request(
+            "POST",
+            "/1/messages.json",
+            urllib.parse.urlencode(params),
+            {"Content-type": "application/x-www-form-urlencoded"},
+        )
 
-        # TODO: Missing the message reception
+        response = conn.getresponse()
+        if response.status != 200:
+            logger.error(f"Failed to send message: {response.reason}")
+        else:
+            logger.info(
+                f"Notification sent: {message}, level={level}, device={device if device else 'all'}"
+            )
+        conn.close()

paradox/lib/help.py

@@ -36,9 +36,6 @@
     "yaml": dict(
         mandatory=False, desc="the IP150 connection", install_name="pyyaml>=5.2.0"
     ),
-    "chump": dict(
-        mandatory=False, desc="the Pushover interface", install_name="chump>=1.6.0"
-    ),
     "pydbus": dict(
         mandatory=False, desc="the Signal interface", install_name="pydbus>=0.6.0"
     ),

requirements.txt

@@ -1,5 +1,4 @@
 argparse>=1.4.0
-chump>=1.6.0
 construct~=2.9.43
 flake8
 paho_mqtt>=1.5.0,<2

setup.cfg

@@ -55,8 +55,6 @@ YAML =
 Pushbullet =
     pushbullet.py>=0.11.0
     ws4py>=0.4.2
-Pushover =
-    chump>=1.6.0
 Signal =
     pygobject>=3.20.0
     pydbus>=0.6.0