<jupyter_start><jupyter_text>PACKAGES, OPTIONS, CONSTANTS<jupyter_code>import os
from os import listdir
from os.path import isfile, join

import requests
from time import sleep
import credentials # script python con las credenciales almacenadas en un diccionario
import locale

import pandas as pd
pd.set_option('display.max_columns', None)
from datetime import datetime

from matplotlib import pyplot as plt
import seaborn as sns<jupyter_output><empty_output><jupyter_text>FUNCTIONS<jupyter_code>def etl_hospitalizaciones():
    """
    Generación de df histórico de ingresos, altas y ocupación de hospitales a partir de extracción de CMBD.
    """
    
    # Hospitalizaciones
    filename = 'CMBD_6_20181217-135856.xlsx'
    path = os.path.join('..', 'data', 'external', 'hospitalizaciones', filename)
    raw_df = pd.read_excel(path)
    
    date_cols = ['Fecha de ingreso', 'Fecha de alta']
    keep_cols = ['Sexo', 'Edad', 'Diagnóstico Principal', 'Estancia', 'Coste']
    
    for col in date_cols:
        """
        Dates are codified as integer in DDMMYYYY format, so there's no padding zero if day < 10, i.e:
        28042013 -> 2013-04-28
        7022013 -> 2013-02-07
        """
        raw_df[col + ' new'] = pd.to_datetime(raw_df[col].map(lambda col: str(col).zfill(8)), format='%d%m%Y', errors = 'coerce')
    
    interim_df = raw_df[keep_cols].copy()
    
    for col in date_cols:
        interim_df[col] = raw_df[col + ' new'].copy()
        
     # Ingresos y altas by date
    ingresos_diarios_df = interim_df.groupby(['Fecha de ingreso']).size().reset_index(name = 'Ingresos')
    ingresos_diarios_df.set_index('Fecha de ingreso', inplace = True)
    ingresos_diarios_df.index.names = ['Fecha']
    
    altas_diarias_df = interim_df.groupby(['Fecha de alta']).size().reset_index(name = 'Altas')
    altas_diarias_df.set_index('Fecha de alta', inplace = True)
    altas_diarias_df.index.names = ['Fecha']
    
    # Join by date, cumsum
    hospitalizaciones_df = altas_diarias_df.join(ingresos_diarios_df, how= 'outer')
    hospitalizaciones_df['Ingresos acumulados'] = hospitalizaciones_df['Ingresos'].cumsum()
    hospitalizaciones_df['Altas acumuladas'] = hospitalizaciones_df['Altas'].cumsum()
    hospitalizaciones_df['Ocupacion'] = hospitalizaciones_df['Ingresos acumulados'] - hospitalizaciones_df['Altas acumuladas']
    
    # Export to csv
    hospitalizaciones_df.to_csv(os.path.join('..', 'data', 'processed', 'hospitalizaciones'))
    
    return hospitalizaciones_df


def etl_calidad_aire():
    """
    Generación de df con datos de calidad del aire a partir de histórico en conjunto de ficheros csv.
    """
    
    path = os.path.join('..', 'data', 'external', 'calidad_aire')
    
    # Cargar tabla magnitudes, unidades y técnicas de medida.
    mag_uni_tec_df = pd.read_csv(os.path.join(path,'magnitudes_unidades_tecnicas_medida.csv'))
    
    # Listar ficheros de datos de calidad del aire.
    list_of_files = [f for f in listdir(path) if isfile(join(path, f))]
    list_of_csv_files = [file for file in list_of_files if file.endswith('csv') and file.startswith('datos')]
    
    # Concatenar ficheros de calidad del aire en df.
    raw_df = pd.concat([pd.read_csv(os.path.join(path, file), delimiter = ';') for file in list_of_csv_files])

    #raw_df = pd.read_csv(csv_path, delimiter = ';')
    raw_df = raw_df.merge(mag_uni_tec_df[['MAGNITUD ID', 'MAGNITUD ABREV']], 
                          left_on = 'MAGNITUD', 
                          right_on = 'MAGNITUD ID', 
                          how= 'left') # Pegar descripción de magnitud.
    
    # Trasponer wide to long => añomes to día
    id_cols = ['PROVINCIA', 'MUNICIPIO', 'ESTACION', 'MAGNITUD ABREV', 'PUNTO_MUESTREO', 'ANO', 'MES']
    value_cols = ['D' + str(n).zfill(2) for n in range(1,32,1)]
    
    interim_df = pd.melt(raw_df, id_cols, value_cols)
    interim_df['DIA'] = interim_df['variable'].map(lambda var: var[1:])
    interim_df['FECHA'] = pd.to_datetime(interim_df['ANO'].astype(str) 
                                          + interim_df['MES'].astype(str).str.zfill(2)
                                          + interim_df['DIA'], 
                                         format='%Y%m%d', errors = 'coerce')
    
    key = ['FECHA', 'MAGNITUD ABREV']
    calidad_aire_t_df = interim_df[key + ['value']].groupby(key).mean()
    
    # Trasponer long to wide => fecha, medida 1, medida 2, ..., medida n
    calidad_aire_df = calidad_aire_t_df.pivot_table(index= 'FECHA', 
                                                    columns= 'MAGNITUD ABREV', 
                                                    values= 'value')
    
    return calidad_aire_df


def etl_tabaquismo():
    """
    Generación de df con datos históricos de consumo de tabaco, a partir de agregación de datos de INCLASNS.
    """
    
    # Cargar csv
    path = os.path.join('..', 'data', 'external', 'tabaquismo')
    filename = "inclasns_tabaquismo.csv"
    raw_df = pd.read_csv(os.path.join(path, filename))
    
    # Generar fecha a partir de año
    raw_df['FECHA'] = pd.to_datetime(raw_df['ANO'].astype(str) + '01' + '01', 
                                     format='%Y%m%d', errors = 'coerce')
    
    # Pivotar wide to long => generar [CCAA, value]
    drop_cols = ['ANO']
    id_cols = ['FECHA', 'SEXO']
    value_cols = list(set(raw_df.columns.values).difference(set(drop_cols).union(set(id_cols)))) # Las columnas CCAA son todas las del df menos las id de pivot y las que dropeamos
    interim_df = pd.melt(raw_df, id_cols, value_cols)
    interim_df.rename(columns = {"variable": "CCAA", "value": "PCT_tabaquismo"}, # Renombrar por legibilidad de la siguiente trasposición
                      inplace = True)
    
    # Trasponer long to wide => generar [PCT_tabaquismo_H, PCT_Tabaquismo_M, PCT_Tabaquismo
    tabaquismo_df = interim_df.pivot_table(index= ['FECHA','CCAA'], 
                                           columns= 'SEXO', 
                                           values= 'PCT_tabaquismo')
    
    rename_cols = {col: 'PCT_tabaquismo_' + col.lower() for col in tabaquismo_df.columns.values}
    tabaquismo_df.rename(columns = rename_cols, 
                         inplace = True)
    
    # Dividir entre 100 para representar porcentaje.
    for col in tabaquismo_df.columns.values:
        tabaquismo_df[col] = tabaquismo_df[col] / 100
        
    tabaquismo_df.reset_index(inplace = True)
    tabaquismo_df.rename_axis(None, axis=1, inplace = True)
    tabaquismo_df.set_index(keys = 'FECHA', inplace = True)
    
    return tabaquismo_df


def etl_polen():
    """
    Generación de df con concentraciones de distintos tipos de polen a nivel mensual, a partir de calendario polínico de España.
    """
    
    path = os.path.join('..', 'data', 'external', 'polen')
    filename = "calendario_polinico_espana.csv"
    polen_df = pd.read_csv(os.path.join(path, filename))
    polen_df.set_index('Mes', inplace = True)
    
    return polen_df


def etl_climatologia():
    """
    Generación de df con información de datos climáticos.
    """
    
    # Parámetros de la request a la API
    anyo_ini = 2000
    anyo_fin = 2021
    endpoint = "https://opendata.aemet.es/opendata/api/valores/climatologicos/diarios/datos/"
    estacion = 3195
    querystring = {'api_key': credentials.api_keys['aemet_opendata']}
    headers = {'cache-control': "no-cache"}
    
    # La API solo permite traerse 5 años, de modo que lanzamos peticiones secuenciales, 1 por año.
    response_all_years = list() # almacenaremos los datos de todos los años en esta lista
    for anyo in range(anyo_ini, anyo_fin + 1):
        
        url = endpoint + "fechaini/{anyo}-01-01T00%3A00%3A00UTC/fechafin/{anyo}-12-31T23%3A59%3A59UTC/estacion/{estacion}".format(anyo = anyo, estacion = estacion)
        
        while True: # La API no permite sobrepasar cierto número de peticiones por minuto; si alcanzamos el límite, reintentamos pasados 60 segundos.
            try:
                first_response = requests.request("GET", url, headers=headers, params=querystring)
                print(first_response.json())
                second_response = requests.request("GET", first_response.json()['datos'], headers=headers, params=querystring)
                response_all_years.extend(second_response.json())
                
            except:
                sleep(60)
            break

    interim_df = pd.json_normalize(response_all_years)
    
    
    # La parte decimal de las variables numéricas viene separada por una coma, es necesario convertir a float
        
    def str_to_float(x): #
        try:
            _float = float(x.replace(',','.'))
        except:
            _float = float("NaN")
        return _float
    
    num_cols = ['altitud','tmed', 'prec', 'tmin', 'tmax', 'dir', 'velmedia', 'racha', 'presMax', 'presMin']
    interim_df[num_cols] = interim_df[num_cols].applymap(str_to_float)
    
    # Dropeamos las variables relativas a la estación meteorológica y establecemos la fecha como índice
    drop_cols = ['indicativo', 'nombre', 'provincia']
    climatologia_df = interim_df.drop(drop_cols, axis='columns')
    climatologia_df.set_index('fecha', inplace = True)
    
    return climatologia_df<jupyter_output><empty_output><jupyter_text>ETL PIPELINE<jupyter_code>hospitalizaciones_df = etl_hospitalizaciones()
calidad_aire_df = etl_calidad_aire()
tabaquismo_df = etl_tabaquismo()
polen_df = etl_polen()
climatologia_df = etl_climatologia()<jupyter_output>{'descripcion': 'exito', 'estado': 200, 'datos': 'https://opendata.aemet.es/opendata/sh/8bfbad4c', 'metadatos': 'https://opendata.aemet.es/opendata/sh/b3aa9d28'}
{'descripcion': 'exito', 'estado': 200, 'datos': 'https://opendata.aemet.es/opendata/sh/9f221341', 'metadatos': 'https://opendata.aemet.es/opendata/sh/b3aa9d28'}
{'descripcion': 'exito', 'estado': 200, 'datos': 'https://opendata.aemet.es/opendata/sh/78b8b6a9', 'metadatos': 'https://opendata.aemet.es/opendata/sh/b3aa9d28'}
{'descripcion': 'exito', 'estado': 200, 'datos': 'https://opendata.aemet.es/opendata/sh/2a5bb8c5', 'metadatos': 'https://opendata.aemet.es/opendata/sh/b3aa9d28'}
{'descripcion': 'exito', 'estado': 200, 'datos': 'https://opendata.aemet.es/opendata/sh/b1baa35b', 'metadatos': 'https://opendata.aemet.es/opendata/sh/b3aa9d28'}
{'descripcion': 'exito', 'estado': 200, 'datos': 'https://opendata.aemet.es/opendata/sh/52731fa2', 'metadatos': 'https://opendata.aemet.es/opendata/sh/b3aa9d28'}
{'descripcion': 'exito', 'es[...]<jupyter_text>DATA PREP PIPELINE<jupyter_code>join_sources_df = hospitalizaciones_df.join(calidad_aire_df, how= 'left')
join_sources_df = join_sources_df.join(tabaquismo_df.loc[tabaquismo_df['CCAA'] == "Comunidad de Madrid"], how= 'left')
join_sources_df['Mes'] = join_sources_df.index.month
join_sources_df = join_sources_df.join(polen_df, on= 'Mes', how= 'left')
join_sources_df = join_sources_df.join(climatologia_df, how= 'left')
join_sources_df.describe()
join_sources_df.info()<jupyter_output><class 'pandas.core.frame.DataFrame'>
DatetimeIndex: 2968 entries, 2007-08-02 to 2015-12-31
Data columns (total 47 columns):
 #   Column                           Non-Null Count  Dtype  
---  ------                           --------------  -----  
 0   Altas                            2921 non-null   float64
 1   Ingresos                         2967 non-null   float64
 2   Ingresos acumulados              2967 non-null   float64
 3   Altas acumuladas                 2921 non-null   float64
 4   Ocupacion                        2920 non-null   float64
 5   BEN                              2968 non-null   float64
 6   CH4                              2968 non-null   float64
 7   CO                               2968 non-null   float64
 8   EBE                              2968 non-null   float64
 9   NMHC                             2968 non-null   float64
 10  NO                               2968 non-null   float64
 11  NO2                              2968 non-null   float64
 12  NO[...]<jupyter_text>EDA<jupyter_code>fig_dims = (20, 10)
fig, ax = plt.subplots(figsize=fig_dims)
plot_cols = ['Altas','Ingresos','Ocupacion']
sns.lineplot(data=hospitalizaciones_df[plot_cols], ax = ax);

fig_dims = (20, 10)
fig, ax = plt.subplots(figsize=fig_dims)
sns.lineplot(data=calidad_aire_df, ax = ax);

fig_dims = (20, 10)
fig, ax = plt.subplots(figsize=fig_dims)
filtro = tabaquismo_df['CCAA'] == "Comunidad de Madrid"
sns.lineplot(data=tabaquismo_df.loc[filtro], ax = ax);

fig_dims = (20, 10)
fig, ax = plt.subplots(figsize=fig_dims)
sns.lineplot(data=climatologia_df, ax = ax);<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup