Pjsua   B
last analyzed

Complexity

Total Complexity 50

Size/Duplication

Total Lines 258
Duplicated Lines 0 %

Importance

Changes 2
Bugs 0 Features 0
Metric Value
dl 0
loc 258
rs 8.6206
c 2
b 0
f 0
wmc 50

16 Methods

Rating   Name   Duplication   Size   Complexity  
B __init__() 0 25 1
B start() 0 45 2
C call() 0 43 7
B sound_devices() 0 12 5
A lib() 0 2 1
A name() 0 2 1
A hangup() 0 6 2
A current_call_dump() 0 13 3
A thread_register() 0 1 1
B destroy() 0 19 5
A stop() 0 10 3
A sound_codecs() 0 11 3
A player() 0 2 1
C is_admin_number() 0 23 7
A recorder() 0 2 1
C self_check() 0 25 7

How to fix   Complexity   

Complex Class

Complex classes like Pjsua often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

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 by Thomas
The argument kwargs seems to be unused.
Loading history...
Unused Code introduced by Thomas
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 by Thomas
The argument kwargs seems to be unused.
Loading history...
Unused Code introduced by Thomas
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 by Thomas Meissner
Undefined variable 'doorpi'
Loading history...
111
            media_cfg   = doorpi.sipphone.pjsua_lib.Config.create_MediaConfig(),
0 ignored issues
show
Comprehensibility Best Practice introduced by Thomas Meissner
Undefined variable 'doorpi'
Loading history...
112
            log_cfg     = doorpi.sipphone.pjsua_lib.Config.create_LogConfig()
0 ignored issues
show
Comprehensibility Best Practice introduced by Thomas Meissner
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 by Thomas Meissner
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 by Thomas Meissner
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 by Thomas Meissner
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 by Thomas Meissner
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, *args, **kwargs):
0 ignored issues
show
Unused Code introduced by Tobias Ramforth
The argument kwargs seems to be unused.
Loading history...
Unused Code introduced by Tobias Ramforth
The argument args seems to be unused.
Loading history...
181
        self.lib.thread_register('pjsip_handle_events')
182
183
        self.lib.handle_events(self.call_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 by Thomas Meissner
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
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 by Thomas Meissner
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