Pjsua.sound_devices() - Code Metrics - Inspection of "Merge pull request #161 from motom001/call_url-Act..." - motom001/DoorPi - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 0dd74a...cac84e )

by Thomas

created 2016-05-23 18:27 UTC

Pjsua.sound_devices() B

↳ Parent: Pjsua

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	5
dl	0
loc	12
rs	8.5454

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import logging
logger = logging.getLogger(__name__)
logger.debug("%s loaded", __name__)

import pjsua as pj

from time import sleep

from doorpi.sipphone.pjsua_lib.Config import *
from doorpi.sipphone.pjsua_lib.SipPhoneAccountCallBack import SipPhoneAccountCallBack
from doorpi.sipphone.pjsua_lib.SipPhoneCallCallBack import SipPhoneCallCallBack
from doorpi.sipphone.pjsua_lib.Recorder import PjsuaRecorder
from doorpi.sipphone.pjsua_lib.Player import PjsuaPlayer
from AbstractBaseClass import SipphoneAbstractBaseClass

from doorpi import DoorPi

def get(*args, **kwargs): return Pjsua()

class Pjsua(SipphoneAbstractBaseClass):

    @property
    def name(self): return 'PJSUA wrapper'

    @property
    def lib(self): return self.__Lib

    @property
    def recorder(self): return self.__recorder

    @property
    def player(self): return self.__player

    @property
    def sound_devices(self):
        try:
            all_devices = []
            for sound_device in self.lib.enum_snd_dev():
                all_devices.append({
                  'name':       sound_device,
                  'capture':    True if sound_device.input_channels > 0 else False,
                  'record':     True if sound_device.output_channels > 0 else False
                })
            return all_devices
        except: return []

    @property
    def sound_codecs(self):
        try:
            all_codecs = []
            for codec in self.lib.enum_codecs():
                all_codecs.append({
                    'name':         codec.name,
                    'channels':     codec.channel_count,
                    'bitrate':      codec.avg_bps
                })
        except: return []

    @property
    def current_call_dump(self):
        try:
            return {
                'direction':        'incoming' if self.current_call.info().role == 0 else 'outgoing',
                'remote_uri':       self.current_call.info().remote_uri,
                'total_time':       self.current_call.info().call_time,
                'level_incoming':   self.lib.conf_get_signal_level(0)[0],
                'level_outgoing':   self.lib.conf_get_signal_level(0)[1],
                'camera':           False
            }
        except:
            return {}

    def thread_register(self, name): return self.lib.thread_register(name)

    def __init__(self, *args, **kwargs):

        logger.debug("__init__")

        DoorPi().event_handler.register_event('OnSipPhoneCreate', __name__)
        DoorPi().event_handler.register_event('OnSipPhoneStart', __name__)
        DoorPi().event_handler.register_event('OnSipPhoneDestroy', __name__)

        DoorPi().event_handler.register_event('OnSipPhoneRecorderCreate', __name__)
        DoorPi().event_handler.register_event('OnSipPhoneRecorderDestroy', __name__)

        DoorPi().event_handler.register_event('BeforeSipPhoneMakeCall', __name__)
        DoorPi().event_handler.register_event('OnSipPhoneMakeCall', __name__)
        DoorPi().event_handler.register_event('AfterSipPhoneMakeCall', __name__)
        
        DoorPi().event_handler.register_event('OnSipPhoneCallTimeoutNoResponse', __name__)
        DoorPi().event_handler.register_event('OnSipPhoneCallTimeoutMaxCalltime', __name__)

        self.__Lib = None
        self.__account = None
        self.current_callcallback = None
        self.current_account_callback = None
        self.__recorder = None
        self.__player = None

        self.call_timeout = 30

    def start(self):
        DoorPi().event_handler('OnSipPhoneCreate', __name__)
        self.__Lib = pj.Lib.instance()
        if self.__Lib is None: self.__Lib = pj.Lib()

        logger.debug("init Lib")
        self.__Lib.init(
            ua_cfg      = doorpi.sipphone.pjsua_lib.Config.create_UAConfig(),

            media_cfg   = doorpi.sipphone.pjsua_lib.Config.create_MediaConfig(),

            log_cfg     = doorpi.sipphone.pjsua_lib.Config.create_LogConfig()

        )

        logger.debug("init transport")
        transport = self.__Lib.create_transport(
            type        = pj.TransportType.UDP,
            cfg         = doorpi.sipphone.pjsua_lib.Config.create_TransportConfig()

        )
        logger.debug("Listening on: %s",str(transport.info().host))
        logger.debug("Port: %s",str(transport.info().port))

        logger.debug("Lib.start()")
        self.lib.start(0)

        DoorPi().event_handler.register_action(
            event_name      = 'OnTimeTick',
            action_object   = 'pjsip_handle_events:50'
        )

        logger.debug("init Acc")
        self.current_account_callback = SipPhoneAccountCallBack()
        self.__account = self.__Lib.create_account(
            acc_config  = doorpi.sipphone.pjsua_lib.Config.create_AccountConfig(),

            set_default = True,
            cb          = self.current_account_callback
        )

        self.call_timeout = doorpi.sipphone.pjsua_lib.Config.call_timeout()

        self.max_call_time = doorpi.sipphone.pjsua_lib.Config.max_call_time()


        DoorPi().event_handler('OnSipPhoneStart', __name__)

        self.__recorder = PjsuaRecorder()
        self.__player = PjsuaPlayer()

        logger.debug("start successfully")

    def stop(self, timeout = -1):
        if self.current_call is None:
            logger.debug('no call? -> nothing to do to clean up')
            return True
        else:
            call_info = self.current_call.info()
            if call_info.total_time > timeout:
                logger.info('call timeout - call.info().total_time %s', call_info.total_time)
                return self.hangup()
            return True

    def destroy(self):
        logger.debug("destroy")
        DoorPi().event_handler('OnSipPhoneDestroy', __name__)

        if self.lib is not None:
            self.lib.handle_events()
            self.__Lib.destroy()
            self.lib.handle_events()

        try:
            timeout = 0
            while timeout < 5 and self.__Lib is not None:
                sleep(0.1)
                timeout += 0.1
                self.lib.handle_events()

        except:
            DoorPi().event_handler.unregister_source(__name__, True)
            return

    def self_check(self, timeout):
        self.lib.thread_register('pjsip_handle_events')

        self.lib.handle_events(timeout)

        if self.current_call is not None:
            if self.current_call.is_valid() is 0:
                del self.current_callcallback
                self.current_callcallback = None
                del self.current_call
                self.current_call = None

            try:
                if self.current_call.info().call_time == 0 \
                and self.current_call.info().total_time > self.call_timeout:
                    logger.info("call timeout - hangup current call after %s seconds", self.call_timeout)
                    self.current_call.hangup()
                    DoorPi().event_handler('OnSipPhoneCallTimeoutNoResponse', __name__)

                if self.current_call.info().call_time > self.max_call_time:
                    logger.info("max call time reached - hangup current call after %s seconds", self.max_call_time)
                    self.current_call.hangup()
                    DoorPi().event_handler('OnSipPhoneCallTimeoutMaxCalltime', __name__)
            except:
                pass

    def call(self, number):
        DoorPi().event_handler('BeforeSipPhoneMakeCall', __name__, {'number':number})
        logger.debug("call(%s)",str(number))

        self.lib.thread_register('call_theard')

        sip_server = doorpi.sipphone.pjsua_lib.Config.sipphone_server()

        sip_uri = "sip:"+str(number)+"@"+str(sip_server)

        if self.lib.verify_sip_url(sip_uri) is not 0:
            logger.warning("SIP-URI %s is not valid (Errorcode: %s)", sip_uri, self.lib.verify_sip_url(sip_uri))
            return false

        else:
            logger.debug("SIP-URI %s is valid", sip_uri)

        DoorPi().event_handler('OnSipPhoneMakeCall', __name__)
        if not self.current_call or self.current_call.is_valid() is 0:
            lck = self.lib.auto_lock()
            self.current_callcallback = SipPhoneCallCallBack()
            self.current_call = self.__account.make_call(
                sip_uri,
                self.current_callcallback
            )
            del lck

        elif self.current_call.info().remote_uri == sip_uri:
            if self.current_call.info().total_time <= 1:
                logger.debug("same call again while call is running since %s seconds? -> skip", str(self.current_call.info().total_time))
            else:
                logger.debug("press twice with call duration > 1 second? Want to hangup current call? OK...")
                #self.current_call.hangup()
                self.lib.hangup_all()
        else:
            logger.debug("new call needed? hangup old first...")
            try:
                # self.current_call.hangup()
                self.lib.hangup_all()
            except pj.Error as e:
                logger.exception("Exception: %s", str(e))
            self.call(Number)


        DoorPi().event_handler('AfterSipPhoneMakeCall', __name__)
        return self.current_call

    def is_admin_number(self, remote_uri = None):
        logger.debug("is_admin_number (%s)",remote_uri)

        if remote_uri is None:
            if self.current_call is not None:
                remote_uri = self.current_call.info().remote_uri
            else:
                logger.debug("couldn't catch current call - no parameter and no current_call from doorpi itself")
                return False

        possible_admin_numbers = DoorPi().config.get_keys('AdminNumbers')
        for admin_number in possible_admin_numbers:
            if admin_number == "*":
                logger.info("admin numbers are deactivated by using '*' as single number")
                return True
            if "sip:"+admin_number+"@" in remote_uri:
                logger.debug("%s is an adminnumber", remote_uri)
                return True
            if "sip:"+admin_number is remote_uri:
                logger.debug("%s is adminnumber %s", remote_uri, admin_number)
                return True
        logger.debug("%s is not an adminnumber", remote_uri)
        return False

    def hangup(self):
        if self.current_call:
            logger.debug("Received hangup request, cancelling current call")
            self.lib.hangup_all()
        else:
            logger.debug("Ignoring hangup request as there is no ongoing call")



1			#!/usr/bin/env python
2			# -- coding: utf-8 --
3
4			import logging
5			logger = logging.getLogger(__name__)
6			logger.debug("%s loaded", __name__)
7
8			import pjsua as pj
9
10			from time import sleep
11
12			from doorpi.sipphone.pjsua_lib.Config import *
13			from doorpi.sipphone.pjsua_lib.SipPhoneAccountCallBack import SipPhoneAccountCallBack
14			from doorpi.sipphone.pjsua_lib.SipPhoneCallCallBack import SipPhoneCallCallBack
15			from doorpi.sipphone.pjsua_lib.Recorder import PjsuaRecorder
16			from doorpi.sipphone.pjsua_lib.Player import PjsuaPlayer
17			from AbstractBaseClass import SipphoneAbstractBaseClass
18
19			from doorpi import DoorPi
20
21			def get(args, *kwargs): return Pjsua()
			0 ignored issues – show Unused Code introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report The argument `kwargs` seems to be unused. Loading history... Unused Code introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report The argument `args` seems to be unused. Loading history...
22			class Pjsua(SipphoneAbstractBaseClass):
23
24			@property
25			def name(self): return 'PJSUA wrapper'
26
27			@property
28			def lib(self): return self.__Lib
29
30			@property
31			def recorder(self): return self.__recorder
32
33			@property
34			def player(self): return self.__player
35
36			@property
37			def sound_devices(self):
38			try:
39			all_devices = []
40			for sound_device in self.lib.enum_snd_dev():
41			all_devices.append({
42			'name': sound_device,
43			'capture': True if sound_device.input_channels > 0 else False,
44			'record': True if sound_device.output_channels > 0 else False
45			})
46			return all_devices
47			except: return []
48
49			@property
50			def sound_codecs(self):
51			try:
52			all_codecs = []
53			for codec in self.lib.enum_codecs():
54			all_codecs.append({
55			'name': codec.name,
56			'channels': codec.channel_count,
57			'bitrate': codec.avg_bps
58			})
59			except: return []
60
61			@property
62			def current_call_dump(self):
63			try:
64			return {
65			'direction': 'incoming' if self.current_call.info().role == 0 else 'outgoing',
66			'remote_uri': self.current_call.info().remote_uri,
67			'total_time': self.current_call.info().call_time,
68			'level_incoming': self.lib.conf_get_signal_level(0)[0],
69			'level_outgoing': self.lib.conf_get_signal_level(0)[1],
70			'camera': False
71			}
72			except:
73			return {}
74
75			def thread_register(self, name): return self.lib.thread_register(name)
76
77			def __init__(self, args, *kwargs):
			0 ignored issues – show Unused Code introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report The argument `kwargs` seems to be unused. Loading history... Unused Code introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report The argument `args` seems to be unused. Loading history...
78			logger.debug("__init__")
79
80			DoorPi().event_handler.register_event('OnSipPhoneCreate', __name__)
81			DoorPi().event_handler.register_event('OnSipPhoneStart', __name__)
82			DoorPi().event_handler.register_event('OnSipPhoneDestroy', __name__)
83
84			DoorPi().event_handler.register_event('OnSipPhoneRecorderCreate', __name__)
85			DoorPi().event_handler.register_event('OnSipPhoneRecorderDestroy', __name__)
86
87			DoorPi().event_handler.register_event('BeforeSipPhoneMakeCall', __name__)
88			DoorPi().event_handler.register_event('OnSipPhoneMakeCall', __name__)
89			DoorPi().event_handler.register_event('AfterSipPhoneMakeCall', __name__)
90
91			DoorPi().event_handler.register_event('OnSipPhoneCallTimeoutNoResponse', __name__)
92			DoorPi().event_handler.register_event('OnSipPhoneCallTimeoutMaxCalltime', __name__)
93
94			self.__Lib = None
95			self.__account = None
96			self.current_callcallback = None
97			self.current_account_callback = None
98			self.__recorder = None
99			self.__player = None
100
101			self.call_timeout = 30
102
103			def start(self):
104			DoorPi().event_handler('OnSipPhoneCreate', __name__)
105			self.__Lib = pj.Lib.instance()
106			if self.__Lib is None: self.__Lib = pj.Lib()
107
108			logger.debug("init Lib")
109			self.__Lib.init(
110			ua_cfg = doorpi.sipphone.pjsua_lib.Config.create_UAConfig(),
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'doorpi' Loading history...
111			media_cfg = doorpi.sipphone.pjsua_lib.Config.create_MediaConfig(),
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'doorpi' Loading history...
112			log_cfg = doorpi.sipphone.pjsua_lib.Config.create_LogConfig()
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'doorpi' Loading history...
113			)
114
115			logger.debug("init transport")
116			transport = self.__Lib.create_transport(
117			type = pj.TransportType.UDP,
118			cfg = doorpi.sipphone.pjsua_lib.Config.create_TransportConfig()
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'doorpi' Loading history...
119			)
120			logger.debug("Listening on: %s",str(transport.info().host))
121			logger.debug("Port: %s",str(transport.info().port))
122
123			logger.debug("Lib.start()")
124			self.lib.start(0)
125
126			DoorPi().event_handler.register_action(
127			event_name = 'OnTimeTick',
128			action_object = 'pjsip_handle_events:50'
129			)
130
131			logger.debug("init Acc")
132			self.current_account_callback = SipPhoneAccountCallBack()
133			self.__account = self.__Lib.create_account(
134			acc_config = doorpi.sipphone.pjsua_lib.Config.create_AccountConfig(),
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'doorpi' Loading history...
135			set_default = True,
136			cb = self.current_account_callback
137			)
138
139			self.call_timeout = doorpi.sipphone.pjsua_lib.Config.call_timeout()
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'doorpi' Loading history...
140			self.max_call_time = doorpi.sipphone.pjsua_lib.Config.max_call_time()
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'doorpi' Loading history...
141
142			DoorPi().event_handler('OnSipPhoneStart', __name__)
143
144			self.__recorder = PjsuaRecorder()
145			self.__player = PjsuaPlayer()
146
147			logger.debug("start successfully")
148
149			def stop(self, timeout = -1):
150			if self.current_call is None:
151			logger.debug('no call? -> nothing to do to clean up')
152			return True
153			else:
154			call_info = self.current_call.info()
155			if call_info.total_time > timeout:
156			logger.info('call timeout - call.info().total_time %s', call_info.total_time)
157			return self.hangup()
158			return True
159
160			def destroy(self):
161			logger.debug("destroy")
162			DoorPi().event_handler('OnSipPhoneDestroy', __name__)
163
164			if self.lib is not None:
165			self.lib.handle_events()
166			self.__Lib.destroy()
167			self.lib.handle_events()
168
169			try:
170			timeout = 0
171			while timeout < 5 and self.__Lib is not None:
172			sleep(0.1)
173			timeout += 0.1
174			self.lib.handle_events()
175
176			except:
177			DoorPi().event_handler.unregister_source(__name__, True)
178			return
179
180			def self_check(self, timeout):
181			self.lib.thread_register('pjsip_handle_events')
182
183			self.lib.handle_events(timeout)
184
185			if self.current_call is not None:
186			if self.current_call.is_valid() is 0:
187			del self.current_callcallback
188			self.current_callcallback = None
189			del self.current_call
190			self.current_call = None
191
192			try:
193			if self.current_call.info().call_time == 0 \
194			and self.current_call.info().total_time > self.call_timeout:
195			logger.info("call timeout - hangup current call after %s seconds", self.call_timeout)
196			self.current_call.hangup()
197			DoorPi().event_handler('OnSipPhoneCallTimeoutNoResponse', __name__)
198
199			if self.current_call.info().call_time > self.max_call_time:
200			logger.info("max call time reached - hangup current call after %s seconds", self.max_call_time)
201			self.current_call.hangup()
202			DoorPi().event_handler('OnSipPhoneCallTimeoutMaxCalltime', __name__)
203			except:
204			pass
205
206			def call(self, number):
207			DoorPi().event_handler('BeforeSipPhoneMakeCall', __name__, {'number':number})
208			logger.debug("call(%s)",str(number))
209
210			self.lib.thread_register('call_theard')
211
212			sip_server = doorpi.sipphone.pjsua_lib.Config.sipphone_server()
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'doorpi' Loading history...
213			sip_uri = "sip:"+str(number)+"@"+str(sip_server)
214
215			if self.lib.verify_sip_url(sip_uri) is not 0:
216			logger.warning("SIP-URI %s is not valid (Errorcode: %s)", sip_uri, self.lib.verify_sip_url(sip_uri))
217			return false
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'false' Loading history...
218			else:
219			logger.debug("SIP-URI %s is valid", sip_uri)
220
221			DoorPi().event_handler('OnSipPhoneMakeCall', __name__)
222			if not self.current_call or self.current_call.is_valid() is 0:
223			lck = self.lib.auto_lock()
224			self.current_callcallback = SipPhoneCallCallBack()
225			self.current_call = self.__account.make_call(
226			sip_uri,
227			self.current_callcallback
228			)
229			del lck
230
231			elif self.current_call.info().remote_uri == sip_uri:
232			if self.current_call.info().total_time <= 1:
233			logger.debug("same call again while call is running since %s seconds? -> skip", str(self.current_call.info().total_time))
234			else:
235			logger.debug("press twice with call duration > 1 second? Want to hangup current call? OK...")
236			#self.current_call.hangup()
237			self.lib.hangup_all()
238			else:
239			logger.debug("new call needed? hangup old first...")
240			try:
241			# self.current_call.hangup()
242			self.lib.hangup_all()
243			except pj.Error as e:
244			logger.exception("Exception: %s", str(e))
245			self.call(Number)
			0 ignored issues – show Comprehensibility Best Practice introduced 2015-12-28 07:44 UTC by Report Bug Copy Issue Report Undefined variable 'Number' Loading history...
246
247			DoorPi().event_handler('AfterSipPhoneMakeCall', __name__)
248			return self.current_call
249
250			def is_admin_number(self, remote_uri = None):
251			logger.debug("is_admin_number (%s)",remote_uri)
252
253			if remote_uri is None:
254			if self.current_call is not None:
255			remote_uri = self.current_call.info().remote_uri
256			else:
257			logger.debug("couldn't catch current call - no parameter and no current_call from doorpi itself")
258			return False
259
260			possible_admin_numbers = DoorPi().config.get_keys('AdminNumbers')
261			for admin_number in possible_admin_numbers:
262			if admin_number == "*":
263			logger.info("admin numbers are deactivated by using '*' as single number")
264			return True
265			if "sip:"+admin_number+"@" in remote_uri:
266			logger.debug("%s is an adminnumber", remote_uri)
267			return True
268			if "sip:"+admin_number is remote_uri:
269			logger.debug("%s is adminnumber %s", remote_uri, admin_number)
270			return True
271			logger.debug("%s is not an adminnumber", remote_uri)
272			return False
273
274			def hangup(self):
275			if self.current_call:
276			logger.debug("Received hangup request, cancelling current call")
277			self.lib.hangup_all()
278			else:
279			logger.debug("Ignoring hangup request as there is no ongoing call")
280
281

motom001 / DoorPi

Push — master ( 0dd74a...cac84e )

Pjsua.sound_devices() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like