asterisk-misc/ring_mon/ring.py

#!/usr/bin/env python3

import json
import logging
import tempfile
from argparse import ArgumentParser
from pathlib import Path

from panoramisk.actions import Action

import asterisk_misc_common

log = logging.getLogger("ring")

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.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):
    print("Shutting down...")


def main():
    global STATEFILE
    p = ArgumentParser()
    p.add_argument("--state-file", type=Path, default=Path("./ring-state.json"))

    asterisk_misc_common.add_arguments(p)
    args = p.parse_args()
    asterisk_misc_common.do_common_options(args)
    STATEFILE = args.state_file

    manager = asterisk_misc_common.get_manager(args)
    asterisk_misc_common.run_manager(
        manager, args, on_startup=init, on_shutdown=on_shutdown
    )


if __name__ == "__main__":
    main()