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

copie d'une base de données vierge et conforme si elle n'existe pas

Browse Source
This commit is contained in:
Vincent BENOIT
2022-11-15 11:06:04 +01:00
parent 5de4445de7
commit 5ea494e645
1 changed files with 102 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/intercom.py
+102 -71
View File
@@ -3,7 +3,7 @@
# @author: benoit.vince84@free.fr # @author: benoit.vince84@free.fr
# @date: Septembre 2022 # @date: Septembre 2022
# @brief: Programme Intercom à partir du module GSM du GNSS_HAT # @brief: Programme Intercom à partir du module GSM
################################################################### ###################################################################
# Importation de modules externes # # Importation de modules externes #
@@ -16,8 +16,9 @@ import RPi.GPIO as GPIO
from datetime import datetime, date from datetime import datetime, date
import calendar import calendar
import json import json
from jsonschema import validate import jsonschema
import socket import socket
import shutil
from signal import signal, SIGINT from signal import signal, SIGINT
from threading import Thread from threading import Thread
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.schedulers.background import BackgroundScheduler
@@ -26,54 +27,65 @@ from apscheduler.schedulers.background import BackgroundScheduler
# Class et Methods # # Class et Methods #
def validate_json(json_data={}, logger=None): def validate_json(json_data={}, logger=None):
''' Validate json by schema ''' Validate json database by schema
:param json_data: :param json_data:
json data loaded json data loaded
:param logger: :param logger:
werkzeug context logger object
:return int: :return bool:
0 => If validates with 2.0 True if OK, otherwise False
1 => otherwise
''' '''
if not isinstance(logger, log.Logger):
return False
if not isinstance(json_data, dict): if not isinstance(json_data, dict):
logger.error("error parameter, expecting dict, get {}".format(type(json_data))) logger.error("error parameter, expecting dict, get {}".format(type(json_data)))
return False return False
if not isinstance(logger, log.Logger):
logger.error("error parameter, expecting logging.Logger, get {}".format(type(logger)))
return False
ret = True ret = True
try: try:
with open(os.path.join('/home/pi', 'db.json.schema'), 'r') as f: with open(os.path.join('/home/pi', 'db.json.schema'), 'r') as f:
schema=json.load(f) schema=json.load(f)
try:
validate(instance=json_data, schema=schema)
except jsonschema.exceptions.ValidationError as e:
logger.error("Validator error")
ret = False
except jsonschema.exceptions.SchemaError as e:
logger.error("Schema error")
ret = False
except FileNotFoundError as e: except FileNotFoundError as e:
logger.error("Impossible d'ouvrir le fichier de validation ({})".format(e)) logger.error("Impossible d'ouvrir le fichier de validation ({})".format(e))
ret = False ret = False
try:
jsonschema.validate(instance=json_data, schema=schema)
except jsonschema.exceptions.ValidationError as e:
logger.error("Erreur de validation de la base de données : {}".format(e))
ret = False
except jsonschema.exceptions.SchemaError as e:
logger.error("Erreur de schéma : {}".format(e))
ret = False
return ret return ret
def get_conf(logger=None): def get_conf(logger=None):
''' Get configuration ''' Get configuration
:param logger:
logger object
:return dict: :return dict:
Configuration dictionnary Configuration dictionnary
''' '''
if not isinstance(logger, log.Logger): if not isinstance(logger, log.Logger):
logger.error("error parameter, expecting logging.Logger, get {}".format(type(logger)))
return None return None
config = None config = None
# copy database if not exists
if not os.path.exists(os.path.join("/etc/kineintercom", "db.json")):
logger.warning("Le fichier de la base de données n'existe pas ...")
try:
shutil.copyfile(os.path.join("/usr/share/kineintercom", "database_origin.json"),
os.path.join("/etc/kineintercom", "db.json"))
except PermissionError as e:
logger.error("Erreur de permission: {}".format(e))
return None
try: try:
with open(os.path.join("/etc/kineintercom", "db.json"), 'r') as f: with open(os.path.join("/etc/kineintercom", "db.json"), 'r') as f:
try: try:
@@ -100,7 +112,7 @@ def send_at_cmd(cmd='', timeout=0.0, serObj=None, logger=None):
:param serObj: :param serObj:
serial object serial object
:param log: :param logger:
logger object logger object
:return int: :return int:
@@ -121,7 +133,7 @@ def send_at_cmd(cmd='', timeout=0.0, serObj=None, logger=None):
try: try:
if timeout > 0.0: if timeout > 0.0:
serObj.timeout = timeout serObj.timeout = timeout + 0.1
else: else:
serObj.timeout = None serObj.timeout = None
serObj.write(bytes(cmd+'\r', 'utf-8')) serObj.write(bytes(cmd+'\r', 'utf-8'))
@@ -129,19 +141,12 @@ def send_at_cmd(cmd='', timeout=0.0, serObj=None, logger=None):
out = '' out = ''
outlst = [] outlst = []
while serObj.in_waiting > 0: while serObj.in_waiting > 0:
#c = serObj.read(1).decode('utf-8', 'replace')
#if c == '\r' or c == '\n':
# if out != '':
# outlst.append(out)
# out = ''
#else:
# out += c
# remove \r and \n chars from out string # remove \r and \n chars from out string
out += serObj.read_until().decode('utf-8', 'replace').replace('\r','').replace('\n','') out += serObj.read_until().decode('utf-8', 'replace').replace('\r','').replace('\n','')
if out != '': if out != '':
outlst.append(out) outlst.append(out)
out = '' out = ''
time.sleep(timeout)
logger.debug("Reponse: {}".format(outlst)) logger.debug("Reponse: {}".format(outlst))
if 'OK' in outlst: if 'OK' in outlst:
return 0, outlst return 0, outlst
@@ -162,16 +167,23 @@ def send_at_cmd(cmd='', timeout=0.0, serObj=None, logger=None):
def set_sim_pin(serObj=None, pin_actif=False, code_pin="", logger=None): def set_sim_pin(serObj=None, pin_actif=False, code_pin="", logger=None):
''' Set SIM PIN if necessary ''' Set SIM PIN if necessary
AT+CPIN=<CODE PIN> : Enter PIN (response READY: MT is not pending for any password)
:param serObj: :param serObj:
serial object serial object
:param config: :param pin_actif:
dictionary config object from JSON database attribute from dictionary config object
:param code_pin:
attribute from dictionary config object
:param log: :param log:
logger object logger object
:return bool: :return bool:
True if OK, otherwise False
''' '''
if not isinstance(logger, log.Logger): if not isinstance(logger, log.Logger):
return False return False
@@ -196,28 +208,30 @@ def set_sim_pin(serObj=None, pin_actif=False, code_pin="", logger=None):
elif ret == 1: elif ret == 1:
logger.warning("Timeout avec la commande AT: {}".format('AT+CPIN?')) logger.warning("Timeout avec la commande AT: {}".format('AT+CPIN?'))
else: else:
if rsp[1].split('+CPIN: ')[1] == 'SIM PIN': for item in rsp:
logger.info('SIM verrouillée ...') if item.startswith('+CPIN:'):
# Must enter SIM PIN if item.split('+CPIN: ')[1] == 'SIM PIN':
if not pin_actif: logger.info('SIM verrouillée ...')
logger.error("Configuration en conflit avec la réponse du module GSM") # Must enter SIM PIN
return False if not pin_actif:
else: logger.error("Configuration en conflit avec la réponse du module GSM")
# Enter the SIM PIN configured in database return False
ret, rsp = send_at_cmd(cmd='AT+CPIN='+code_pin, else:
timeout=0.0, # Enter the SIM PIN configured in database
serObj=serObj, ret, _ = send_at_cmd(cmd='AT+CPIN='+code_pin,
logger=logger) timeout=2.0,
if ret == 2: serObj=serObj,
logger.error("Erreur avec la commande AT: {}".format('AT+CPIN=<CODE_PIN>')) logger=logger)
return False if ret == 2:
elif ret == 1: logger.error("Erreur avec la commande AT: {}".format('AT+CPIN=<CODE_PIN>'))
logger.warning("Timeout avec la commande AT: {}".format('AT+CPIN=<CODE_PIN>')) return False
else: elif ret == 1:
logger.info("code PIN saisi ...") logger.warning("Timeout avec la commande AT: {}".format('AT+CPIN=<CODE_PIN>'))
elif rsp[1].split('+CPIN: ')[1] == 'READY': else:
# SIM PIN already notified logger.info("code PIN validé ...")
logger.info('SIM déverrouillée ...') elif item.split('+CPIN: ')[1] == 'READY':
# SIM PIN already notified
logger.info('SIM déverrouillée ...')
return True return True
@@ -227,15 +241,11 @@ def init_gsm_com(serObj=None, config={}, logger=None):
AT : test command AT : test command
ATE1 : ATE1 :
AT+CMEE=2 : AT+CMEE=2 :
AT+CPIN? : Enter PIN (response READY: MT is not pending for any password
AT+CREG? : Network registration - return the status of result code
AT+CSQ :
AT+CLTS=1 : AT+CLTS=1 :
AT+CMGF=1 : Set the format of messages to Text mode AT+CMGF=1 : Set the format of messages to Text mode
AT+CLIP=1 : The calling line identifty (CLI) of calling party when AT+CLIP=1 : The calling line identifty (CLI) of calling party when
receiving a mobile terminated call receiving a mobile terminated call
AT+VTD=1 : Tone Duration (in 1/10 seconds) AT+VTD=1 : Tone Duration (in 1/10 seconds)
ATS0=2 : Set Number of Ring before Automatically Answering the call
:param serObj: :param serObj:
serial object serial object
@@ -247,6 +257,7 @@ def init_gsm_com(serObj=None, config={}, logger=None):
logger object logger object
:return bool: :return bool:
True if OK, otherwise False
''' '''
if not isinstance(logger, log.Logger): if not isinstance(logger, log.Logger):
return False return False
@@ -288,11 +299,11 @@ def update_gsm_com(serObj=None, config={}, logger=None):
:param config: :param config:
dictionary config object from JSON database dictionary config object from JSON database
:param log: :param logger:
logger object logger object
:return bool: :return bool:
True if ok, otherwise False True if OK, otherwise False
''' '''
if not isinstance(logger, log.Logger): if not isinstance(logger, log.Logger):
return False return False
@@ -322,7 +333,7 @@ def info_gsm_com(serObj=None, config={}, logger=None):
:param config: :param config:
dictionary config object from JSON database dictionary config object from JSON database
:param log: :param logger:
logger object logger object
:return bool: :return bool:
@@ -442,22 +453,21 @@ def verify_caller(buf="", num="", logger=None):
:param num: :param num:
Caller phone number Caller phone number
:param log: :param logger:
logger object logger object
:return bool: :return bool:
True if authorized, False otherwise True if authorized, False otherwise
''' '''
if not isinstance(logger, log.Logger):
return False, ""
if not isinstance(buf, str): if not isinstance(buf, str):
logger.error("error parameter, expecting str, get {}".format(type(buf))) logger.error("error parameter, expecting str, get {}".format(type(buf)))
return False, "" return False, ""
if not isinstance(num, str): if not isinstance(num, str):
logger.error("error parameter, expecting str, get {}".format(type(num))) logger.error("error parameter, expecting str, get {}".format(type(num)))
return False, "" return False, ""
if not isinstance(logger, log.Logger):
logger.error("error parameter, expecting logging.Logger, get {}".format(type(logger)))
return False, ""
outlst = buf[7:].split(',') outlst = buf[7:].split(',')
logger.debug("=> {}".format(outlst)) logger.debug("=> {}".format(outlst))
@@ -469,6 +479,12 @@ def verify_caller(buf="", num="", logger=None):
def listener(sock, logger): def listener(sock, logger):
''' Thread socket listener ''' Thread socket listener
:param sock:
socket object
:param logger:
logger object
''' '''
global FLAG_CONF_UPDATE global FLAG_CONF_UPDATE
@@ -509,13 +525,13 @@ def listener(sock, logger):
logger.debug("Fin du serveur de communication") logger.debug("Fin du serveur de communication")
return return
def verify_open_hours(conf=None, logger=None): def verify_open_hours(conf={}, logger=None):
''' Verify if GSM HAT must be opened with conf hours ''' Verify if GSM HAT must be opened with conf hours
:param conf: :param conf:
configuration object configuration object
:param log: :param logger:
logger object logger object
:return bool: :return bool:
@@ -523,6 +539,9 @@ def verify_open_hours(conf=None, logger=None):
''' '''
if not isinstance(logger, log.Logger): if not isinstance(logger, log.Logger):
return False return False
if not isinstance(conf, dict):
logger.error("error parameter, expecting dict, get {}".format(type(conf)))
return False
flag = False flag = False
my_date = date.today() my_date = date.today()
@@ -542,6 +561,9 @@ def verify_open_hours(conf=None, logger=None):
def init_module(): def init_module():
''' initialisation of GNSS/GPS/GSM HAT Module ''' initialisation of GNSS/GPS/GSM HAT Module
:return bool:
True if OK, otherwise False
''' '''
try: try:
GPIO.setmode(GPIO.BOARD) GPIO.setmode(GPIO.BOARD)
@@ -553,6 +575,9 @@ def init_module():
def setup_module(): def setup_module():
''' Setup module (Set/Reset) ''' Setup module (Set/Reset)
:return bool:
True if OK, otherwise False
''' '''
while True: while True:
try: try:
@@ -565,9 +590,18 @@ def setup_module():
#GPIO.cleanup() #GPIO.cleanup()
return True return True
def cron_verify_hours(op=None, config=None, logger=None): def cron_verify_hours(op='', config=None, logger=None):
''' cron job to check opened or closed hours ''' cron job to check opened or closed hours
''' '''
if not isinstance(logger, log.Logger):
return
if not isinstance(config, dict):
logger.error("error parameter, expecting dict, get {}".format(type(config)))
return
if not isinstance(op, str):
logger.error("error parameter, expecting str, get {}".format(type(op)))
return
global GSM_MODULE_STATE global GSM_MODULE_STATE
global GSM_MODULE_INIT_STATE global GSM_MODULE_INIT_STATE
if op == 'Horaires': if op == 'Horaires':
@@ -678,7 +712,7 @@ def main():
hl = log.FileHandler(os.path.join('/var/log/KineIntercom','Intercom.log')) hl = log.FileHandler(os.path.join('/var/log/KineIntercom','Intercom.log'))
fl.setLevel(log.DEBUG) fl.setLevel(log.DEBUG)
hl.setLevel(log.INFO) hl.setLevel(log.INFO)
formatter = log.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') formatter = log.Formatter('%(asctime)s - %(funcName)s - %(levelname)s - %(message)s')
fl.setFormatter(formatter) fl.setFormatter(formatter)
hl.setFormatter(formatter) hl.setFormatter(formatter)
logger.addHandler(fl) logger.addHandler(fl)
@@ -773,9 +807,6 @@ def main():
else: else:
GSM_MODULE_INIT_STATE = True GSM_MODULE_INIT_STATE = True
sys.exit(0)
# Retreive GSM infos # Retreive GSM infos
logger.info('Récupération des infos du module ...') logger.info('Récupération des infos du module ...')
ret = info_gsm_com(serObj=ser, config=config, logger=logger) ret = info_gsm_com(serObj=ser, config=config, logger=logger)