ring.py: provide info about phone status

is it ringing? is it on call?

ring_mon/ring.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+
+import asyncio
+import json
+import logging
+import tempfile
+from argparse import ArgumentParser
+from pathlib import Path
+
+from panoramisk import Manager
+from panoramisk.actions import Action
+
+log = logging.getLogger("ring")
+
+LOOP = asyncio.get_event_loop()
+STATEFILE = None
+
+
+async def run_app(Cls, manager):
+    instance = Cls(manager)
+    await instance.run()
+
+
+async def update(manager):
+    manager.send_command("sip show peers")
+    pass
+
+
+async def on_tutto(manager, msg):
+    for prefix in ["EndpointList", "CoreShowChannel", "TestEvent"]:
+        if msg.event.startswith(prefix):
+            return
+    log.debug("     (%s)", msg.event)
+
+
+def public_dir(x):
+    return [attr for attr in dir(x) if not attr.startswith("_")]
+
+
+class RingManager:
+    def __init__(self, manager):
+        self.manager = manager
+        self.events = []
+        self.log = logging.getLogger(self.__class__.__name__)
+        self.tmp_channels = []
+        self.channels = []
+
+        self.endpoints = []
+        self.tmp_endpoints = []
+
+    async def run(self):
+        self.manager.register_event("EndpointList", self.on_endpoint)
+        self.manager.register_event("EndpointListComplete", self.on_endpoints)
+
+    async def on_endpoint(self, _, msg):
+        self.tmp_endpoints.append(msg)
+
+    async def on_endpoints(self, _, msg):
+        self.endpoints = self.tmp_endpoints
+        self.tmp_endpoints = []
+
+        state_map = {
+            "In use": "Up",
+            "Ringing": "Ringing",
+        }
+
+        state = {}
+        for e in self.endpoints:
+            if e.get("DeviceState") in state_map:
+                state[e.auths] = state_map[e.get("DeviceState")]
+        self.save({"devices": state, "onair": {}})
+
+    def save(self, state):
+        tmp = tempfile.NamedTemporaryFile(
+            mode="w",
+            dir=STATEFILE.parent.absolute(),
+            prefix="statefile-",
+            suffix=".json",
+            delete=False,
+        )
+        json.dump(state, tmp)
+        tmp.close()
+        Path(tmp.name).rename(STATEFILE)
+
+
+def refresh(manager, *args):
+    # manager.send_action(Action({"Action": "CoreShowChannels"}), as_list=False)
+    manager.send_action(Action({"Action": "PJSIPShowEndpoints"}), as_list=False)
+
+
+async def init(manager):
+    # manager.send_command('sip show peers')
+
+    manager.register_event("*", on_tutto)
+    for event in [
+        "FullyBooted",
+        "Hangup",
+        "Newchannel",
+        "Newexten",
+        "DialBegin",
+        "DialEnd",
+        "DeviceStateChange",
+    ]:
+        manager.register_event(event, refresh)
+
+    await run_app(RingManager, manager)
+
+
+async def on_shutdown(m):
+    pass
+
+
+def main():
+    global STATEFILE
+    p = ArgumentParser()
+    p.add_argument("--state-file", type=Path, default=Path("./ring-state.json"))
+    args = p.parse_args()
+    STATEFILE = args.state_file
+
+    logging.basicConfig(level=logging.DEBUG)
+    logging.getLogger("panoramisk.manager").setLevel(logging.WARNING)
+    manager = Manager(
+        loop=LOOP,
+        host="127.0.0.1",
+        port=5038,
+        ssl=False,
+        encoding="utf8",
+        username="ring",
+        secret="campanellino",
+        ping_delay=10,  # Delay after start
+        ping_interval=10,  # Periodically ping AMI (dead or alive)
+        reconnect_timeout=2,  # Timeout reconnect if connection lost
+    )
+    manager.connect(run_forever=True, on_startup=init, on_shutdown=on_shutdown)
+
+
+if __name__ == "__main__":
+    main()