Utilisateur:Arthurlafontaine

le but du projet est de construire une animation du déplacement du virus covid 19 sur une carte de Belgique depuis le début de la contamination et de la questionner pour en trouver la possible souche du virus en Belgique.

Pour ce faire je me base sur la data base de l'Institut scientifique de santé publique fédéral belge.[1]

pour Python j'utilise les bibliothèques Request et Alchemy.

Pour mon plan j'utilise deux sources d'information: sciensano: -les cas confirmés par date, age , sexe et province -les cas confirmés par date et municipalité

Rapidapi: -pour la géolocalisation des villes

code

# Ajout outils pour les requêtes http
import requests

# Ajout outils de manipulation du format json
import json

# Ajout Outils de manipulation d'objet de type Date/Time
from datetime import datetime, timedelta

# Ajout outils de manipulation de la base de donnée
import sqlalchemy
from sqlalchemy import create_engine, and_, or_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Date
from sqlalchemy.orm import sessionmaker

# Ajout outils expressions régulières 
import re


def prepareDb():
   
    # mise en place du moteur de base de donnée ( fichier covid.sqlite, encodage utf-8 )
    engine = create_engine('sqlite:///covid.sqlite', echo=False,encoding='utf-8')
    Session = sessionmaker(bind=engine)
   
    # Declaration de notre object de base do donnée 
    CovidDb = declarative_base()
    
    # Déclaration du schema de la table COVID19BE_CASES_AGESEX 
    class COVID19BE_CASES_AGESEX(CovidDb):
        __tablename__ = 'COVID19BE_CASES_AGESEX'
    
        id = Column(Integer, primary_key=True)
        date = Column(Date)
        province = Column(String)
        region = Column(String)
        sex = Column(String)
        cases = Column(Integer)
    
   # Déclaration du schema de la table COVID19BE_CASES_MUNI
    class COVID19BE_CASES_MUNI(CovidDb):
        __tablename__ = 'COVID19BE_CASES_MUNI'
    
        id = Column(Integer, primary_key=True)
        nis5 = Column(Integer)
        date = Column(Date)
        city = Column(String)
        admin = Column(String)
        province = Column(String)
        region = Column(String)
        cases = Column(Integer)

    # Déclaration du schema de la table GEOLOC
    class GEOLOC(CovidDb):
        __tablename__ = 'GEOLOC'

        id = Column(Integer, primary_key=True)
        city = Column(String)
        latitude = Column(String)
        longitude = Column(String)
   
    # Creation des tables si non existentes
    CovidDb.metadata.create_all(engine)
    
    # Creation du contexte pour manipuler les données de la Base de donnée
    session = Session()

Récupération et import des données dans la Database locale ( 25/04/2021 )

 

# Traitement table COVID19BE_CASES_AGESEX
    
    try:
        # Recupération des donnée sciensano AGE/SEXE/CAS  
        res = requests.get('https://epistat.sciensano.be/Data/COVID19BE_CASES_AGESEX.json')
        # Supression des données avant mise à jour ( methode plus rapide que de tester si la valeur existe  déja ) 
        session.query(COVID19BE_CASES_AGESEX).delete()
        # Boucle de traitement des données reçue dde sciensano 
        for item in res.json():
            # Sciensano renvoie des entrées parfois incomplètes ou invalide -> Test si tout les champs sont présent.
            if ( 'DATE' in item  and 'PROVINCE' in item and 'REGION' in item and 'SEX' in item and 'CASES' in item ):
                # Ajout de l'entrée dans la Base de donnée ( en mémoire ) 
                record = COVID19BE_CASES_AGESEX( date=datetime.strptime(item['DATE'],"%Y-%m-%d"), province=item['PROVINCE'], region=item['REGION'], sex=item['SEX'], cases=item['CASES'])
                session.add(record)
        # Ecriture des ajouts en mémore vers le fichier de la base de donnée
        session.commit()
    except:
        # Site inaccessible -> Pas de mise à jour des données
        print("AGE/SEXE/CAS: Utilisation des données existentes sans mise à jour...")

    
    # Traitement table COVID19BE_CASES_MUNI 
    
    try:
         # Recupération des donnée sciensano VILLE/CAS
        res = requests.get('https://epistat.sciensano.be/Data/COVID19BE_CASES_MUNI.json')
        # Supression des données avant mise à jour ( methode plus rapide que de tester si la valeur existe  déja )
        session.query(COVID19BE_CASES_MUNI).delete()
        # Boucle de traitement des données reçue dde sciensano
        for item in res.json():
            # Sciensano renvoie des entrées parfois incomplètes ou invalide -> Test si tout les champs sont présent.
            if ('NIS5' in item and 'DATE' in item and 'TX_DESCR_FR' in item and 'TX_ADM_DSTR_DESCR_FR' in item and 'PROVINCE' in item and 'REGION' in item and 'CASES' in item ):
                record = COVID19BE_CASES_MUNI( nis5=item['NIS5'], date=datetime.strptime(item['DATE'],"%Y-%m-%d"),city=re.sub(' *\(.*','',item['TX_DESCR_FR']),admin=item['TX_ADM_DSTR_DESCR_FR'],province=item['PROVINCE'], region=item['REGION'],cases=item['CASES'])
                session.add(record)
        # Ecriture des ajouts en mémore vers le fichier de la base de donnée
        session.commit()
    except:
        # Site inaccessible -> Pas de mise à jour des données
        print("VILLE/CAS: Utilisation des données existentes sans mise à jour...")

    # Traitement table  GEOLOC

    # Recherche dans la Base de donnée des Villes existentes ( Distinc() utilisé pour supprimer les doublons ) 
    cities = session.query(COVID19BE_CASES_MUNI.city).distinct(COVID19BE_CASES_MUNI.city)

    # Preparatin API rapidapi tel que documenté sur https://rapidapi.com/wirefreethought/api/geodb-cities/endpoints
    url = "https://wft-geo-db.p.rapidapi.com/v1/geo/cities"
    headers = {
    'x-rapidapi-key': "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    'x-rapidapi-host': "wft-geo-db.p.rapidapi.com"
    }

    # Boucle de traitement des villes présente dans la table COVID19BE_CASES_MUNI 
    for city in cities:
        # Recherche dans la table GEOLOC de la ville en cours de traitement 
        db_record = session.query(GEOLOC).filter_by(city = city.city).first()
        # La ville n'est pas dans la table GEOLOC
        if ( db_record == None ):
            # Definition de la recherche a faire sur le site wft-geo-db.p.rapidapi.com/v1/geo/cities
            querystring = {"countryIds":"BE","namePrefix":city.city,"languageCode":"FR"}
            # Recherche online stocké dans coord 
            try:
                coord = requests.request("GET", url, headers=headers, params=querystring)
                # Vérification de nombre d'occurence retourné par l'API
                if ( coord.json()['metadata']['totalCount'] == 0 ):
                    # Aucun résultat
                    print ( city.city, "non trouvé")
                else:
                    # Résulat disponible
                    print("Ajout ",city.city)
                    # Ajout des coordonnées Lat/Long dans la table GEOLOC 
                    record = GEOLOC( city = city.city, latitude = coord.json()['data'][0]['latitude'],longitude = coord.json()['data'][0]['longitude'])
                    session.add(record)
            except:
                # Site inaccessible -> abandon complet de la boucle de traitement
                print("Site injoignable")
                break
        # Ecriture des nouvelles données depuis la mémoire vers le fichier de la base de donnée.
    session.commit()
    return session