src/backend/asterisk.py

   1 #!/usr/bin/python3
   2
   3 import asyncio
   4 import datetime
   5 import logging
   6 import panoramisk
   7
   8 from . import misc
   9 from .decorators import *
  10
  11 loop = asyncio.get_event_loop()
  12
  13 class Asterisk(misc.Object):
  14         def init(self):
  15                 self.__manager = None
  16
  17                 # Connect as soon as the event loop starts
  18                 loop.create_task(self.connect())
  19
  20         @property
  21         def manager(self):
  22                 if not self.__manager:
  23                         raise RuntimeError("Asterisk is not connected")
  24
  25                 return self.__manager
  26
  27         async def connect(self):
  28                 """
  29                         Connects to Asterisk
  30                 """
  31                 manager = panoramisk.Manager(
  32                         host     = self.settings.get("asterisk-ami-host"),
  33                         username = self.settings.get("asterisk-ami-username"),
  34                         secret   = self.settings.get("asterisk-ami-secret"),
  35
  36                         on_connect=self._on_connect,
  37                 )
  38
  39                 # Connect
  40                 await manager.connect()
  41
  42                 return manager
  43
  44         def _on_connect(self, manager):
  45                 logging.debug("Connection to Asterisk established")
  46
  47                 # Close any existing connections
  48                 if self.__manager:
  49                         self.__manager.close()
  50
  51                 self.__manager = manager
  52
  53         async def ping(self):
  54                 manager = await self.manager()
  55
  56                 result = manager.ping()
  57                 print(result)
  58
  59         async def get_sip_channels(self, filter=None):
  60                 channels = []
  61
  62                 for data in await self.manager.send_action({"Action" : "CoreShowChannels"}):
  63                         # Skip header and trailer
  64                         if data.eventlist:
  65                                 continue
  66
  67                         # Parse channel
  68                         channel = Channel(self.backend, data)
  69
  70                         # Apply filter
  71                         if filter and not channel.matches(filter):
  72                                 continue
  73
  74                         channels.append(channel)
  75
  76                 return channels
  77
  78
  79 class Channel(misc.Object):
  80         def init(self, data):
  81                 self.data = data
  82
  83         def __str__(self):
  84                 return self.connected_line
  85
  86         @property
  87         def account_code(self):
  88                 return self.data.AccountCode
  89
  90         @property
  91         def connected_line(self):
  92                 return self.data.ConnectedLineName or self.data.ConnectedLineNum
  93
  94         def matches(self, filter):
  95                 return filter in (
  96                         self.data.CallerIDNum,
  97                 )
  98
  99         @property
 100         def duration(self):
 101                 h, m, s = self.data.Duration.split(":")
 102
 103                 try:
 104                         h, m, s = int(h), int(m), int(s)
 105                 except TypeError:
 106                         return 0
 107
 108                 return datetime.timedelta(hours=h, minutes=m, seconds=s)
 109
 110         def is_connected(self):
 111                 return self.data.ChannelStateDesc == "Up"
 112
 113         def is_ringing(self):
 114                 return self.data.ChannelStateDesc == "Ringing"