FAELIX
/
sipcentric
2
0
Fork 0
Code Issues Releases Wiki Activity

formatting changes (blackening)

Browse Source
master
Marek Isalski 2 years ago
parent bf13265261
commit 016df8873b
3 changed files with 240 additions and 286 deletions
Show Stats Download Patch File Download Diff File

2
LICENSE
Unescape Escape View File

@ -1,7 +1,7 @@
The MIT License (MIT)
Copyright (c) 2015 Sipcentric Ltd.
Copyright (c) 2022 Faelix Limited
Copyright (c) 2022 Faelix Limited.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

338
src/sipcentric/__init__.py
Unescape Escape View File

@ -1,4 +1,4 @@
#!/usr/bin/env python3
0 #!/usr/bin/env python3
# -*- coding: utf-8 -*-
# sipcentric/__init__.py
@ -20,242 +20,208 @@ logger = logging.getLogger(__name__)
class Sipcentric(object):
def __init__(self, username, password, base='https://pbx.sipcentric.com/api/v1', customer='me'):
self.username = username # Account username
self.password = password # Account password
self.base = base # Base API URL (default: https://pbx.sipcentric.com/api/v1)
self.customer = customer # Customer (default: me)
# Resources
self.account = Account(self)
self.callBundles = CallBundles(self)
self.recordings = Recordings(self)
self.phoneBook = PhoneBook(self)
self.timeIntervals = TimeIntervals(self)
self.endpoints = Endpoints(self)
self.phoneNumbers = PhoneNumbers(self)
self.sms = Sms(self)
self.creditStatus = CreditStatus(self)
self.calls = Calls(self)
self.sounds = Sounds(self)
self.outgoingCallerIds = OutgoingCallerIds(self)
self.Stream = Stream(self)
def _request(self, uri, method='GET', data=None, params=None):
url = self.base + '/customers/' + self.customer + '/' + uri
auth = requests.auth.HTTPBasicAuth(self.username, self.password) # Basic auth
if method == 'GET':
if params:
r = requests.get(url, auth=auth, params=params, verify=True, timeout=3.000)
else:
r = requests.get(url, auth=auth, verify=True, timeout=3.000)
elif method == 'POST':
headers = {'content-type': 'application/json'}
if params:
r = requests.post(url, auth=auth, headers=headers, data=json.dumps(data), params=params, verify=True, timeout=3.000)
else:
r = requests.post(url, auth=auth, headers=headers, data=json.dumps(data), verify=True, timeout=3.000)
if (r.status_code == 200) or (r.status_code == 201):
try:
response = r.json()
return response
except:
return True
elif r.status_code == 401:
raise AuthenticationException('We couldn\'t authenticate you with the API. Make sure you are using the correct credentials from the \'Web Users\' section of the control panel. If you dont have an account, sign up for one at https://my.nimvelo.com/signup')
return False
else:
if r.json():
raise Exception('HTTP Error ' + str(r.status_code), r.json())
else:
raise Exception('HTTP Error ' + str(r.status_code), 'Something went wrong with the request')
return False
def __init__(
self,
username,
password,
base="https://pbx.sipcentric.com/api/v1",
customer="me",
):
self.username = username # Account username
self.password = password # Account password
self.base = base # Base API URL (default: https://pbx.sipcentric.com/api/v1)
self.customer = customer # Customer (default: me)
# Resources
self.account = Account(self)
self.callBundles = CallBundles(self)
self.recordings = Recordings(self)
self.phoneBook = PhoneBook(self)
self.timeIntervals = TimeIntervals(self)
self.endpoints = Endpoints(self)
self.phoneNumbers = PhoneNumbers(self)
self.sms = Sms(self)
self.creditStatus = CreditStatus(self)
self.calls = Calls(self)
self.sounds = Sounds(self)
self.outgoingCallerIds = OutgoingCallerIds(self)
self.stream = Stream(self)
def _request(self, uri, method="GET", data=None, params=None):
url = self.base + "/customers/" + self.customer + "/" + uri
auth = requests.auth.HTTPBasicAuth(self.username, self.password) # Basic auth
if method == "GET":
if params:
r = requests.get(
url, auth=auth, params=params, verify=True, timeout=3.000
)
else:
r = requests.get(url, auth=auth, verify=True, timeout=3.000)
elif method == "POST":
headers = {"content-type": "application/json"}
if params:
r = requests.post(
url,
auth=auth,
headers=headers,
data=json.dumps(data),
params=params,
verify=True,
timeout=3.000,
)
else:
r = requests.post(
url,
auth=auth,
headers=headers,
data=json.dumps(data),
verify=True,
timeout=3.000,
)
if (r.status_code == 200) or (r.status_code == 201):
try:
response = r.json()
return response
except:
return True
elif r.status_code == 401:
raise AuthenticationException(
"We couldn't authenticate you with the API. Make sure you are using the correct credentials from the 'Web Users' section of the control panel. If you dont have an account, sign up for one at https://my.nimvelo.com/signup"
)
return False
else:
if r.json():
raise Exception("HTTP Error " + str(r.status_code), r.json())
else:
raise Exception(
"HTTP Error " + str(r.status_code),
"Something went wrong with the request",
)
return False
class Account(object):
def __init__(self, parent):
self.parent = parent
self.uri = "" # Not needed for the base of the customer
def __init__(self, parent):
self.parent = parent
self.uri = '' # Not needed for the base of the customer
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class CallBundles(object):
def __init__(self, parent):
self.parent = parent
self.uri = "callbundles"
def __init__(self, parent):
self.parent = parent
self.uri = 'callbundles'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class Recordings(object):
def __init__(self, parent):
self.parent = parent
self.uri = "recordings"
def __init__(self, parent):
self.parent = parent
self.uri = 'recordings'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class PhoneBook(object):
def __init__(self, parent):
self.parent = parent
self.uri = "phonebook"
def __init__(self, parent):
self.parent = parent
self.uri = 'phonebook'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class TimeIntervals(object):
def __init__(self, parent):
self.parent = parent
self.uri = "timeintervals"
def __init__(self, parent):
self.parent = parent
self.uri = 'timeintervals'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class Endpoints(object):
def __init__(self, parent):
self.parent = parent
self.uri = "endpoints"
def __init__(self, parent):
self.parent = parent
self.uri = 'endpoints'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class PhoneNumbers(object):
def __init__(self, parent):
self.parent = parent
self.uri = "phonenumbers"
def __init__(self, parent):
self.parent = parent
self.uri = 'phonenumbers'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class Sms(object):
def __init__(self, parent):
self.parent = parent
self.uri = "sms"
def __init__(self, parent):
self.parent = parent
self.uri = 'sms'
def get(self):
def get(self):
return self.parent._request(self.uri)
return self.parent._request(self.uri)
def post(self, to=None, _from=None, body=None):
data = {
'type': 'smsmessage',
'to': to,
'from': _from,
'body': body
}
return self.parent._request(self.uri, method='POST', data=data)
def post(self, to=None, _from=None, body=None):
data = {"type": "smsmessage", "to": to, "from": _from, "body": body}
return self.parent._request(self.uri, method="POST", data=data)
class CreditStatus(object):
def __init__(self, parent):
self.parent = parent
self.uri = "creditstatus"
def __init__(self, parent):
self.parent = parent
self.uri = 'creditstatus'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class Calls(object):
def __init__(self, parent):
self.parent = parent
self.uri = "calls"
def __init__(self, parent):
self.parent = parent
self.uri = 'calls'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class Sounds(object):
def __init__(self, parent):
self.parent = parent
self.uri = "sounds"
def __init__(self, parent):
self.parent = parent
self.uri = 'sounds'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class OutgoingCallerIds(object):
def __init__(self, parent):
self.parent = parent
self.uri = "outgoingcallerids"
def __init__(self, parent):
self.parent = parent
self.uri = 'outgoingcallerids'
def get(self):
return self.parent._request(self.uri)
def get(self):
return self.parent._request(self.uri)
class AuthenticationException(Exception):
pass
pass
if __name__ == '__main__':
logging.error('Do not run directly, import module first!')
sys.exit()
if __name__ == "__main__":
logging.error("Do not run directly, import module first!")
sys.exit()

186
src/sipcentric/stream/__init__.py
Unescape Escape View File

@ -10,108 +10,96 @@ import multiprocessing
import requests
import time
import logging
import simplejson as json
logger = logging.getLogger(__name__)
class Stream(object):
'''Allows you to connect to the Nimvelo (Sipcentric) streaming API
and register callbacks to your own functions.
'''
def __init__(self, parent):
self.parent = parent
self.process = multiprocessing.Process(target=self.__run)
self.username = self.parent.username # Account username
self.password = self.parent.password # Account password
self.base = self.parent.base + '/stream' # Base streaming URL (default: https://pbx.sipcentric.com/api/v1/stream)
self.heartbeat = None
self.eventsCallback = None
self.incomingcallCallback = None
self.smsreceivedCallback = None
def __proccess(self, event):
event = json.loads(event)
logger.info('Processing event')
logger.debug(event)
if event['event'] == 'heartbeat':
self.heartbeat = time.time()
return True
elif event['event'] == 'incomingcall':
if self.incomingcallCallback:
self.incomingcallCallback(event['values'])
return True
elif event['event'] == 'smsreceived':
if self.smsreceivedCallback:
self.smsreceivedCallback(event['values'])
return True
if self.eventsCallback:
self.eventsCallback(event)
return True
def __run(self):
stream = '' # Used as a buffer for the stream data
data = False # Data is not JSON until we detect it
level = 0 # JSON object depth
r = requests.get(self.base, verify=True, auth=(self.username, self.password), stream=True)
for i in r.iter_content():
if i == '{':
stream += i
level += 1
data = True
elif i == '}':
stream += i
data = False
level -= 1
if level <= 0:
self.__proccess(stream)
stream = ''
elif data is True:
stream += i
def register(self, type, callback):
# Register a function to a callback in the class
if type == 'incomingcall':
self.incomingcallCallback = callback
elif type == 'smsreceived':
self.smsreceivedCallback = callback
elif type == 'events':
self.eventsCallback = callback
logger.info('Callback registered')
def connect(self):
# Start multiprocessing thread
self.process.start()
logger.info('Connected')
def disconnect(self):
# Terminate multiprocessing thread
self.process.terminate()
logger.info('Disconnected')
"""Allows you to connect to the Nimvelo (Sipcentric) streaming API
and register callbacks to your own functions.
"""
def __init__(self, parent):
self.parent = parent
self.process = multiprocessing.Process(target=self.__run)
self.username = self.parent.username # Account username
self.password = self.parent.password # Account password
self.base = (
self.parent.base + "/stream"
) # Base streaming URL (default: https://pbx.sipcentric.com/api/v1/stream)
self.heartbeat = None
self.eventsCallback = None
self.incomingcallCallback = None
self.smsreceivedCallback = None
def __proccess(self, event):
event = json.loads(event)
logger.info("Processing event")
logger.debug(event)
if event["event"] == "heartbeat":
self.heartbeat = time.time()
return True
elif event["event"] == "incomingcall":
if self.incomingcallCallback:
self.incomingcallCallback(event["values"])
return True
elif event["event"] == "smsreceived":
if self.smsreceivedCallback:
self.smsreceivedCallback(event["values"])
return True
if self.eventsCallback:
self.eventsCallback(event)
return True
def __run(self):
stream = "" # Used as a buffer for the stream data
data = False # Data is not JSON until we detect it
level = 0 # JSON object depth
r = requests.get(
self.base, verify=True, auth=(self.username, self.password), stream=True
)
for i in r.iter_content():
if i == "{":
stream += i
level += 1
data = True
elif i == "}":
stream += i
data = False
level -= 1
if level <= 0:
self.__proccess(stream)
stream = ""
elif data is True:
stream += i
def register(self, type, callback):
# Register a function to a callback in the class
if type == "incomingcall":
self.incomingcallCallback = callback
elif type == "smsreceived":
self.smsreceivedCallback = callback
elif type == "events":
self.eventsCallback = callback
logger.info("Callback registered")
def connect(self):
# Start multiprocessing thread
self.process.start()
logger.info("Connected")
def disconnect(self):
# Terminate multiprocessing thread
self.process.terminate()
logger.info("Disconnected")

Write Preview
Loading…
Cancel
Save