123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130 |
- #!/usr/bin/env python3
- import json
- import logging
- import tempfile
- from argparse import ArgumentParser
- from pathlib import Path
- from subprocess import Popen
- import asterisk_misc_common
- from panoramisk.actions import Action
- log = logging.getLogger("ring")
- ARGS = 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=ARGS.state_file.parent.absolute(),
- prefix="statefile-",
- suffix=".json",
- delete=False,
- )
- json.dump(state, tmp)
- tmp.close()
- Path(tmp.name).rename(ARGS.state_file)
- self.log.info("renamed")
- if ARGS.on_save is not None:
- Popen([str(ARGS.on_save.resolve())])
- 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 ARGS
- p = ArgumentParser()
- p.add_argument("--state-file", type=Path, default=Path("./ring-state.json"))
- p.add_argument("--on-save", type=Path, default=None)
- asterisk_misc_common.add_arguments(p)
- ARGS = p.parse_args()
- asterisk_misc_common.do_common_options(ARGS)
- 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()
|