clerie/fieldpoc
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Use extension attributes in DECT thread

Browse Source
This commit is contained in:
clerie 2023-06-18 15:27:06 +02:00
parent 6c391dc9bb
commit e51cf2fc48
1 changed files with 12 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
fieldpoc/dect.py
View File

@ -63,7 +63,7 @@ class Dect:
# accessible by extension number # accessible by extension number
extensions_by_num = {e.num: e for e in extensions} extensions_by_num = {e.num: e for e in extensions}
# accessible by device IPEI, for devices with static IPEI in configuration # accessible by device IPEI, for devices with static IPEI in configuration
extensions_by_ipei = {e._c['dect_ipei']: e for _, e in extensions_by_num.items() if e._c.get('dect_ipei')} extensions_by_ipei = {e.dect_ipei: e for _, e in extensions_by_num.items() if e.dect_ipei is not None}
# signal if new extensions got created # signal if new extensions got created
created_tmp_ext = False created_tmp_ext = False
@ -92,13 +92,12 @@ class Dect:
continue continue
# update user in OMM if name of extension changed # update user in OMM if name of extension changed
elif e._c['name'] != user.name: elif e.name != user.name:
self.c.set_user_name(user.uid, e._c['name']) self.c.set_user_name(user.uid, e.name)
if e.dect_ipei is not None and user.relType != mitel_ommclient2.types.PPRelTypeType("Unbound"):
if e._c.get('dect_ipei') and user.relType != mitel_ommclient2.types.PPRelTypeType("Unbound"):
d = self.c.get_device(user.ppn) d = self.c.get_device(user.ppn)
if d.ipei != e._c['dect_ipei']: if d.ipei != e.dect_ipei:
logger.debug(f"Detaching {user} {d}") logger.debug(f"Detaching {user} {d}")
self.c.detach_user_device(user.uid, user.ppn) self.c.detach_user_device(user.uid, user.ppn)
@ -132,13 +131,13 @@ class Dect:
self.fp.temp_extensions.pop(ui.num) self.fp.temp_extensions.pop(ui.num)
self.c.set_user_num(d.uid, e.num) self.c.set_user_num(d.uid, e.num)
self.c.set_user_sipauth(d.uid, e.num, self.get_sip_password_for_number(e.num)) self.c.set_user_sipauth(d.uid, e.num, self.get_sip_password_for_number(e.num))
self.c.set_user_name(user.uid, e._c['name']) self.c.set_user_name(user.uid, e.name)
# create a new user on OMM if none exist and bind it to not bound device on OMM # create a new user on OMM if none exist and bind it to not bound device on OMM
else: else:
logger.debug(f'Creating and binding user for {d}') logger.debug(f'Creating and binding user for {d}')
user = self.create_and_bind_user(d, e.num) user = self.create_and_bind_user(d, e.num)
self.c.set_user_name(user.uid, e._c['name']) self.c.set_user_name(user.uid, e.name)
# assign any unbound device without static assigned IPEI a temporary extension # assign any unbound device without static assigned IPEI a temporary extension
elif d.relType == mitel_ommclient2.types.PPRelTypeType("Unbound"): elif d.relType == mitel_ommclient2.types.PPRelTypeType("Unbound"):
@ -156,14 +155,17 @@ class Dect:
# update rounting when new extensions got created # update rounting when new extensions got created
if created_tmp_ext: if created_tmp_ext:
logger.info("notify for routing update")
self.fp.queues['routing'].put({"type": "sync"}) self.fp.queues['routing'].put({"type": "sync"})
logger.info("finished sync")
def claim_extension(self, current_extension, token): def claim_extension(self, current_extension, token):
new_extension = None new_extension = None
# find an extension that can be claimed with the provided token # find an extension that can be claimed with the provided token
for ext in self.fp.extensions.extensions_by_type("dect"): for ext in self.fp.extensions.extensions_by_type("dect"):
if ext._c.get('dect_claim_token') == token: if ext.dect_claim_token == token:
new_extension = ext new_extension = ext
break break
@ -178,7 +180,7 @@ class Dect:
# assign new extension to the user on OMM # assign new extension to the user on OMM
self.c.set_user_num(user.uid, new_extension.num) self.c.set_user_num(user.uid, new_extension.num)
self.c.set_user_sipauth(user.uid, e.num, self.get_sip_password_for_number(new_extension.num)) self.c.set_user_sipauth(user.uid, e.num, self.get_sip_password_for_number(new_extension.num))
self.c.set_user_name(user.uid, new_extension._c['name']) self.c.set_user_name(user.uid, new_extension.name)
def run(self): def run(self):
logger.info("starting dect thread") logger.info("starting dect thread")