-
Notifications
You must be signed in to change notification settings - Fork 0
/
raftar_tx.py
98 lines (78 loc) · 3.53 KB
/
raftar_tx.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
# RAFTAR is the Ropey As Fuck TAlkback Resource.
# Copyright (c) 2017 Callum McLean
# Released under the MIT Licence - see LICENSE.md for details.
import logging
import signal
import time
import linphone
class RAFTaRTX:
def __init__(self, user, passwd, codecs, snd_dev_pb='', snd_dev_cap=''):
self.running = True
callbacks = {
'call_state_changed': self.call_state_changed,
'registration_state_changed': self.registration_state_changed
}
self.codecs = codecs
logging.basicConfig(level=logging.INFO)
signal.signal(signal.SIGINT, self.on_sigint)
linphone.set_log_handler(self.log_handler)
self.core = linphone.Core.new(callbacks, None, None)
self.core.max_calls = 1
self.core.echo_cancellation_enabled = False
self.core.video_capture_enabled = False
self.core.video_display_enabled = False
self.core.stun_server = 'stun.linphone.org'
self.core.firewall_policy = linphone.FirewallPolicy.PolicyUseIce
self.core.mic_gain_db = 0.0
self.core.playback_gain_db = 0.0
if snd_dev_pb:
self.core.playback_device = snd_dev_pb
if snd_dev_cap:
self.core.capture_device = snd_dev_cap
#for codec in self.core.audio_codecs:
# if codec.mime_type.upper() in self.codecs:
# self.core.enable_payload_type(codec, True)
# logging.info("Enabled codec: {0}".format(codec.mime_type))
# else:
# self.core.enable_payload_type(codec, False)
# logging.info("Disabled codec: {0}".format(codec.mime_type))
self.configure_sip_account(user, passwd)
def on_sigint(self, signal, frame):
self.core.terminate_all_calls()
self.running = False
def log_handler(self, level, msg):
method = getattr(logging, level)
method(msg)
def call_state_changed(self, core, call, state, message):
if state == linphone.CallState.End:
logging.info("Call ended. Reason: {reason}".format(
reason=call.reason))
elif state == linphone.CallState.Connected:
logging.info("Call connected. Using codec {0}".format(call.used_audio_codec))
elif state == linphone.CallState.Idle:
self.start_call("sip:[email protected]")
def registration_state_changed(self, core, call, state, message):
if state == linphone.RegistrationState.Ok:
logging.info("Registration OK, starting first call...")
self.start_call("sip:[email protected]")
def configure_sip_account(self, username, password):
proxy_cfg = self.core.create_proxy_config()
addr = linphone.Address.new(
"sip:{0}@sip.linphone.org".format(username))
proxy_cfg.identity_address = addr
proxy_cfg.server_addr = "sip:sip.linphone.org;transport=tls"
proxy_cfg.register_enabled = True
self.core.add_proxy_config(proxy_cfg)
auth_info = self.core.create_auth_info(
username, None, password, None, None, 'sip.linphone.org')
self.core.add_auth_info(auth_info)
def start_call(self, call_address):
logging.info("Attempting call...")
params = self.core.create_call_params(None)
params.audio_bandwidth_limit = 128
address = linphone.Address.new(call_address)
self.call = self.core.invite_address_with_params(address, params)
def run(self):
while self.running:
self.core.iterate()
time.sleep(0.03)