vincent/KineIntercom
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

update des fonctions

Browse Source
This commit is contained in:
BENOIT Vincent
2022-11-09 20:47:48 +01:00
parent 1f82550e8e
commit e36055faf1
1 changed files with 73 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/intercom.py
+73 -20
View File
@@ -188,8 +188,6 @@ def init_gsm_com(serObj=None, config={}, logger=None):
cmd_lst = ['AT', cmd_lst = ['AT',
'ATE1', 'ATE1',
'AT+CMEE=2', 'AT+CMEE=2',
'AT+CREG?',
'AT+CSQ',
'AT+CLTS=1', 'AT+CLTS=1',
'AT+CLIP=1', 'AT+CLIP=1',
'AT+CMGF=1', 'AT+CMGF=1',
@@ -198,13 +196,14 @@ def init_gsm_com(serObj=None, config={}, logger=None):
cmd_lst.append('AT+CPIN='+config['CODE_PIN']) cmd_lst.append('AT+CPIN='+config['CODE_PIN'])
else: else:
cmd_lst.append('AT+CPIN?') cmd_lst.append('AT+CPIN?')
logger.info("Initialisation des commandes GSM ...")
for cmd in cmd_lst: for cmd in cmd_lst:
ret, _ = send_at_cmd(cmd=cmd, timeout=0.0, serObj=serObj, logger=logger) ret, _ = send_at_cmd(cmd=cmd, timeout=0.0, serObj=serObj, logger=logger)
if ret == 2: if ret == 2:
logger.error("Erreur avec la commande AT: {}".format(cmd)) logger.error("Erreur avec la commande AT: {}".format(cmd))
return False return False
elif ret == 1: elif ret == 1:
lgger.warning("Timeout avec la commande AT: {}".format(cmd)) logger.warning("Timeout avec la commande AT: {}".format(cmd))
return True return True
def update_gsm_com(serObj=None, config={}, logger=None): def update_gsm_com(serObj=None, config={}, logger=None):
@@ -240,7 +239,7 @@ def update_gsm_com(serObj=None, config={}, logger=None):
logger.error("Erreur avec la commande AT: {}".format(cmd)) logger.error("Erreur avec la commande AT: {}".format(cmd))
return False return False
elif ret == 1: elif ret == 1:
lgger.warning("Timeout avec la commande AT: {}".format(cmd)) logger.warning("Timeout avec la commande AT: {}".format(cmd))
return True return True
def info_gsm_com(serObj=None, config={}, logger=None): def info_gsm_com(serObj=None, config={}, logger=None):
@@ -274,49 +273,93 @@ def info_gsm_com(serObj=None, config={}, logger=None):
{'cmd':'AT+CSMINS?', 'obj':'control', 'item':'sim_inserted'}, {'cmd':'AT+CSMINS?', 'obj':'control', 'item':'sim_inserted'},
{'cmd':'AT+CSPN?', 'obj':'control', 'item':'service_provider'}, {'cmd':'AT+CSPN?', 'obj':'control', 'item':'service_provider'},
{'cmd':'AT+CGSN', 'obj':'manufacturer', 'item':'serial_number'}, {'cmd':'AT+CGSN', 'obj':'manufacturer', 'item':'serial_number'},
{'cmd':'AT+CCALR?', 'obj':'control', 'item':'call_ready'}] {'cmd':'AT+CCALR?', 'obj':'control', 'item':'call_ready'},
{'cmd':'AT+CREG?', 'obj':'control', 'item':''},
{'cmd':'AT+CSQ', 'obj':'control', 'item':'signal_dbm'}]
for idx, cmd in enumerate(cmd_lst): for idx, cmd in enumerate(cmd_lst):
ret, rsp = send_at_cmd(cmd=cmd['cmd'], timeout=0.0, serObj=serObj, logger=logger) ret, rsp = send_at_cmd(cmd=cmd['cmd'], timeout=0.0, serObj=serObj, logger=logger)
if ret == 2: if ret == 2:
logger.error("Erreur avec la commande AT: {}".format(cmd)) logger.error("Erreur avec la commande AT: {}".format(cmd))
return False return False
elif ret == 1: elif ret == 1:
lgger.warning("Timeout avec la commande AT: {}".format(cmd)) logger.warning("Timeout avec la commande AT: {}".format(cmd))
else: else:
logger.debug('cmd: {} - rsp: {}'.format(cmd, rsp)) logger.debug('cmd: {} - rsp: {}'.format(cmd, rsp))
if cmd['cmd'] == 'AT+CGMI': if cmd['cmd'] == 'AT+CGMI':
#cmd: AT+CGMI - rsp: ['AT+CGMI', 'SIMCOM_Ltd', 'OK'] # Request Manufacturer Identification
config['INFOS'][cmd['obj']][cmd['item']] = rsp[1] config['INFOS'][cmd['obj']][cmd['item']] = rsp[1]
elif cmd['cmd'] == 'AT+CGMM': elif cmd['cmd'] == 'AT+CGMM':
#cmd: AT+CGMM - rsp: ['AT+CGMM', 'SIMCOM_SIM868', 'OK'] # Request Model Identification
config['INFOS'][cmd['obj']][cmd['item']] = rsp[1] config['INFOS'][cmd['obj']][cmd['item']] = rsp[1]
elif cmd['cmd'] == 'AT+CGMR': elif cmd['cmd'] == 'AT+CGMR':
#cmd: AT+CGMR - rsp: ['AT+CGMR', 'Revision:1418B02SIM868M32_BT_EAT', 'OK'] # Request TA Revision Identification of Software Release
config['INFOS'][cmd['obj']][cmd['item']] = rsp[1].split(':')[1] config['INFOS'][cmd['obj']][cmd['item']] = rsp[1].split(':')[1]
elif cmd['cmd'] == 'AT+COPS?': elif cmd['cmd'] == 'AT+COPS?':
#cmd: AT+COPS? - rsp: ['AT+COPS?', '+COPS: 0,0,"SFR"', 'OK'] # Operator Selection
config['INFOS'][cmd['obj']][cmd['item']] = rsp[1].split(',')[2][1:-1] config['INFOS'][cmd['obj']][cmd['item']] = rsp[1].split(',')[2][1:-1]
elif cmd['cmd'] == 'AT+CSMINS?': elif cmd['cmd'] == 'AT+CSMINS?':
#cmd: AT+CSMINS? - rsp: ['AT+CSMINS?', '+CSMINS: 0,1', 'OK'] # SIM Inserted Status Reporting
if rsp[1].split(',')[1] == '1': if rsp[1].split(',')[1] == '1':
config['INFOS'][cmd['obj']][cmd['item']] = True config['INFOS'][cmd['obj']][cmd['item']] = True
elif rsp[1].split(',')[1] == '0': elif rsp[1].split(',')[1] == '0':
config['INFOS'][cmd['obj']][cmd['item']] = False config['INFOS'][cmd['obj']][cmd['item']] = False
elif cmd['cmd'] == 'AT+CSPN?': elif cmd['cmd'] == 'AT+CSPN?':
#cmd: AT+CSPN? - rsp: ['AT+CSPN?', '+CSPN: "La Poste Mobile",0', 'OK'] # Get Service Provider Name from SIM
config['INFOS'][cmd['obj']][cmd['item']] = rsp[1].split(': ')[1].split(',')[0][1:-1] config['INFOS'][cmd['obj']][cmd['item']] = rsp[1].split(': ')[1].split(',')[0][1:-1]
elif cmd['cmd'] == 'AT+CGSN': elif cmd['cmd'] == 'AT+CGSN':
#cmd: AT+CGSN - rsp: ['AT+CGSN', '864866057705260', 'OK'] # Request Product Serial Number Identification
config['INFOS'][cmd['obj']][cmd['item']] = rsp[1] config['INFOS'][cmd['obj']][cmd['item']] = rsp[1]
elif cmd['cmd'] == 'AT+CCALR?': elif cmd['cmd'] == 'AT+CCALR?':
#cmd: AT+CCALR? - rsp: ['AT+CCALR?', '+CCALR: 1', 'OK'] # Call Ready Query
if rsp[1].split(': ')[1] == '1': if rsp[1].split(': ')[1] == '1':
config['INFOS'][cmd['obj']][cmd['item']] = True config['INFOS'][cmd['obj']][cmd['item']] = True
elif rsp[1].split(': ')[1] == '0': elif rsp[1].split(': ')[1] == '0':
config['INFOS'][cmd['obj']][cmd['item']] = False config['INFOS'][cmd['obj']][cmd['item']] = False
elif cmd['cmd'] == 'AT+CSQ':
# Signal Quality Report
val = rsp[1].split('+CSQ: ')[1].split(',')[0]
for item in SIGNAL:
if item['val'] == int(val):
config['INFOS'][cmd['obj']][cmd['item']] = item['dbm']
config['INFOS'][cmd['obj']]['signal_qos'] = item['condition']
break
with open(os.path.join("/etc/kineintercom", "db.json"), 'w') as f:
json.dump(config, f)
return True return True
SIGNAL = [{"val":2, "dbm":"-109", 'condition':'Marginal'},
{"val":3, "dbm":"-107", 'condition':'Marginal'},
{"val":4, "dbm":"-105", 'condition':'Marginal'},
{"val":5, "dbm":"-103", 'condition':'Marginal'},
{"val":6, "dbm":"-101", 'condition':'Marginal'},
{"val":7, "dbm":"-99", 'condition':'Marginal'},
{"val":8, "dbm":"-97", 'condition':'Marginal'},
{"val":9, "dbm":"-95", 'condition':'Marginal'},
{"val":10, "dbm":"-93", 'condition':'OK'},
{"val":11, "dbm":"-91", 'condition':'OK'},
{"val":12, "dbm":"-89", 'condition':'OK'},
{"val":13, "dbm":"-87", 'condition':'OK'},
{"val":14, "dbm":"-85", 'condition':'OK'},
{"val":15, "dbm":"-83", 'condition':'Good'},
{"val":16, "dbm":"-81", 'condition':'Good'},
{"val":17, "dbm":"-79", 'condition':'Good'},
{"val":18, "dbm":"-77", 'condition':'Good'},
{"val":19, "dbm":"-75", 'condition':'Good'},
{"val":20, "dbm":"-73", 'condition':'Excellent'},
{"val":21, "dbm":"-71", 'condition':'Excellent'},
{"val":22, "dbm":"-69", 'condition':'Excellent'},
{"val":23, "dbm":"-67", 'condition':'Excellent'},
{"val":24, "dbm":"-65", 'condition':'Excellent'},
{"val":25, "dbm":"-63", 'condition':'Excellent'},
{"val":26, "dbm":"-61", 'condition':'Excellent'},
{"val":27, "dbm":"-59", 'condition':'Excellent'},
{"val":28, "dbm":"-57", 'condition':'Excellent'},
{"val":29, "dbm":"-55", 'condition':'Excellent'},
{"val":30, "dbm":"-53", 'condition':'Excellent'}]
def verify_caller(buf="", num="", logger=None): def verify_caller(buf="", num="", logger=None):
''' Verify phone number of caller ''' Verify phone number of caller
example : +CLIP: "0607297154",129,"",0,"",0 example : +CLIP: "0607297154",129,"",0,"",0
@@ -477,7 +520,7 @@ def cron_verify_hours(op=None, config=None, logger=None):
else: else:
# Si le module GSM doit être éteint et qu'il est allumé, on l'eteint # Si le module GSM doit être éteint et qu'il est allumé, on l'eteint
if GSM_MODULE_STATE: if GSM_MODULE_STATE:
logger.debug("Fermeture du module GSM HAT ...") logger.info("Fermeture du module GSM HAT ...")
if not setup_module(): if not setup_module():
logger.error('Erreur de fermeture du module GSM') logger.error('Erreur de fermeture du module GSM')
else: else:
@@ -512,13 +555,19 @@ def process(buf="", config={}, ser=None, logger=None):
# Pick up # Pick up
ret, _ = send_at_cmd(cmd='ATA', timeout=0.0, serObj=ser, logger=logger) ret, _ = send_at_cmd(cmd='ATA', timeout=0.0, serObj=ser, logger=logger)
if ret == 0: if ret == 0:
time.sleep(0.2)
logger.info('Pick up the call ...') logger.info('Pick up the call ...')
# Configure DTMF duration
ret, _ = send_at_cmd(cmd='AT+VTD='+str(config['DTMF_DURATION']), timeout=0.0, serObj=ser, logger=logger)
if ret:
logger.error("Erreur de configuration de la durée ({}) du code DTMF".format(config['DTMF_DURATION']))
time.sleep(0.2) time.sleep(0.2)
# send DMTF tone # send DMTF tone
send_at_cmd(cmd='AT+VTS='+config['DTMF_CODE'], timeout=0.0, serObj=ser, logger=logger) send_at_cmd(cmd='AT+VTS='+config['DTMF_CODE'], timeout=0.0, serObj=ser, logger=logger)
# Waiting DTMF tone duration (1/10 sec) + 2 seconds # Waiting DTMF tone duration (1/10 sec) + 2 seconds
time.sleep(float(config['DTMF_DURATION']/10) + 2) time.sleep(float(config['DTMF_DURATION']/10) + 1)
# Hang up # Hang up
logger.info('Hang up ...')
ret, _ = send_at_cmd(cmd='ATH', timeout=0.0, serObj=ser, logger=logger) ret, _ = send_at_cmd(cmd='ATH', timeout=0.0, serObj=ser, logger=logger)
if ret > 0: if ret > 0:
logger.error('Cannot hang up ...') logger.error('Cannot hang up ...')
@@ -616,10 +665,13 @@ def main():
# Verify if the GSM module must be opened or not # Verify if the GSM module must be opened or not
if config['OPERATION'] == 'Horaires': if config['OPERATION'] == 'Horaires':
logger.info("Mode \"Horaires\" activé ...")
opened = verify_open_hours(conf=config['HORAIRES'], logger=logger) opened = verify_open_hours(conf=config['HORAIRES'], logger=logger)
elif config['OPERATION'] == 'Manuel ON': elif config['OPERATION'] == 'Manuel ON':
logger.info("Mode \"Manuel ON\" activé ...")
opened = True opened = True
elif config['OPERATION'] == 'Manuel OFF': elif config['OPERATION'] == 'Manuel OFF':
logger.info("Mode \"Manuel OFF\" activé ...")
opened = False opened = False
else: else:
logger.error("Wrong 'OPERATION' parameter in database") logger.error("Wrong 'OPERATION' parameter in database")
@@ -699,12 +751,13 @@ def main():
while True: while True:
# Drapeau de mise à jour de la configuration, par le configurateur via le serveur (socket) # Drapeau de mise à jour de la configuration, par le configurateur via le serveur (socket)
if FLAG_CONF_UPDATE: if FLAG_CONF_UPDATE:
logger.info("Configuration must be reloaded ...") logger.info("Configuration doit être rechargée ...")
# reload conf in memory # reload conf in memory
config = get_conf(logger) config = get_conf(logger)
logger.info("Mode \"{}\" activé ...".format(config['OPERATION']))
# update GSM parameters with new parameters # update GSM parameters with new parameters
if not update_gsm_com(serObj=ser, config=config, logger=logger): #if not update_gsm_com(serObj=ser, config=config, logger=logger):
logger.warning('Impossible de mettre à jour les paramètres GSM') # logger.warning('Impossible de mettre à jour les paramètres GSM')
if FLAG_HORAIRE_UPDATE: if FLAG_HORAIRE_UPDATE:
logger.info("Restart scheduler ...") logger.info("Restart scheduler ...")
# Stop job scheduler to restart it # Stop job scheduler to restart it