3 Automatización

En el taller anterior vimos toda una serie de pasos para preprocesar los datos de COVUD-19. En esta actividad lo único que vamos a hacer es definir un par de funciones que realizan todo el flujo de preproceso. De esta forma podemos repetir todo el procedimiento de forma fácil.

import os # hablar con el sistema operativo
import glob # listar directorios y ese tipo de operaciones
import itertools # herramientas para iterar objetos
from pathlib import Path # manipular rutas a directorios
import zipfile # comprimir y descomprimir archivos
import numpy as np # operaciones vectorizadas
import pandas as pd # DataFrames
from datetime import timedelta, date, datetime # Manejar fechas
import openpyxl # leer/escribir archivos de exel
import requests # Hablar con direcciones web
import logging

3.1 Bajar y guardar datos

En el taller anterior bajamos los datos directamente del sitio de la Secretaría de Salud, ahora vamos a automatizar el proceso de descarga de datos de forma que, desde Python, podamos descargar los datos y asegurarnos de que tenemos la última versión disponible.

Descargar y guardar archivos en Python es relativamente sencillo, vamos a usar tres módulos de la distribución base de Python:

os. Este módulo provee herramientas para interactuar con el sistema operativo. La vamos a usar para construir los paths en donde vamos a guardar los datos y preguntar si el archivo ya existe.
requests. Esta librería provee diferentes formas de interactuar con el protocolo HTTP. La vamos a usar para hacer las peticiones a la página y procesar la respuesta.
zipfile. Esta librería sirve para trabajar con archivos comprimidos en formato zip. En nuestro caso la usaremos para descomprimir los diccionarios.

La parte complicada de entender es el uso de requests pra comunicarse con la página en donde están los datos.

r = requests.get("https://www.centrogeo.org.mx/")
r.content[0:500]

b'\r\n\r\n\r\n\r\n\r\n\r\n\r\n\r\n<!DOCTYPE html>\r\n<html lang="es-es" dir="ltr" class=\'com_content view-featured itemid-101 home j31 mm-hover\'>\r\n<head>\r\n<base href="https://www.centrogeo.org.mx/" />\n\t<meta http-equiv="content-type" content="text/html; charset=utf-8" />\n\t<meta name="keywords" content="M\xc3\xa9xico, CONACYT, CentroGeo, Ciencias de Informaci\xc3\xb3n Geoespacial, Centro de Investigaci\xc3\xb3n, Ciencias de Informaci\xc3\xb3n, Investigaci\xc3\xb3n, Geoespacial" />\n\t<meta name="rights" content="Esta obra est\xc3\xa1 bajo una licencia d'

Como ven, una petición de tipo get simplemente nos regresa, a través de la propiedad content, el contenido de la respuesta del servidor. En el caso de la página de CentroGeo, el contenido es el HTML de la página (que podríamos ver mejor con un browser), pero en el caso de que la dirección apunte a un archivo de descarga, el contenido es el stream de datos del archivo. Este stream de datos lo podemos usar como entrada para escribir un archivo utilizando la función open.

La función open va a tomar como entrada el path en donde queremos guardar el archivo, este path puede ser simplemente una cadena de caracteres, sin embargo esto haría que nuestro código no fuera interoperable entre sistemas operativos, entonces, en lugar de escribir el path como cadena de caracteres, vamos a escribirlo como un objeto de os:

os.path.join("datos", "datos_covid.zip")

'datos/datos_covid.zip'

Esta forma de construir el path nos asegura que va a funcionar en cualquier sistema operativo.

Ya con estas explicaciones, podemos escribir la función que descarga los datos:

def bajar_datos_salud(directorio_datos='data/'):
    '''
        Descarga el ultimo archivo disponible en datos abiertos y los diccionarios correspondientes.
    '''
    fecha_descarga = datetime.now().date()
    url_datos = 'https://datosabiertos.salud.gob.mx/gobmx/salud/datos_abiertos/datos_abiertos_covid19.zip'    
    archivo_nombre = f'{fecha_descarga.strftime("%y%m%d")}COVID19MEXICO.csv.zip'
    archivo_ruta = os.path.join(directorio_datos, archivo_nombre)
    url_diccionario = 'https://datosabiertos.salud.gob.mx/gobmx/salud/datos_abiertos/diccionario_datos_covid19.zip'
    diccionario_ruta = os.path.join(directorio_datos, 'diccionario.zip')
    if os.path.exists(archivo_ruta):
        logging.debug(f'Ya existe {archivo_nombre}')
    else:
        print(f'Bajando datos...')
        r = requests.get(url_datos, allow_redirects=True)
        open(archivo_ruta, 'wb').write(r.content)
        r = requests.get(url_diccionario, allow_redirects=True)
        open(diccionario_ruta, 'wb').write(r.content)
        with zipfile.ZipFile(diccionario_ruta, 'r') as zip_ref:
          zip_ref.extractall(directorio_datos)

Para utilizar la función hacemos:

bajar_datos_salud('datos/')

Bajando datos...

3.2 Preproceso

Ahora, ya que tenemos los datos descargados, vamos a empaquetar en una función el flujo de preproceso que trabajamos en el taller anterior. Esta función va a tomar como entrada la carpeta en donde se encuentran los datos y dicionariosy el nombre del archivo de datos que queremos procesar. Toma dos parámetros adicionales, uno para decidir si queremos resolver o no las claves binarias y otro para definir la entidad que queremos procesar.

def carga_datos_covid19_MX(data_dir='datos/', archivo='datos_abiertos_covid19.zip', resolver_claves='si_no_binarias', entidad='09'):
    """
        Lee en un DataFrame el CSV con el reporte de casos de la Secretaría de Salud de México publicado en una fecha dada. Esta función
        también lee el diccionario de datos que acompaña a estas publicaciones para preparar algunos campos, en particular permite la funcionalidad
        de generar columnas binarias para datos con valores 'SI', 'No'.

        **Nota 2**: Por las actualizaciones a los formatos de datos, esta función sólo va a servir para archivos posteriores a 20-11-28

        resolver_claves: 'sustitucion', 'agregar', 'si_no_binarias', 'solo_localidades'. Resuelve los valores del conjunto de datos usando el
        diccionario de datos y los catálogos. 'sustitucion' remplaza los valores en las columnas, 'agregar'
        crea nuevas columnas. 'si_no_binarias' cambia valores SI, NO, No Aplica, SE IGNORA, NO ESPECIFICADO por 1, 0, 0, 0, 0 respectivamente.

    """
    catalogo_nombre ='201128 Catalogos.xlsx'
    catalogo_path = os.path.join(data_dir, catalogo_nombre)
    descriptores_nombre = '201128 Descriptores.xlsx'
    descriptores_path = os.path.join(data_dir, descriptores_nombre)
    data_file = os.path.join(data_dir, archivo)
    print(data_file)
    df = pd.read_csv(data_file, dtype=object, encoding='latin-1')
    if entidad is not None:
      df = df[df['ENTIDAD_RES'] == entidad]
    # Hay un error y el campo OTRA_COMP es OTRAS_COMP según los descriptores
    df.rename(columns={'OTRA_COM': 'OTRAS_COM'}, inplace=True)
    # Asignar clave única a municipios
    df['MUNICIPIO_RES'] = df['ENTIDAD_RES'] + df['MUNICIPIO_RES']
    df['CLAVE_MUNICIPIO_RES'] = df['MUNICIPIO_RES']
    # Leer catalogos
    nombres_catalogos = ['Catálogo de ENTIDADES',
                         'Catálogo MUNICIPIOS',
                         'Catálogo RESULTADO',
                         'Catálogo SI_NO',
                         'Catálogo TIPO_PACIENTE']
    nombres_catalogos.append('Catálogo CLASIFICACION_FINAL')
    nombres_catalogos[2] = 'Catálogo RESULTADO_LAB'

    dict_catalogos = pd.read_excel(catalogo_path,
                              nombres_catalogos,
                              dtype=str,
                              engine='openpyxl')

    entidades = dict_catalogos[nombres_catalogos[0]]
    municipios = dict_catalogos[nombres_catalogos[1]]
    tipo_resultado = dict_catalogos[nombres_catalogos[2]]
    cat_si_no = dict_catalogos[nombres_catalogos[3]]
    cat_tipo_pac = dict_catalogos[nombres_catalogos[4]]
    # Arreglar los catálogos que tienen mal las primeras líneas
    dict_catalogos[nombres_catalogos[2]].columns = ["CLAVE", "DESCRIPCIÓN"]
    dict_catalogos[nombres_catalogos[5]].columns = ["CLAVE", "CLASIFICACIÓN", "DESCRIPCIÓN"]


    clasificacion_final = dict_catalogos[nombres_catalogos[5]]


    # Resolver códigos de entidad federal
    cols_entidad = ['ENTIDAD_RES', 'ENTIDAD_UM', 'ENTIDAD_NAC']
    df['CLAVE_ENTIDAD_RES'] = df['ENTIDAD_RES']
    df[cols_entidad] = df[cols_entidad].replace(to_replace=entidades['CLAVE_ENTIDAD'].values,
                                               value=entidades['ENTIDAD_FEDERATIVA'].values)

    # Construye clave unica de municipios de catálogo para resolver nombres de municipio
    municipios['CLAVE_MUNICIPIO'] = municipios['CLAVE_ENTIDAD'] + municipios['CLAVE_MUNICIPIO']

    # Resolver códigos de municipio
    municipios_dict = dict(zip(municipios['CLAVE_MUNICIPIO'], municipios['MUNICIPIO']))
    df['MUNICIPIO_RES'] = df['MUNICIPIO_RES'].map(municipios_dict.get)

    # Resolver resultados

    df.rename(columns={'RESULTADO_LAB': 'RESULTADO'}, inplace=True)
    tipo_resultado['DESCRIPCIÓN'].replace({'POSITIVO A SARS-COV-2': 'Positivo SARS-CoV-2'}, inplace=True)

    tipo_resultado = dict(zip(tipo_resultado['CLAVE'], tipo_resultado['DESCRIPCIÓN']))
    df['RESULTADO'] = df['RESULTADO'].map(tipo_resultado.get)
    clasificacion_final = dict(zip(clasificacion_final['CLAVE'], clasificacion_final['CLASIFICACIÓN']))
    df['CLASIFICACION_FINAL'] = df['CLASIFICACION_FINAL'].map(clasificacion_final.get)
    # Resolver datos SI - NO

    # Necesitamos encontrar todos los campos que tienen este tipo de dato y eso
    # viene en los descriptores, en el campo FORMATO_O_FUENTE
    descriptores = pd.read_excel(f'{data_dir}201128 Descriptores_.xlsx',
                                 index_col='Nº',
                                 engine='openpyxl')
    descriptores.columns = list(map(lambda col: col.replace(' ', '_'), descriptores.columns))
    descriptores['FORMATO_O_FUENTE'] = descriptores.FORMATO_O_FUENTE.str.strip()

    datos_si_no = descriptores.query('FORMATO_O_FUENTE == "CATÁLOGO: SI_ NO"')
    cat_si_no['DESCRIPCIÓN'] = cat_si_no['DESCRIPCIÓN'].str.strip()

    campos_si_no = datos_si_no.NOMBRE_DE_VARIABLE
    nuevos_campos_si_no = campos_si_no

    if resolver_claves == 'agregar':
        nuevos_campos_si_no = [nombre_var + '_NOM' for nombre_var in campos_si_no]
    elif resolver_claves == 'si_no_binarias':
        nuevos_campos_si_no = [nombre_var + '_BIN' for nombre_var in campos_si_no]
        cat_si_no['DESCRIPCIÓN'] = list(map(lambda val: 1 if val == 'SI' else 0, cat_si_no['DESCRIPCIÓN']))

    df[nuevos_campos_si_no] = df[datos_si_no.NOMBRE_DE_VARIABLE].replace(
                                                to_replace=cat_si_no['CLAVE'].values,
                                                value=cat_si_no['DESCRIPCIÓN'].values)

    # Resolver tipos de paciente
    cat_tipo_pac = dict(zip(cat_tipo_pac['CLAVE'], cat_tipo_pac['DESCRIPCIÓN']))
    df['TIPO_PACIENTE'] = df['TIPO_PACIENTE'].map(cat_tipo_pac.get)

    df = procesa_fechas(df)

    return df

def procesa_fechas(covid_df):
    df = covid_df.copy()
    df['FECHA_INGRESO'] = pd.to_datetime(df['FECHA_INGRESO'], format="%Y-%m-%d")
    df['FECHA_SINTOMAS'] = pd.to_datetime(df['FECHA_SINTOMAS'], format="%Y-%m-%d")
    df['FECHA_DEF'] = pd.to_datetime(df['FECHA_DEF'], format="%Y-%m-%d", errors='coerce')
    df['DEFUNCION'] = (df['FECHA_DEF'].notna()).astype(int)
    df['EDAD'] = df['EDAD'].astype(int)
    return df

3.3 Preprocesar usando nuestras funciones

df = carga_datos_covid19_MX(entidad='09')
df

datos/datos_abiertos_covid19.zip

	FECHA_ACTUALIZACION	ID_REGISTRO	ORIGEN	SECTOR	ENTIDAD_UM	SEXO	ENTIDAD_NAC	ENTIDAD_RES	MUNICIPIO_RES	TIPO_PACIENTE	...	CARDIOVASCULAR_BIN	OBESIDAD_BIN	RENAL_CRONICA_BIN	TABAQUISMO_BIN	OTRO_CASO_BIN	TOMA_MUESTRA_LAB_BIN	TOMA_MUESTRA_ANTIGENO_BIN	MIGRANTE_BIN	UCI_BIN	DEFUNCION
1	2023-01-03	180725	2	9	CIUDAD DE MÉXICO	2	CIUDAD DE MÉXICO	CIUDAD DE MÉXICO	TLALPAN	HOSPITALIZADO	...	0	0	0	0	0	0	1	0	0	0
4	2023-01-03	1933c0	1	12	CIUDAD DE MÉXICO	2	CIUDAD DE MÉXICO	CIUDAD DE MÉXICO	IZTAPALAPA	AMBULATORIO	...	0	0	0	0	0	0	1	0	0	0
8	2023-01-03	0741e4	2	6	CIUDAD DE MÉXICO	2	CIUDAD DE MÉXICO	CIUDAD DE MÉXICO	MIGUEL HIDALGO	HOSPITALIZADO	...	0	0	1	0	0	1	0	0	0	1
13	2023-01-03	1c4d2e	2	9	CIUDAD DE MÉXICO	1	CIUDAD DE MÉXICO	CIUDAD DE MÉXICO	TLALPAN	AMBULATORIO	...	0	0	0	0	0	0	1	0	0	0
15	2023-01-03	0a6cd6	2	6	CIUDAD DE MÉXICO	1	NAYARIT	CIUDAD DE MÉXICO	IZTAPALAPA	AMBULATORIO	...	0	0	0	0	0	0	1	0	0	0
...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...	...
6393642	2023-01-03	m1cd235	2	12	MÉXICO	1	MÉXICO	CIUDAD DE MÉXICO	GUSTAVO A. MADERO	HOSPITALIZADO	...	0	0	0	0	0	0	0	0	0	0
6394417	2023-01-03	m0dbc4c	2	12	MÉXICO	1	AGUASCALIENTES	CIUDAD DE MÉXICO	AZCAPOTZALCO	AMBULATORIO	...	0	0	0	0	0	1	0	0	0	0
6394626	2023-01-03	m13431e	2	12	MÉXICO	1	CIUDAD DE MÉXICO	CIUDAD DE MÉXICO	AZCAPOTZALCO	AMBULATORIO	...	0	0	0	0	0	0	1	0	0	0
6394988	2023-01-03	m1493ea	2	12	MÉXICO	1	MÉXICO	CIUDAD DE MÉXICO	GUSTAVO A. MADERO	AMBULATORIO	...	1	0	0	0	0	1	0	0	0	0
6395781	2023-01-03	m0a22b8	2	12	MÉXICO	2	CIUDAD DE MÉXICO	CIUDAD DE MÉXICO	TLALPAN	AMBULATORIO	...	0	0	0	0	0	0	0	0	0	0

1896084 rows × 63 columns

3.4 Guardando el resultado

Listo, con nuestras funciones tenemos ya nuestros datos preprocesados, ahora vamos a guardarlos para poder utlizarlos rápidamente en otros notebooks. En general tenemos muchas opciones para guardar los datos, csv, por ejemplo. En esta ocasión vamos a usar un formato nativo de Python el pickle, que es una forma de serializar un objeto de Python. Pandas nos provee una función para guardar directamente un dataframe como pickle:

df.to_pickle("data/datos_covid_ene19.pkl")

En la documentación de to_pickle pueden ver las opcioones completas.

df.to_csv("datos/covid_enero_2023_procesados.csv")