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

correction bug sur la gestion du code pin

Browse Source
This commit is contained in:
Vincent BENOIT
2022-11-25 15:41:51 +01:00
parent 2ad3941908
commit 917cb3c704
1 changed files with 85 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
kineintercom/intercom.py
View File

@@ -216,20 +216,22 @@ def set_sim_pin(serObj=None, pin_actif=False, code_pin="", logger=None):
:param log:
logger object
:return bool:
True if OK, otherwise False
:return int:
0 if OK
1 if sim code error
2 if error
'''
if not isinstance(logger, log.Logger):
return False
return 2
if not isinstance(serObj, serial.serialposix.Serial):
logger.error("error parameter, expecting serial.serialposix.Serial, get {}".format(type(serObj)))
return False
return 2
if not isinstance(pin_actif, bool):
logger.error("error parameter, expecting bool, get {}".format(type(pin_actif)))
return False
return 2
if not isinstance(code_pin, str):
logger.error("error parameter, expecting str, get {}".format(type(code_pin)))
return False
return 2
# SIM PIN mandatory or not
ret, rsp = send_at_cmd(cmd='AT+CPIN?',
@@ -238,10 +240,10 @@ def set_sim_pin(serObj=None, pin_actif=False, code_pin="", logger=None):
logger=logger)
if ret == 2:
logger.error("Erreur avec la commande AT: {}".format('AT+CPIN?'))
return False
return 2
elif ret == 1:
logger.warning("Timeout avec la commande AT: {}".format('AT+CPIN?'))
return False
return 2
else:
for item in rsp:
if item.startswith('+CPIN:'):
@@ -250,7 +252,7 @@ def set_sim_pin(serObj=None, pin_actif=False, code_pin="", logger=None):
# Must enter SIM PIN
if not pin_actif:
logger.error("Configuration en conflit avec la réponse du module GSM")
return False
return 1
else:
# Enter the SIM PIN configured in database
ret, _ = send_at_cmd(cmd='AT+CPIN='+code_pin,
@@ -259,17 +261,17 @@ def set_sim_pin(serObj=None, pin_actif=False, code_pin="", logger=None):
logger=logger)
if ret == 2:
logger.error("Erreur avec la commande AT: {}".format('AT+CPIN=<CODE_PIN>'))
return False
return 1
elif ret == 1:
logger.warning("Timeout avec la commande AT: {}".format('AT+CPIN=<CODE_PIN>'))
return False
return 2
else:
logger.info("code PIN validé ...")
elif item.split('+CPIN: ')[1] == 'READY':
# SIM PIN already notified
logger.info('SIM déverrouillée ...')
return True
return 0
def init_gsm_com(serObj=None, config={}, logger=None):
''' Init GSM Communication
@@ -292,17 +294,19 @@ def init_gsm_com(serObj=None, config={}, logger=None):
:param log:
logger object
:return bool:
True if OK, otherwise False
:return int:
0 if OK
1 if sim code error
2 if error
'''
if not isinstance(logger, log.Logger):
return False
return 2
if not isinstance(serObj, serial.serialposix.Serial):
logger.error("error parameter, expecting serial.serialposix.Serial, get {}".format(type(serObj)))
return False
return 2
if not isinstance(config, dict):
logger.error("error parameter, expecting dict, get {}".format(type(config)))
return False
return 2
cmd_lst = [{'cmd':'ATE1', 'timeout':0.0, 'func':None},
{'cmd':'AT+CMEE=2', 'timeout':0.0, 'func':None}]
@@ -311,18 +315,26 @@ def init_gsm_com(serObj=None, config={}, logger=None):
cmd_lst.append({'cmd':'AT+CMGF=1', 'timeout':0.0, 'func':None})
cmd_lst.append({'cmd':'AT+VTD='+str(config['DTMF_DURATION']), 'timeout':0.0, 'func':None})
if not set_sim_pin(serObj=serObj, pin_actif=config['PIN_ACTIF'], code_pin=config['CODE_PIN'], logger=logger):
return False
ret = set_sim_pin(serObj=serObj,
pin_actif=config['PIN_ACTIF'],
code_pin=config['CODE_PIN'],
logger=logger)
if ret > 0:
return ret
logger.info("Initialisation des commandes GSM ...")
for item in cmd_lst:
ret, rsp = send_at_cmd(cmd=item['cmd'], timeout=item['timeout'], serObj=serObj, logger=logger)
ret, rsp = send_at_cmd(cmd=item['cmd'],
timeout=item['timeout'],
serObj=serObj,
logger=logger)
if ret == 2:
logger.error("Erreur avec la commande AT: {}".format(item['cmd']))
return False
elif ret == 1:
logger.warning("Timeout avec la commande AT: {}".format(item['cmd']))
return True
ret = 2
return ret
def update_gsm_com(serObj=None, config={}, logger=None):
''' Init GSM Communication
@@ -721,16 +733,20 @@ def process(buf="", config={}, ser=None, logger=None):
else:
logger.error("Erreur de connexion avec le correspondant")
elif buf.startswith('+CMTI: '): # receive msg
msg_num = buf.split('+CMTI: ')[1].split(',')[1]
search = re.match('^(?:\+CMTI: \"SM\",)([0-9]+)$', buf)
send_at_cmd(cmd='AT+CMGF=1', timeout=0.0, serObj=ser, logger=logger)
ret, rsp = send_at_cmd(cmd='AT+CMGR='+msg_num, timeout=0.0, serObj=ser, logger=logger)
if ret == 0:
phone_num = rsp[1].split('+CMGR: ')[1].split(',')[1][1:-1]
msg = rsp[2]
logger.info('[{}] {}'.format(phone_num, msg))
if search:
msg_num = search.groups()[0]
ret, rsp = send_at_cmd(cmd='AT+CMGR='+msg_num, timeout=0.0, serObj=ser, logger=logger)
if ret == 0:
phone_num = rsp[1].split('+CMGR: ')[1].split(',')[1][1:-1]
msg = rsp[2]
logger.info('[{}] {}'.format(phone_num, msg))
else:
logger.error("Impossible de lire le SMS")
ret, rsp = send_at_cmd(cmd='AT+CMGD='+str(msg_num), timeout=0.0, serObj=ser, logger=logger)
else:
logger.error("Impossible de lire le SMS")
ret, rsp = send_at_cmd(cmd='AT+CMGD='+str(msg_num), timeout=0.0, serObj=ser, logger=logger)
logger.error('aucune correspondance')
return True
@@ -738,6 +754,7 @@ def process(buf="", config={}, ser=None, logger=None):
# Corps principal du programme #
GSM_MODULE_STATE = False
GSM_MODULE_SIM_STATE = False
GSM_MODULE_INIT_STATE = False
FLAG_CONF_UPDATE = False
FLAG_HORAIRE_UPDATE = False
@@ -761,6 +778,7 @@ def main():
logger.addHandler(handler)
global GSM_MODULE_STATE
global GSM_MODULE_SIM_STATE
global GSM_MODULE_INIT_STATE
global FLAG_CONF_UPDATE
global FLAG_HORAIRE_UPDATE
@@ -845,20 +863,33 @@ def main():
# Initialize GSM communication
logger.info("Initialisation des commandes AT nécessaires ...")
ret = init_gsm_com(serObj=ser, config=config, logger=logger)
if not ret:
if ret == 2:
logger.error("Erreur d'initialisation de la com GSM avec le module")
ser.close()
sys.exit(1)
elif ret == 1:
logger.error("Erreur de code PIN")
GSM_MODULE_INIT_STATE = False
GSM_MODULE_STATE = False
# On force en Manuel OFF
config['OPERATION'] = 'Manuel OFF'
# mise à jour de la base de données
with open(os.path.join("/etc/kineintercom", "db.json"), 'w') as f:
json.dump(config, f)
# On eteint le module GSM en cas de mauvais code PIN
if not setup_module():
logger.error("Erreur de configuration des GPIOs de la board")
ser.close()
sys.exit(1)
else:
GSM_MODULE_INIT_STATE = True
# Retreive GSM infos
logger.info('Récupération des infos du module ...')
ret = info_gsm_com(serObj=ser, config=config, logger=logger)
if not ret:
logger.error("Erreur d'initialisation de la com GSM avec le module")
ser.close()
sys.exit(1)
# Retreive GSM infos
logger.info('Récupération des infos du module ...')
ret = info_gsm_com(serObj=ser, config=config, logger=logger)
if not ret:
logger.error("Erreur d'initialisation de la com GSM avec le module")
ser.close()
sys.exit(1)
else:
# Si le module GSM doit être éteint et qu'il est allumé, on l'eteint
if GSM_MODULE_STATE:
@@ -870,6 +901,7 @@ def main():
else:
GSM_MODULE_STATE = False
GSM_MODULE_INIT_STATE = False
GSM_MODULE_SIM_STATE = False
server_addr = "/tmp/uds_socket"
# Make sure the socket does not already exist
@@ -931,9 +963,23 @@ def main():
time.sleep(10)
# Initialize GSM serial communication
ret = init_gsm_com(serObj=ser, config=config, logger=logger)
if not ret:
if ret == 2:
logger.error("Erreur d'initialisation de la com GSM avec le module")
break
elif ret == 1:
logger.error("Erreur de code PIN")
GSM_MODULE_INIT_STATE = False
GSM_MODULE_STATE = False
# On force en Manuel OFF
config['OPERATION'] = 'Manuel OFF'
# mise à jour de la base de données
with open(os.path.join("/etc/kineintercom", "db.json"), 'w') as f:
json.dump(config, f)
# On eteint le module GSM en cas de mauvais code PIN
if not setup_module():
logger.error("Erreur de configuration des GPIOs de la board")
ser.close()
sys.exit(1)
else:
GSM_MODULE_INIT_STATE = True
logger.info('Récupération des infos du module ...')