asterisk: Use correct event loop to connect

[ipfire.org.git] / src / backend / asterisk.py

#!/usr/bin/python3

import asyncio
import datetime
import logging
import panoramisk
import urllib.parse

from . import accounts
from . import misc
from .decorators import *

class Asterisk(misc.Object):
	def init(self):
		self.__manager = None

		loop = asyncio.get_event_loop()

		# Connect as soon as the event loop starts
		loop.create_task(self.connect())

	@property
	def manager(self):
		if not self.__manager:
			raise RuntimeError("Asterisk is not connected")

		return self.__manager

	async def connect(self):
		"""
			Connects to Asterisk
		"""
		manager = panoramisk.Manager(
			host     = self.settings.get("asterisk-ami-host"),
			username = self.settings.get("asterisk-ami-username"),
			secret   = self.settings.get("asterisk-ami-secret"),

			on_connect=self._on_connect,
		)

		# Connect
		await manager.connect()

		return manager

	def _on_connect(self, manager):
		logging.debug("Connection to Asterisk established")

		# Close any existing connections
		if self.__manager:
			self.__manager.close()

		self.__manager = manager

	async def _fetch(self, cls, action, filter=None, data={}):
		objects = []

		# Collect arguments
		args = { "Action" : action } | data

		# Run the action and parse all messages
		for data in await self.manager.send_action(args):
			if not "Event" in data or not data.Event == cls.event:
				continue

			# Create the object and append it to the list
			o = cls(self.backend, data)

			# Filter out anything unwanted
			if filter and not o.matches(filter):
				continue

			objects.append(o)

		return objects

	async def get_sip_channels(self, filter=None):
		return await self._fetch(Channel, "CoreShowChannels", filter=filter)

	async def get_registrations(self, filter=None):
		return await self._fetch(Registration, "PJSIPShowContacts", filter=filter)

	async def get_outbound_registrations(self):
		return await self._fetch(OutboundRegistration, "PJSIPShowRegistrationsOutbound")

	async def get_queues(self):
		# Fetch all queues
		queues = { q.name : q for q in await self._fetch(Queue, "QueueSummary") }

		# Fetch all members
		for member in await self._fetch(QueueMember, "QueueStatus"):
			# Append to the matching queue
			try:
				queues[member.queue].members.append(member)
			except KeyError:
				pass

		return queues.values()

	async def get_conferences(self):
		conferences = await self._fetch(Conference, "ConfbridgeListRooms")

		# Fetch everything else
		async with asyncio.TaskGroup() as tasks:
			for c in conferences:
				tasks.create_task(c._fetch())

		return conferences


class Channel(misc.Object):
	event = "CoreShowChannel"

	def init(self, data):
		self.data = data

	def __str__(self):
		return self.connected_line

	@property
	def account_code(self):
		return self.data.AccountCode

	@property
	def connected_line(self):
		return self.data.ConnectedLineName or self.data.ConnectedLineNum

	def matches(self, filter):
		return filter in (
			self.data.CallerIDNum,
		)

	@property
	def duration(self):
		h, m, s = self.data.Duration.split(":")

		try:
			h, m, s = int(h), int(m), int(s)
		except TypeError:
			return 0

		return datetime.timedelta(hours=h, minutes=m, seconds=s)

	def is_connected(self):
		return self.data.ChannelStateDesc == "Up"

	def is_ringing(self):
		return self.data.ChannelStateDesc == "Ringing"


class Registration(misc.Object):
	event = "ContactList"

	def init(self, data):
		self.data = data

	def __lt__(self, other):
		if isinstance(other, self.__class__):
			if isinstance(self.user, accounts.Account):
				if isinstance(other.user, accounts.Account):
					return self.user < other.user
				else:
					return self.user.name < other.user
			else:
				if isinstance(other.user, accounts.Account):
					return self.user < other.user.name
				else:
					return self.user < other.user

		return NotImplemented

	def __str__(self):
		return self.user_agent

	def matches(self, filter):
		return self.data.Endpoint == filter

	@lazy_property
	def uri(self):
		return urllib.parse.urlparse(self.data.Uri)

	@lazy_property
	def uri_params(self):
		params = {}

		for param in self.uri.params.split(";"):
			key, _, value = param.partition("=")

			params[key] = value

		return params

	@property
	def transport(self):
		return self.uri_params.get("transport")

	@lazy_property
	def user(self):
		return self.backend.accounts.get_by_sip_id(self.data.Endpoint) or self.data.Endpoint

	@property
	def address(self):
		# Remove the user
		user, _, address = self.uri.path.partition("@")

		# Remove the port
		address, _, port = address.rpartition(":")

		return address

	@property
	def user_agent(self):
		return self.data.UserAgent.replace("_", " ")

	@property
	def roundtrip(self):
		try:
			return int(self.data.RoundtripUsec) / 1000
		except ValueError:
			pass


class OutboundRegistration(misc.Object):
	event = "OutboundRegistrationDetail"

	def init(self, data):
		self.data = data

	def __lt__(self, other):
		if isinstance(other, self.__class__):
			return self.server < other.server or self.username < other.username

		return NotImplemented

	@lazy_property
	def uri(self):
		return urllib.parse.urlparse(self.data.ClientUri)

	@property
	def server(self):
		username, _, server = self.uri.path.partition("@")

		return server

	@property
	def username(self):
		username, _, server = self.uri.path.partition("@")

		return username

	@property
	def status(self):
		return self.data.Status


class Queue(misc.Object):
	event = "QueueSummary"

	def init(self, data):
		self.data = data

		self.members = []

	def __str__(self):
		return self.name

	@property
	def name(self):
		return self.data.Queue

	def is_available(self):
		return self.data.Available == "1"

	@property
	def callers(self):
		return int(self.data.Callers)


class QueueMember(misc.Object):
	event = "QueueMember"

	def init(self, data):
		self.data = data

	def __str__(self):
		return self.name

	@property
	def name(self):
		return self.data.Name

	@property
	def queue(self):
		return self.data.Queue

	@property
	def calls_taken(self):
		return int(self.data.CallsTaken)

	def is_in_call(self):
		return self.data.InCall == "1"

	@property
	def last_call_at(self):
		return datetime.datetime.fromtimestamp(int(self.data.LastCall))

	@property
	def logged_in_at(self):
		return datetime.datetime.fromtimestamp(int(self.data.LoginTime))

	# XXX status?


class Conference(misc.Object):
	event = "ConfbridgeListRooms"

	def init(self, data):
		self.data = data

	def __str__(self):
		return self.name

	@property
	def name(self):
		return self.data.Conference

	async def _fetch(self):
		# Fetch all members
		self.members = await self.backend.asterisk._fetch(
			ConferenceMember, "ConfbridgeList", data={ "Conference" : self.name, })


class ConferenceMember(misc.Object):
	event = "ConfbridgeList"

	def init(self, data):
		self.data = data

	def __str__(self):
		return self.name

	def __lt__(self, other):
		if isinstance(other, self.__class__):
			return not self.duration < other.duration

		return NotImplemented

	@property
	def name(self):
		return "%s <%s>" % (self.data.CallerIDName, self.data.CallerIDNum)

	def is_admin(self):
		return self.data.Admin == "Yes"

	def is_muted(self):
		return self.data.Muted == "Yes"

	@property
	def duration(self):
		return datetime.timedelta(seconds=int(self.data.AnsweredTime))