people: Drop conferences page

[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 *

loop = asyncio.get_event_loop()

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

		# 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 ping(self):
		manager = await self.manager()

		result = manager.ping()
		print(result)

	async def get_sip_channels(self, filter=None):
		channels = []

		for data in await self.manager.send_action({"Action" : "CoreShowChannels"}):
			# Skip header and trailer
			if data.eventlist:
				continue

			# Parse channel
			channel = Channel(self.backend, data)

			# Apply filter
			if filter and not channel.matches(filter):
				continue

			channels.append(channel)

		return channels

	async def get_registrations(self, filter=None):
		registrations = []

		for data in await self.manager.send_action({"Action" : "PJSIPShowContacts"}):
			# Skip header and trailer
			if data.eventlist:
				continue

			# Parse registration
			registration = Registration(self.backend, data)

			# Apply filter
			if filter and not registration.matches(filter):
				continue

			registrations.append(registration)

		return registrations

	async def get_outbound_registrations(self):
		registrations = []

		for data in await self.manager.send_action({"Action" : "PJSIPShowRegistrationsOutbound"}):
			if not data.Event == "OutboundRegistrationDetail":
				continue

			registration = OutboundRegistration(self.backend, data)
			registrations.append(registration)

		return registrations

	async def get_queues(self):
		queues = {}

		# Fetch all queues
		for data in await self.manager.send_action({"Action" : "QueueSummary"}):
			if not data.Event == "QueueSummary":
				continue

			queue = Queue(self.backend, data)
			queues[queue.name] = queue

		# Fetch all members
		for data in await self.manager.send_action({"Action" : "QueueStatus"}):
			print(data)

			if not data.Event == "QueueMember":
				continue

			member = QueueMember(self.backend, data)

			# Append to the matching queue
			try:
				queues[member.queue].members.append(member)
			except KeyError:
				pass

		return queues.values()


class Channel(misc.Object):
	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):
	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):
		return int(self.data.RoundtripUsec) / 1000


class OutboundRegistration(misc.Object):
	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):
	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):
	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?