commit inicial

README.md

+# Consumidor de Datos de Kafka para el Cálculo Diario de Casos de COVID-19 para los Últimos 30 Días
+
+## Descripción
+
+Este consumidor de Kafka fue creado como parte de la plataforma EpI-PUMA y tiene como objetivo proveer una base de datos que contenga únicamente los casos confirmados de COVID-19 en México para los últimos 30 días agrupados por fecha, municipio, grupo de edad y tipo de caso.
+
+La base de datos generada por este conector fue pensada como una herramienta para demostrar la flexibilidad del sistema Apache Kafka, así como para facilitar la comparación de las predicciones realizadas por los modelos estadísticos de la plataforma EpI-PUMA con los datos reales de los casos de COVID-19 en México.
+
+## Base de Datos
+
+Este conector genera una base de datos de PostgreSQL de manera automática. La base de datos generada posee la siguiente estructura:
+
+![Screenshot](datos/database.png)
+
+## Requisitos
+
+Para poder ejecutar el proceso de carga de datos en Kafka es necesario tener instalados los siguientes paquetes en el sistema donde se ejecutará el proceso:
+  - Python 3 y Pip
+
+Además, en el mismo sistema o en uno dedicado se debe tener instalado y configurado un servidor de Apache Kafka con un tópico alimentado por el productor [KafkaCovid19](https://git.c3.unam.mx/alonso.ballesteros/KafkaCovid19).
+
+
+## Instalación
+
+Basta con ejecutar el siguiente comando para instalar las librerías necesarias para ejecutar el proceso de carga de datos en Kafka:
+
+```sh
+pip install -r requirements.txt
+```
+
+## Configuración
+
+Los ajustes del conector se realizan en el archivo **conector.py**. Estos ajustes deben realizarse antes de ejecutar el conector, ya que de lo contrario fallará con una excepción.
+
+### Base de datos de SQLite
+
+- Es simplemente el nombre del archivo que se usará para guardar la base de datos local. Esta base se emplea como un buffer para guardar los registros que se reciben de Kafka antes de guardarlos en la base de datos de PostgreSQL.
+- No es necesario crear el archivo de la base de datos, puesto que el conector lo creará de manera automática.
+
+```python
+# ----------- Datos SQLite ------------
+SQ_ARCHIVO = 'datos.db'
+```
+
+### Base de datos de PostgreSQL
+
+- Debe ser una base de datos de PostgreSQL completamente vacía, puesto que el conector creará las tablas necesarias de manera automática.
+
+```python
+# --------- Datos PostgreSQL ----------
+PG_HOST       = 'base.datos.com'
+PG_PUERTO     = 5433
+PG_USUARIO    = 'usuario'
+PG_CLAVE      = 'clave'
+PG_BASE_DATOS = 'epi_puma'
+```
+
+### Apache Kafka
+
+- En el campo de **KAFKA_BOOTSTRAP_SERVER** se debe omitir el protocolo **http://**.
+- El campo **KAFKA_TOPIC** debe coincidir con el nombre del tópico que alimenta el productor [KafkaCovid19](https://git.c3.unam.mx/alonso.ballesteros/KafkaCovid19).
+- Se sugiere que el campo **KAFKA_GROUP_ID** sea único para este consumidor de Kafka.
+
+```python
+# ------------ Datos Kafka ------------
+KAFKA_BOOTSTRAP_SERVER = 'servidor.kafka:9092'
+KAFKA_SCHEMA_REGISTRY  = 'http://servidor.kafka:8081'
+KAFKA_GROUP_ID         = 'covid19_epi_puma_30d'
+KAFKA_TOPIC            = 'covid19'
+KAFKA_PARTITION        = 0
+```
+
+### Conector
+
+- El campo **ARCHIVO_FECHA** denota el nombre del archivo que se usará para guardar la fecha del último registro que se guardó en la base de datos de PostgreSQL. Se recomienda no modificar este campo.
+
+- El campo **FECHA_INICIAL** denota la fecha a partir de la cual se comenzarán a guardar los registros en la base de datos de PostgreSQL. El valor se expresa en el formato **AAAA-MM-DD**.
+
+- El campo **TIEMPO_PURGA_REGISTROS** denota el tiempo que esperará el consumidor sin recibir nuevos registros antes de guardar los registros que tiene en memoria en la base de datos. El valor se expresa en segundos.
+
+- El campo **LIMITE_CARGA_REGISTROS** denota el número máximo de registros que se guardarán en memoria antes de hacerlos persistentes en la base de datos, esto se realiza de manera transaccional. El valor se expresa en número de registros.
+
+```python
+ARCHIVO_FECHA = 'fecha.txt'
+FECHA_INICIAL = '2021-03-01'
+TIEMPO_PURGA_REGISTROS = 60
+LIMITE_CARGA_REGISTROS = 100000
+```
+
+## Conector
+
+El conector posee una interfaz de línea de comandos que permite ejecutarlo de manera sencilla. La interfaz de línea de comandos se muestra a continuación:
+
+```sh
+python3 conector.py [-h] {estatus,iniciar,detener}
+
+positional arguments:
+  {estatus,iniciar,detener}
+
+optional arguments:
+  -h, --help            show this help message and exit
+```
+
+## Ejecución
+
+Para ejecutar el consumidor de Kafka basta con ingresar el siguiente comando:
+
+```sh
+python3 conector.py iniciar
+```
+
+Con este comando el consumidor de Kafka se ejecutará en segundo plano y se mantendrá en ejecución hasta que se ingrese el siguiente comando:
+
+```sh
+python3 conector.py detener
+```
+
+Además, se puede obtener el estatus del consumidor de Kafka con el siguiente comando:
+
+```sh
+python3 conector.py estatus
+```
+
+El conector guardará un archivo de registros en el archivo **conector.log**. Este archivo se encuentra en el directorio donde se ejecuta el conector.
\ No newline at end of file

conector.py

+# Desactiva el bytecode de Python
+import sys
+sys.dont_write_bytecode = True
+
+# Kafka
+import kafka
+
+# Librerías
+import time
+import atexit
+import pandas as pd
+from typing import List
+
+# SQLAlchemy
+import sqlalchemy as sa
+
+# Demonio
+from demonio import Demonio
+
+# Fecha
+from fecha import Fecha
+
+# Funciones
+import general
+
+# Modelos
+import modelos
+from modelos import RegistroLocal, RegistroKafka
+
+# --------------------------- Ajustes del Consumidor --------------------------
+
+# ----------- Datos SQLite ------------
+SQ_ARCHIVO = 'datos.db'
+
+# --------- Datos PostgreSQL ----------
+PG_HOST       = 'base.datos.com'
+PG_PUERTO     = 5433
+PG_USUARIO    = 'usuario'
+PG_CLAVE      = 'clave'
+PG_BASE_DATOS = 'epi_puma'
+
+# ------------ Datos Kafka ------------
+KAFKA_BOOTSTRAP_SERVER = 'servidor.kafka:9092'
+KAFKA_SCHEMA_REGISTRY  = 'http://servidor.kafka:8081'
+KAFKA_GROUP_ID         = 'covid19_epi_puma_30d'
+KAFKA_TOPIC            = 'covid19'
+KAFKA_PARTITION        = 0
+
+# Archivo de la fecha local
+ARCHIVO_FECHA = 'fecha.txt'
+
+# Fecha inicial de la base de datos remota
+FECHA_INICIAL = '2021-03-01'
+
+# Tiempo sin recibir mensajes para considerar que hubo un cambio de día (segundos)
+TIEMPO_PURGA_REGISTROS = 60
+
+# Los registros se guardarán en paquetes de este tamaño (cantidad de registros)
+LIMITE_CARGA_REGISTROS = 100000
+
+# ----------------------------- Variables Globales ----------------------------
+
+# Fecha local
+FECHA_LOCAL = Fecha(ARCHIVO_FECHA, pd.Timestamp(FECHA_INICIAL))
+
+# Conexiones
+CONEXION_KAFKA:  kafka.Conexion             = None
+CONEXION_LOCAL:  modelos.ConexionSQLite     = None
+CONEXION_REMOTA: modelos.ConexionPostgreSQL = None
+
+# Variables globales
+CAMBIO_DE_DIA            = False
+BASE_ACTUALIZADA         = False
+TRANSACCION_FINALIZADA   = False
+TIEMPO_ANTERIOR: float   = None
+DIA_ACTUAL: pd.Timestamp = None
+INDICE_KAFKA: int        = None
+
+# Variables globales de la carga de registros
+INSERTADOS:   List[RegistroKafka] = []
+ELIMINADOS:   List[RegistroKafka] = []
+ACTUALIZADOS: List[RegistroKafka] = []
+PENDIENTES:   List[RegistroKafka] = []
+
+# -------------------------- Funciones de inserción ---------------------------
+
+def insertar_registros_locales(insertados: pd.DataFrame):
+	''' Inserta los registros individuales de la lista INSERTADOS en la base de datos local '''
+
+	if insertados.empty:
+		return
+
+	# Se omiten los registros con un grupo de edad inválido
+	insertados.dropna(subset = ['GRUPO_EDAD'], inplace = True)
+
+	# Se filtran los registros positivos y no positivos
+	positivos = insertados[insertados['RESULTADO'] == 'CONFIRMADO']
+	no_positivos = insertados[insertados['RESULTADO'] != 'CONFIRMADO']
+
+	# Se filtran los registros positivos que tengan una antigüedad mayor a 30 días
+	inicio = FECHA_LOCAL.actual - pd.Timedelta(days = 30)
+	positivos = positivos[(positivos['FECHA_SINTOMAS'] >= inicio) | (positivos['FECHA_DEF'] >= inicio)]
+
+	# Se eliminan todos los registros que no sean positivos
+	ids_eliminados = no_positivos['ID_REGISTRO'].to_list()
+	eliminar: sa.sql.Delete = sa.delete(RegistroLocal)
+	CONEXION_LOCAL.sesion.execute(eliminar.where(RegistroLocal.ID_REGISTRO.in_(ids_eliminados)))
+
+	# Se insertan unicamente los registros positivos
+	if not positivos.empty:
+		CONEXION_LOCAL.sesion.execute(sa.insert(RegistroLocal), positivos.to_dict(orient = 'records'))
+
+def eliminar_registros_locales(eliminados: pd.DataFrame):
+	''' Elimina los registros individuales de la lista ELIMINADOS de la base de datos local '''
+
+	if eliminados.empty:
+		return
+
+	# Se eliminan todos los registros dados
+	ids_eliminados = eliminados['ID_REGISTRO'].to_list()
+	eliminar: sa.sql.Delete = sa.delete(RegistroLocal)
+	CONEXION_LOCAL.sesion.execute(eliminar.where(RegistroLocal.ID_REGISTRO.in_(ids_eliminados)))
+
+def eliminar_registros_anteriores():
+	''' Elimina los registros de la base local que tengan una antigüedad mayor a 30 días '''
+
+	# Se obtiene la fecha de inicio
+	inicio = FECHA_LOCAL.actual - pd.Timedelta(days = 30)
+
+	# Se eliminan los registros sin fecha de defunción con una antigüedad mayor a 30 días
+	CONEXION_LOCAL.sesion.query(RegistroLocal) \
+	.filter(RegistroLocal.FECHA_DEF == None) \
+	.filter(RegistroLocal.FECHA_SINTOMAS < inicio).delete()
+
+	# Se eliminan los registros con fecha de defunción con ambas condiciones verdaderas:
+	# - Una antigüedad mayor a 30 días
+	# - Una fecha de defunción mayor a 30 días
+	CONEXION_LOCAL.sesion.query(RegistroLocal) \
+	.filter(RegistroLocal.FECHA_DEF != None) \
+	.filter(
+		(RegistroLocal.FECHA_SINTOMAS < inicio) &
+		(RegistroLocal.FECHA_DEF < inicio)
+	).delete()
+
+def insertar_registros_remotos():
+	'''
+		Carga los datos almacenados localmente en la base de datos remota
+		- La base de datos local almacena cada registro confirmado de manera individual
+		- La base de datos remota almacena los registros por fecha, municipio y grupo de edad
+	'''
+
+	# Se cargan los registros almacenados localmente
+	registros = pd.read_sql(CONEXION_LOCAL.sesion.query(RegistroLocal).statement, CONEXION_LOCAL.enlace)
+
+	# Se agrega el campo de resultado
+	registros['RESULTADO'] = 'POSITIVO'
+
+	# Se convierten las fechas a formato datetime
+	registros['FECHA_SINTOMAS'] = pd.to_datetime(registros['FECHA_SINTOMAS'])
+	registros['FECHA_DEF'] = pd.to_datetime(registros['FECHA_DEF'], errors = 'coerce')
+
+	# Se convierten los valores NaT a None
+	registros.replace({pd.NaT: None}, inplace = True)
+
+	# Se guardan los registros en la base de datos remota
+	general.cargar_valores_reales(CONEXION_REMOTA, registros, FECHA_LOCAL.actual - pd.Timedelta(days = 1))
+
+# ------------------------- Funciones del consumidor --------------------------
+
+def cantidad_registros_pendientes():
+	''' Regresa la cantidad de registros pendientes por cargar en la base de datos '''
+
+	return len(INSERTADOS) + len(ELIMINADOS) + len(ACTUALIZADOS) + len(PENDIENTES)
+
+def refrescar_dia_actual(fecha_registro: pd.Timestamp, purgar = False):
+	'''
+		Refresca la fecha actual según la fecha de un registro y determina si dicho registro
+		corresponde a un dia diferente al actual, en cuyo caso se toma como un cambio de dia
+	'''
+
+	global DIA_ACTUAL, CAMBIO_DE_DIA
+
+	# Si la fecha del registro es mayor a la fecha actual, se toma como un cambio de dia
+	if fecha_registro > pd.Timestamp(DIA_ACTUAL):
+		if purgar:
+			print('Purgando registros pendientes...\n')
+		DIA_ACTUAL = fecha_registro
+		CAMBIO_DE_DIA = True
+
+	# Si el dia actual es nulo, se asigna la fecha del registro
+	elif DIA_ACTUAL is None:
+		DIA_ACTUAL = fecha_registro
+
+def enlistar_registro(registro: RegistroKafka):
+	''' Enlista el registro en la lista correspondiente para su posterior procesamiento '''
+
+	# Si se detecta un cambio de dia, se continúa con el siguiente ciclo y se guarda
+	# el ultimo registro como pendiente para procesarlo en el siguiente ciclo
+	if CAMBIO_DE_DIA:
+		PENDIENTES.append(registro)
+	elif registro.TYPE == 'INSERT':
+		INSERTADOS.append(registro)
+	elif registro.TYPE == 'DELETE':
+		ELIMINADOS.append(registro)
+	elif registro.TYPE == 'UPDATE':
+		ACTUALIZADOS.append(registro)
+
+def inicio():
+	''' Función de inicialización del consumidor de Kafka '''
+
+	global CONEXION_LOCAL, CONEXION_REMOTA, CONEXION_KAFKA, TIEMPO_ANTERIOR
+
+	# -------------------------- Finalización del script --------------------------
+
+	def finalizar_proceso():
+		''' Finaliza el script y cierra las conexiones a la base de datos y kafka '''
+
+		print('\nFinalizando proceso...')
+
+		# Si la transacción no fue finalizada o hay algún registro pendiente por
+		# cargar en la base de datos, se regresa el indice de kafka al inicio
+		if not TRANSACCION_FINALIZADA or cantidad_registros_pendientes() > 0:
+			CONEXION_KAFKA.regresar_indice_remoto()
+			print('Indice de Kafka restaurado')
+
+		# Se cierran las conexiones de las bases de datos y kafka
+		CONEXION_LOCAL.cerrar()
+		CONEXION_REMOTA.cerrar()
+		CONEXION_KAFKA.cerrar()
+
+		print('Conexiones finalizadas')
+
+	atexit.register(finalizar_proceso)
+
+	# ------------------------- Conexión a la base local --------------------------
+
+	print('Conectando a la base de datos local...')
+
+	CONEXION_LOCAL = modelos.ConexionSQLite \
+	(
+		SQ_ARCHIVO
+	)
+
+	print('  Base de datos conectada\n')
+
+	# ------------------------ Conexión a la base remota --------------------------
+
+	print('Conectando a la base de datos remota...')
+
+	CONEXION_REMOTA = modelos.ConexionPostgreSQL \
+	(
+		PG_HOST,
+		PG_PUERTO,
+		PG_USUARIO,
+		PG_CLAVE,
+		PG_BASE_DATOS
+	)
+
+	print('  Base de datos conectada\n')
+
+	# --------------------------- Conexión a Kafka -------------------------------
+
+	print('Conectando a Kafka...')
+
+	CONEXION_KAFKA = kafka.Conexion \
+	(
+		KAFKA_BOOTSTRAP_SERVER,
+		KAFKA_GROUP_ID,
+		KAFKA_SCHEMA_REGISTRY,
+		KAFKA_TOPIC,
+		KAFKA_PARTITION,
+		RegistroKafka.desde_diccionario
+	)
+
+	print('  Kafka conectado\n')
+
+	# Se inicializa la base de datos remota
+	modelos.inicializar_base_remota(CONEXION_REMOTA)
+
+	# Se obtiene el tiempo inicial
+	TIEMPO_ANTERIOR = time.time()
+
+def ciclo():
+	''' Ciclo principal del consumidor de Kafka '''
+
+	global CAMBIO_DE_DIA, TRANSACCION_FINALIZADA, TIEMPO_ANTERIOR
+
+	# Si se detecta un cambio de dia o se alcanza el limite de registros
+	# cargados, se procesan los registros cargados hasta el momento
+	if CAMBIO_DE_DIA or cantidad_registros_pendientes() >= LIMITE_CARGA_REGISTROS:
+
+		# Se marca la transacción como no finalizada
+		TRANSACCION_FINALIZADA = False
+
+		print(f'Procesando {len(INSERTADOS)} registros a insertar...')
+		print(f'Procesando {len(ELIMINADOS)} registros a eliminar...')
+		print(f'Procesando {len(ACTUALIZADOS)} registros a actualizar...')
+
+		# Se convierten los registros de Kafka en DataFrames
+		insertados = general.generar_dataframe(INSERTADOS)
+		eliminados = general.generar_dataframe(ELIMINADOS)
+		actualizados = general.generar_dataframe(ACTUALIZADOS)
+
+		print('Realizando cambios en la base de datos local...')
+
+		# Se eliminan los registros de la lista de 'ELIMINADOS' y 'ACTUALIZADOS'
+		eliminar_registros_locales(eliminados)
+		eliminar_registros_locales(actualizados)
+
+		# Se insertan los registros de la lista de 'INSERTADOS' y 'ACTUALIZADOS'
+		insertar_registros_locales(insertados)
+		insertar_registros_locales(actualizados)
+
+		print('Persistiendo los cambios en la base de datos local...')
+
+		# Se hacen efectivos los cambios en la base de datos local
+		CONEXION_LOCAL.sesion.commit()
+
+		# Se marca la transacción como finalizada
+		TRANSACCION_FINALIZADA = True
+
+		# Se recorre el indice inicial de kafka al indice del ultimo registro procesado
+		CONEXION_KAFKA.avanzar_indice_local()
+
+		print('    Base de datos local actualizada!\n')
+
+		# La base de datos remota se actualiza unicamente si se detecta un cambio de dia
+		if CAMBIO_DE_DIA:
+			print('Cambio de dia detectado...\n')
+			CAMBIO_DE_DIA = False
+
+			# Se actualiza la base de datos remota hasta que se alcance el dia actual
+			while FECHA_LOCAL < DIA_ACTUAL:
+				print('Día local: ', FECHA_LOCAL.actual.date())
+				print('Actualizando la base de datos remota...')
+
+				# Se actualizan los datos remotos
+				eliminar_registros_anteriores()
+				insertar_registros_remotos()
+
+				print('Persistiendo los cambios en las bases de datos...')
+
+				# Se hacen efectivos los cambios en las bases de datos
+				CONEXION_LOCAL.sesion.commit()
+				CONEXION_REMOTA.sesion.commit()
+
+				print('    Base de datos remota actualizada!\n')
+
+				# Se avanza la fecha local
+				FECHA_LOCAL.avanzar()
+
+			print(f'Base de datos remota al día! ({FECHA_LOCAL.actual.date()})')
+
+		# Se limpian las listas de registros
+		INSERTADOS.clear()
+		ELIMINADOS.clear()
+		ACTUALIZADOS.clear()
+
+	# Si no hay registros pendientes, se obtiene el siguiente registro
+	if len(PENDIENTES) == 0:
+
+		# Se obtiene el siguiente registro y el tiempo actual
+		registro: RegistroKafka = CONEXION_KAFKA.obtener_registro()
+		tiempo_actual = time.time()
+
+		# Si el registro no es nulo, se refresca el tiempo anterior
+		if registro is not None:
+			TIEMPO_ANTERIOR = tiempo_actual
+
+		# Si transcurre cierto tiempo sin recibir mensajes, se considera que hubo
+		# un cambio de dia y se cargan los datos leídos hasta el momento
+		elif tiempo_actual - TIEMPO_ANTERIOR >= TIEMPO_PURGA_REGISTROS:
+
+			# Solo se refresca el dia actual si hay registros cargados en las listas
+			if cantidad_registros_pendientes() > 0:
+				refrescar_dia_actual(DIA_ACTUAL + pd.Timedelta(days = 1), purgar = True)
+
+	# En caso contrario, se obtiene el siguiente registro de la lista de pendientes
+	else:
+		registro = PENDIENTES.pop(0)
+
+	# Si el registro es nulo, se continúa con el siguiente ciclo
+	if registro is None:
+		return
+
+	# Se obtiene la fecha del registro
+	fecha_registro = registro.fecha_actualizacion
+
+	# La fecha de actualización de los registros eliminados es la fecha del archivo
+	# anterior, por lo que se incrementa en un dia para que coincida con el dia actual
+	if registro.TYPE == 'DELETE':
+		fecha_registro += pd.Timedelta(days = 1)
+
+	# Se refresca el dia actual
+	refrescar_dia_actual(fecha_registro)
+
+	# Se enlista el registro en la lista correspondiente
+	enlistar_registro(registro)
+
+# Se muestra la interfaz de consola del demonio
+Demonio(inicio, ciclo, 'consumidor.pid', 'consumidor.log').interfaz()
\ No newline at end of file

demonio.py

+''' Clase para facilitar la creación de demonios '''
+
+import os
+import sys
+import log
+import time
+import signal
+import daemon
+import argparse
+from enum import Enum
+from pid import PidFile
+from pathlib import Path
+from typing import Callable
+
+class Resultado(Enum):
+	EXITO = 0
+	ERROR = 1
+
+	@classmethod
+	def negar(cls, resultado: 'Resultado'):
+		return cls.EXITO if resultado == cls.ERROR else cls.ERROR
+
+class Demonio:
+	def __init__(self, inicio: Callable, ciclo: Callable, archivo_pid: str, archivo_log: str):
+		self.ruta = Path(__file__).resolve().parent
+		self.archivo_pid = archivo_pid
+		self.archivo_log = archivo_log
+
+		# La funcion de inicio se ejecuta una sola vez al iniciar el proceso
+		self.inicio = inicio
+
+		# La funcion de ciclo se ejecuta en un ciclo infinito
+		self.ciclo = ciclo
+
+	@property
+	def corriendo(self):
+		return (self.ruta / self.archivo_pid).exists()
+
+	def estatus(self) -> Resultado:
+		if self.corriendo:
+			print('El proceso se esta ejecutando')
+			return Resultado.EXITO
+		else:
+			print('El proceso no se esta ejecutando')
+			return Resultado.ERROR
+
+	def iniciar(self) -> Resultado:
+		if self.corriendo:
+			return Resultado.negar(self.estatus())
+
+		demonio = daemon.DaemonContext \
+		(
+			stdout= open(self.ruta / self.archivo_log, 'a+'),
+			stderr= open(self.ruta / self.archivo_log, 'a+'),
+			pidfile= PidFile(self.archivo_pid, self.ruta)
+		)
+
+		with demonio:
+
+			# Se redirigen los logs
+			log.configurar_logs()
+			sys.stdout = log.LogHandle(log.INFO)
+			sys.stderr = log.LogHandle(log.ERROR)
+
+			# Se muestra el mensaje de inicio
+			print('Proceso ejecutandose en segundo plano')
+			print(f'ID del proceso: {os.getpid()}')
+
+			# Se ejecuta la función de inicio
+			self.inicio()
+
+			# Se ejecuta la función de ciclo
+			while True:
+				try:
+					self.ciclo()
+				except KeyboardInterrupt:
+					print('Proceso detenido por el usuario')
+					return Resultado.EXITO
+				except Exception as e:
+					log.error('Proceso detenido debido a un error')
+					log.error(e)
+					return Resultado.ERROR
+
+	def detener(self):
+		if not self.corriendo:
+			return self.estatus()
+
+		# Se obtiene el PID del proceso
+		with (self.ruta / self.archivo_pid).open() as archivo:
+			pid = archivo.read()
+
+		print('Deteniendo proceso...')
+
+		# Se detiene el proceso
+		while self.corriendo:
+			try:
+				os.kill(int(pid), signal.SIGINT)
+				time.sleep(1)
+			except:
+				log.error('No se pudo detener el proceso')
+				return Resultado.ERROR
+
+		print('Proceso detenido')
+		return Resultado.EXITO
+
+	def interfaz(self):
+		parser = argparse.ArgumentParser()
+		parser.add_argument('accion', choices= ['estatus', 'iniciar', 'detener'])
+		args = parser.parse_args()
+
+		if args.accion == 'estatus':
+			resultado = self.estatus()
+		elif args.accion == 'iniciar':
+			resultado = self.iniciar()
+		elif args.accion == 'detener':
+			resultado = self.detener()
+
+		sys.exit(resultado.value)
\ No newline at end of file

fecha.py

+''' Permite llevar el control de una fecha mediante un archivo de texto '''
+
+# Librerías
+import pandas as pd
+from pathlib import Path
+
+class Fecha:
+	''' Representa una fecha almacenada en un archivo de texto '''
+
+	def __init__(self, archivo: str, fecha = pd.Timestamp.now()):
+		ruta = Path(__file__).resolve().parent
+		self.archivo = ruta / archivo
+
+		if not self.archivo.exists():
+			self.guardar(fecha)
+
+	def __lt__(self, other: pd.Timestamp):
+		''' Regresa verdadero si la fecha almacenada en el archivo es menor a la fecha dada '''
+
+		return self.actual.date() < other.date()
+
+	@property
+	def actual(self):
+		''' Regresa la fecha almacenada en el archivo '''
+
+		return pd.Timestamp(self.archivo.read_text())
+
+	def guardar(self, fecha: pd.Timestamp):
+		''' Guarda la fecha dada en el archivo '''
+
+		self.archivo.write_text(str(fecha.strftime('%Y-%m-%d')))
+
+	def avanzar(self, dias: int = 1):
+		''' Avanza la fecha almacenada en el archivo '''
+
+		self.guardar(self.actual + pd.Timedelta(days = dias))
\ No newline at end of file

general.py

+import pandas as pd
+import sqlalchemy as sa
+from typing import List
+
+# Módulos locales
+from modelos import Conexion, RegistroKafka, GRUPO_EDAD, TIPO_VALOR
+from modelos import Estado, Municipio, GrupoEdad, TipoValor, DetalleValor, ValorReal
+
+def generar_dataframe(registros: List[RegistroKafka]):
+	''' Genera un dataframe a partir de los valores de una lista de registros '''
+
+	# Si no hay registros, se regresa un dataframe vacío
+	if len(registros) == 0:
+		return pd.DataFrame()
+
+	# Se crea el dataframe con los registros y las columnas especificadas
+	df = pd.DataFrame \
+	(
+		[registro.__dict__ for registro in registros],
+		columns =
+		[
+			'ID_REGISTRO',
+			'FECHA_ACTUALIZACION',
+			'ENTIDAD_RES', 'MUNICIPIO_RES',
+			'FECHA_SINTOMAS', 'FECHA_DEF',
+			'EDAD', 'CLASIFICACION_FINAL'
+		]
+	)
+
+	# Se convierten los campos de fecha
+	df['FECHA_ACTUALIZACION'] = pd.to_datetime(df['FECHA_ACTUALIZACION'])
+	df['FECHA_SINTOMAS'] = pd.to_datetime(df['FECHA_SINTOMAS'])
+	df['FECHA_DEF'] = pd.to_datetime(df['FECHA_DEF'], format='AAAA-MM-DD', errors = 'coerce')
+
+	# Función 'resultado' que interpreta la columna de 'CLASIFICACION_FINAL' y 'FECHA_DEF'
+	def resultado(valor):
+		if valor in [1, 2, 3]:
+			return 'CONFIRMADO'
+		elif valor in [4, 5]:
+			return 'INVIABLE'
+		elif valor in [6]:
+			return 'SOSPECHOSO'
+		elif valor in [7]:
+			return 'NEGATIVO'
+		else:
+			return 'DESCONOCIDO'
+
+	# Se crea una nueva columna 'RESULTADO' y se elimina la columna 'CLASIFICACION_FINAL'
+	df['RESULTADO'] = df['CLASIFICACION_FINAL'].apply(resultado)
+	del df['CLASIFICACION_FINAL']
+
+	# Se crea una nueva columna 'GRUPO_EDAD' con los grupos de edad
+	grupos = [17, 29, 39, 49, 59, 1000000]
+	etiquetas = [str(grupo) for grupo in GRUPO_EDAD]
+	df['GRUPO_EDAD'] = pd.cut(df['EDAD'], grupos, labels = etiquetas)
+
+	# Se reemplazan todos los valores de 'NaT' por nulos
+	df.replace({pd.NaT: None}, inplace = True)
+
+	return df
+
+def calcular_confirmados(registros: pd.DataFrame, tipo: TIPO_VALOR, fecha: pd.Timestamp):
+	''' Calcula el número de confirmados en los últimos 30 días a partir de la fecha dada '''
+
+	# Se filtran los datos positivos para la fecha dada según el pivote
+	confirmados = pd.DataFrame()
+	fecha_inicial = fecha - pd.Timedelta(days = 30)
+	datos_periodo = registros[registros[tipo.value] >= fecha_inicial]
+	positivos = datos_periodo[datos_periodo['RESULTADO'] == 'POSITIVO']
+
+	# Se calcula el número de registros confirmados agrupados por entidad y municipio
+	for grupo in GRUPO_EDAD:
+		grupo_confirmados = positivos[positivos['GRUPO_EDAD'] == str(grupo)]
+		grupo_confirmados = grupo_confirmados.groupby(['ENTIDAD_RES', 'MUNICIPIO_RES'])
+		grupo_confirmados = grupo_confirmados.size().reset_index(name = 'CONFIRMADOS')
+		grupo_confirmados['GRUPO_EDAD'] = str(grupo)
+		confirmados = pd.concat([confirmados, grupo_confirmados])
+
+	return confirmados
+
+def obtener_foraneas_general(conexion: Conexion, datos: pd.DataFrame):
+	'''
+		Obtiene las llaves foráneas de los siguientes campos:
+			- ENTIDAD_RES   -> 'fk_estado'
+			- MUNICIPIO_RES -> 'fk_municipio'
+			- GRUPO_EDAD    -> 'fk_grupo_edad'
+			- TIPO_VALOR    -> 'fk_tipo_valor'
+	'''
+
+	# Se obtienen las llaves foráneas de los estados
+	estados = pd.read_sql(conexion.sesion.query(Estado).statement, conexion.enlace)
+	estados = estados[['id', 'id_dge']]
+	estados.columns = ['fk_estado', 'ENTIDAD_RES']
+	datos = pd.merge(datos, estados, on = 'ENTIDAD_RES', how = 'left')
+
+	# Se obtienen las llaves foráneas de los municipios
+	municipios = pd.read_sql(conexion.sesion.query(Municipio).statement, conexion.enlace)
+	municipios = municipios[['id', 'id_dge', 'fk_estado']]
+	municipios.columns = ['fk_municipio', 'MUNICIPIO_RES', 'fk_estado']
+	datos = pd.merge(datos, municipios, on = ['fk_estado', 'MUNICIPIO_RES'], how = 'left')
+
+	# Se obtienen las llaves foráneas de los grupos de edad
+	grupos_edad = pd.read_sql(conexion.sesion.query(GrupoEdad).statement, conexion.enlace)
+	grupos_edad = grupos_edad[['id', 'nombre']]
+	grupos_edad.columns = ['fk_grupo_edad', 'GRUPO_EDAD']
+	datos = pd.merge(datos, grupos_edad, on = 'GRUPO_EDAD', how = 'left')
+
+	# Se obtienen las llaves foráneas de los tipos de valor
+	tipos_valor = pd.read_sql(conexion.sesion.query(TipoValor).statement, conexion.enlace)
+	tipos_valor = tipos_valor[['id', 'tipo']]
+	tipos_valor.columns = ['fk_tipo_valor', 'TIPO_VALOR']
+	datos = pd.merge(datos, tipos_valor, on = 'TIPO_VALOR', how = 'left')
+
+	# Se eliminan los renglones cuyo municipio no se encuentran en la base de datos
+	datos.dropna(subset = ['fk_municipio'], inplace = True)
+
+	# Se asigna la columna de 'valor' con los valores de la columna 'CONFIRMADOS'
+	datos['valor'] = datos['CONFIRMADOS']
+
+	return datos
+
+def obtener_foraneas_detalles(conexion: Conexion, datos: pd.DataFrame):
+	''' Obtiene la llave foránea de la tabla de detalles '''
+
+	# Se obtienen las llaves foráneas de la tabla de detalles
+	detalle_valores = pd.read_sql(conexion.sesion.query(DetalleValor).statement, conexion.enlace)
+	detalle_valores.rename(columns = {'id': 'fk_detalle_valor'}, inplace = True)
+	detalle_valores['fecha'] = pd.to_datetime(detalle_valores['fecha'])
+	interseccion = ['fecha', 'fk_municipio', 'fk_grupo_edad', 'fk_tipo_valor']
+	return pd.merge(datos, detalle_valores, on = interseccion, how = 'left')
+
+def cargar_valores_reales(conexion: Conexion, registros: pd.DataFrame, fecha: pd.Timestamp):
+	''' Carga los valores reales en la base de datos '''
+
+	datos = []
+
+	# Se obtienen los casos confirmados para cada tipo de valor
+	for tipo in TIPO_VALOR:
+		confirmados = calcular_confirmados(registros, tipo, fecha)
+		confirmados['TIPO_VALOR'] = str(tipo)
+		datos.append(confirmados)
+
+	# Se concatenan los datos de los diferentes tipos de valor y se agrega la fecha
+	confirmados = pd.concat(datos)
+	confirmados['fecha'] = fecha
+
+	# Se obtienen las llaves foráneas generales
+	confirmados = obtener_foraneas_general(conexion, confirmados)
+
+	# Se insertan los datos de la tabla de detalles en la base de datos
+	conexion.sesion.execute(sa.insert(DetalleValor), confirmados.to_dict(orient = 'records'))
+
+	# Se obtienen las llaves foráneas de la tabla de detalles
+	confirmados = obtener_foraneas_detalles(conexion, confirmados)
+
+	# Se insertan los valores reales en la base de datos
+	conexion.sesion.execute(sa.insert(ValorReal), confirmados.to_dict(orient = 'records'))
\ No newline at end of file

kafka.py

+''' Contiene la clase 'Conexión' que representa la conexión de un consumidor de Kafka a un tópico '''
+
+from typing import Optional, Callable, Any
+
+# Confluent Kafka
+from confluent_kafka.schema_registry import SchemaRegistryClient
+from confluent_kafka.schema_registry.avro import AvroDeserializer
+from confluent_kafka import Consumer, TopicPartition, OFFSET_STORED
+from confluent_kafka.serialization import SerializationContext, MessageField
+
+class Conexion:
+	''' Representa la conexión de un consumidor de Kafka a un tópico '''
+
+	def __init__ \
+	(
+		self,
+		bootrap_servers: str,
+		group_id: str,
+		schema_registry_url: str,
+		topico_kafka: str,
+		particion_kafka: int,
+		from_dict: Callable[[dict, Any], Optional[Any]]
+	):
+
+		# Se inicializa el esquema y el deserializador de Avro
+		self.schema_registry_client = SchemaRegistryClient({'url': schema_registry_url})
+		self.avro_deserializer = AvroDeserializer(self.schema_registry_client, from_dict = from_dict)
+
+		# Se crea el consumidor de Kafka
+		self.consumer_settings = \
+		{
+			'bootstrap.servers': bootrap_servers,
+			'group.id': group_id,
+			'auto.offset.reset': 'earliest'
+		}
+		self.consumer = Consumer(self.consumer_settings)
+
+		# Se asigna el consumidor a la partición del tópico y se crea el contexto de serialización
+		self.topic_partition = TopicPartition(topico_kafka, particion_kafka, OFFSET_STORED)
+		self.serialization_context = SerializationContext(self.topic_partition.topic, MessageField.VALUE)
+		self.consumer.assign([self.topic_partition])
+
+		# Se usa para almacenar el indice local del primer mensaje leído por el consumidor de Kafka
+		self.indice_local = None
+
+	def cerrar(self):
+		''' Cierra el consumidor de Kafka '''
+
+		self.consumer.close()
+
+	def reasignar(self):
+		''' Reasigna el consumidor de Kafka si se ha perdido la conexión '''
+
+		try:
+			# Esta función devuelve un 'RuntimeError' si el consumidor de Kafka se ha desconectado
+			self.consumer.consumer_group_metadata()
+
+		except RuntimeError as e:
+			# Basta con crear un nuevo consumidor de Kafka para reasignarlo
+			self.consumer = Consumer(self.consumer_settings)
+			self.consumer.assign([self.topic_partition])
+
+	def mover_indice_remoto(self, indice: Optional[int]):
+		''' Posiciona el consumidor de Kafka en el indice indicado '''
+
+		# No se hace nada si el indice es nulo
+		if indice is None:
+			return
+
+		self.reasignar()
+		self.consumer.unassign()
+		self.topic_partition.offset = indice
+		self.consumer.commit(offsets=[self.topic_partition])
+		self.consumer.assign([self.topic_partition])
+
+	def indice_mensaje_remoto(self) -> Optional[int]:
+		'''
+			Obtiene el indice del ultimo mensaje leído por consumidor de Kafka
+			- Si no se ha leído ningún mensaje, devuelve nulo
+		'''
+
+		self.reasignar()
+		indice = self.consumer.position([self.topic_partition])[0].offset
+		return indice if indice > 0 else None
+
+	def regresar_indice_remoto(self):
+		'''
+			Regresa el indice remoto del consumidor para que coincida con el indice local
+			- Se usa para revertir el flujo de datos al punto de partida (indice local)
+		'''
+
+		self.mover_indice_remoto(self.indice_local)
+
+	def avanzar_indice_local(self):
+		'''
+			Avanza el indice local del consumidor para que coincida con el indice remoto
+			- Se usa para actualizar el punto de partida (indice local) del flujo de datos
+		'''
+
+		self.indice_local = self.indice_mensaje_remoto()
+
+	def obtener_registro(self) -> Optional[Any]:
+		''' Obtiene el siguiente registro desde el consumidor de Kafka '''
+
+		self.reasignar()
+
+		# Se lee el siguiente mensaje del consumidor de Kafka
+		mensaje = self.consumer.poll(1)
+
+		# Se regresa nulo si no se ha leído ningún mensaje o si el mensaje contiene un error
+		if mensaje is None or mensaje.error() is not None:
+			return None
+
+		# Se actualiza el indice del primer mensaje leído
+		if self.indice_local is None:
+			self.indice_local = self.indice_mensaje_remoto() - 1
+
+		# Se deserializa el mensaje y se regresa el registro
+		return self.avro_deserializer(mensaje.value(), self.serialization_context)
\ No newline at end of file

log.py

+''' Utilidades para el manejo de logs '''
+
+import logging
+
+# Niveles de logs
+INFO = logging.INFO
+ERROR = logging.ERROR
+
+class LogHandle:
+	''' Permite redirigir la salida estándar y de error a un logger '''
+
+	def __init__(self, level: int):
+		self.level = level
+		self.message = ''
+
+	def write(self, message: str):
+		self.message = self.message + message
+
+		while '\n' in self.message:
+			index = self.message.index('\n')
+			logging.log(self.level, self.message[:index])
+			self.message = self.message[(index + 1):]
+
+	def flush(self):
+		pass
+
+def configurar_logs():
+	''' Configura los logs con el formato deseado '''
+
+	# Se configura el logger
+	logging.basicConfig \
+	(
+		format = '[%(asctime)-19s] [%(levelname)-8s]: %(message)s',
+		datefmt= '%d/%m/%Y %H:%M:%S',
+		level= logging.DEBUG
+	)
+
+	# Se desactivan los logs para la librería 'pid'
+	logging.getLogger('PidFile').setLevel(logging.WARNING)
+
+	# Se desactivan los logs de la librería 'requests'
+	logging.getLogger('urllib3').setLevel(logging.WARNING)
+
+def error(mensaje: str):
+	''' Muestra un mensaje de error '''
+
+	logging.error(mensaje)
\ No newline at end of file

modelos.py

+''' Contiene las clases de conexión y los modelos asociados a las bases de datos '''
+
+import pandas as pd
+from enum import Enum
+from pathlib import Path
+from typing import Optional
+
+# SQLAlchemy
+import sqlalchemy as sa
+import sqlalchemy.orm as orm
+from sqlalchemy.pool import NullPool
+
+# Ruta del archivo actual
+RUTA = Path(__file__).resolve().parent
+
+# Clases base de SQLAlchemy
+SQLite = orm.declarative_base()
+PostgreSQL = orm.declarative_base()
+
+# Diccionario de datos para estados y municipios
+HOJA_ENTIDADES = 'Catálogo de ENTIDADES'
+HOJA_MUNICIPIOS = 'Catálogo MUNICIPIOS'
+ARCHIVO_DICCIONARIO = RUTA / 'datos/201128 Catalogos.xlsx'
+
+# ---------------------- Conexiones a las bases de datos ----------------------
+
+class Conexion:
+	''' Representa una conexión genérica a una base de datos '''
+
+	def __init__(self, cadena_conexion: str, metadata: sa.MetaData):
+		# Se conecta a la base de datos y se crea la sesión de SQLAlchemy
+		self.motor_sql = sa.create_engine(cadena_conexion, echo = False, poolclass = NullPool)
+		self.sesion = orm.Session(bind = self.motor_sql, autoflush = False)
+
+		# Inicializa la base de datos correspondiente con los modelos de SQLAlchemy
+		metadata.create_all(self.motor_sql)
+
+	@property
+	def enlace(self):
+		''' Regresa el enlace de la conexión de la transacción actual '''
+
+		return self.sesion.connection()
+
+	def cerrar(self):
+		''' Cierra la sesión de SQLAlchemy '''
+
+		self.sesion.close()
+
+class ConexionSQLite(Conexion):
+	''' Representa una conexión a una base de datos de SQLite '''
+
+	def __init__ \
+	(
+		self,
+		base_de_datos: str
+	):
+		# Se crea la cadena de conexión a la base de datos de SQLite y se hace la conexión
+		cadena_conexion = f'sqlite:///{RUTA / base_de_datos}'
+		super().__init__(cadena_conexion, SQLite.metadata)
+
+class ConexionPostgreSQL(Conexion):
+	''' Representa una conexión a una base de datos de PostgreSQL '''
+
+	def __init__ \
+	(
+		self,
+		host: str,
+		puerto: int,
+		usuario: str,
+		clave: str,
+		base_de_datos: str
+	):
+		# Se crea la cadena de conexión a la base de datos de PostgreSQL y se hace la conexión
+		cadena_conexion = f'postgresql://{usuario}:{clave}@{host}:{puerto}/{base_de_datos}'
+		super().__init__(cadena_conexion, PostgreSQL.metadata)
+
+# ------------------------------ Tipos de datos -------------------------------
+
+class TIPO_VALOR(Enum):
+	'''
+	Enumeración de los tipos de valores
+	- CONFIRMADO
+	- FALLECIDO
+	'''
+
+	CONFIRMADO = 'FECHA_SINTOMAS'
+	FALLECIDO = 'FECHA_DEF'
+
+	def __str__(self):
+		return self.name
+
+	@classmethod
+	def obtener_datos(cls):
+		''' Genera una lista de diccionarios para su inserción en la base de datos '''
+
+		return [{'tipo' : str(tipo)} for tipo in cls]
+
+class GRUPO_EDAD(Enum):
+	'''
+	Enumeración de los grupos de edad
+	- GRUPO_18a29
+	- GRUPO_30a39
+	- GRUPO_40a49
+	- GRUPO_50a59
+	- GRUPO_60ymas
+	'''
+
+	GRUPO_18a29 = {'edad_inicial': 18, 'edad_final': 29}
+	GRUPO_30a39 = {'edad_inicial': 30, 'edad_final': 39}
+	GRUPO_40a49 = {'edad_inicial': 40, 'edad_final': 49}
+	GRUPO_50a59 = {'edad_inicial': 50, 'edad_final': 59}
+	GRUPO_60ymas = {'edad_inicial': 60, 'edad_final': None}
+
+	def __str__(self):
+		return self.name.replace('GRUPO_', '')
+
+	@classmethod
+	def obtener_datos(cls):
+		''' Genera una lista de diccionarios para su inserción en la base de datos '''
+
+		for grupo in cls:
+			grupo.value['nombre'] = str(grupo)
+		return [grupo.value for grupo in cls]
+
+# ---------------------------- Modelos de PostgreSQL --------------------------
+
+class Estado(PostgreSQL):
+	__tablename__ = 'Estado'
+
+	# Atributos
+	id = sa.Column(sa.Integer, primary_key = True)
+	id_dge = sa.Column(sa.Integer, nullable = False)
+	nombre = sa.Column(sa.String(50), nullable = True)
+
+	# Relaciones
+	municipios = orm.relationship('Municipio', back_populates = 'estado')
+
+class Municipio(PostgreSQL):
+	__tablename__ = 'Municipio'
+
+	# Atributos
+	id = sa.Column(sa.Integer, primary_key = True)
+	id_dge = sa.Column(sa.Integer, nullable = False)
+	nombre = sa.Column(sa.String(100), nullable = True)
+	fk_estado = sa.Column(sa.Integer, sa.ForeignKey('Estado.id'), nullable = False)
+
+	# Relaciones
+	estado = orm.relationship('Estado', back_populates = 'municipios')
+	detalle_valores = orm.relationship('DetalleValor', back_populates = 'municipio')
+
+class GrupoEdad(PostgreSQL):
+	__tablename__ = 'GrupoEdad'
+
+	# Atributos
+	id = sa.Column(sa.Integer, primary_key = True)
+	nombre = sa.Column(sa.String(100), nullable = False)
+	edad_inicial = sa.Column(sa.Integer, nullable = True)
+	edad_final = sa.Column(sa.Integer, nullable = True)
+
+	# Relaciones
+	detalle_valores = orm.relationship('DetalleValor', back_populates = 'grupo_edad')
+
+class TipoValor(PostgreSQL):
+	__tablename__ = 'TipoValor'
+
+	# Atributos
+	id = sa.Column(sa.Integer, primary_key = True)
+	tipo = sa.Column(sa.String(100), nullable = False)
+
+	# Relaciones
+	detalle_valores = orm.relationship('DetalleValor', back_populates = 'tipo_valor')
+
+class DetalleValor(PostgreSQL):
+	__tablename__ = 'DetalleValor'
+
+	# Atributos
+	id = sa.Column(sa.Integer, primary_key = True)
+	fecha = sa.Column(sa.Date, nullable = False)
+	fk_municipio = sa.Column(sa.Integer, sa.ForeignKey('Municipio.id'), nullable = False)
+	fk_grupo_edad = sa.Column(sa.Integer, sa.ForeignKey('GrupoEdad.id'), nullable = False)
+	fk_tipo_valor = sa.Column(sa.Integer, sa.ForeignKey('TipoValor.id'), nullable = False)
+
+	# Relaciones
+	municipio = orm.relationship('Municipio', back_populates = 'detalle_valores')
+	grupo_edad = orm.relationship('GrupoEdad', back_populates = 'detalle_valores')
+	tipo_valor = orm.relationship('TipoValor', back_populates = 'detalle_valores')
+	valor_real = orm.relationship('ValorReal', back_populates = 'detalle_valor', uselist = False)
+	valor_predicho = orm.relationship('ValorPredicho', back_populates = 'detalle_valor', uselist = False)
+
+class ValorReal(PostgreSQL):
+	__tablename__ = 'ValorReal'
+
+	# Atributos
+	id = sa.Column(sa.Integer, primary_key = True)
+	valor = sa.Column(sa.Integer, nullable = False)
+	fk_detalle_valor = sa.Column(sa.Integer, sa.ForeignKey('DetalleValor.id'), nullable = False)
+
+	# Relaciones
+	detalle_valor = orm.relationship('DetalleValor', back_populates = 'valor_real')
+
+class ValorPredicho(PostgreSQL):
+	__tablename__ = 'ValorPredicho'
+
+	# Atributos
+	id = sa.Column(sa.Integer, primary_key = True)
+	valor = sa.Column(sa.Float, nullable = False)
+	fk_detalle_valor = sa.Column(sa.Integer, sa.ForeignKey('DetalleValor.id'), nullable = False)
+
+	# Relaciones
+	detalle_valor = orm.relationship('DetalleValor', back_populates = 'valor_predicho')
+
+# ------------------------------ Modelos de SQLite ----------------------------
+
+class RegistroLocal(SQLite):
+	'''
+		Modelo que representa un registro de COVID-19 en la base de datos local (SQLite)
+		- Unicamente se guardan los registros con casos confirmados de COVID-19
+	'''
+
+	__tablename__ = 'RegistroLocal'
+
+	# Atributos
+	ID_REGISTRO = sa.Column(sa.String, primary_key = True)
+	ENTIDAD_RES = sa.Column(sa.Integer, nullable = False)
+	MUNICIPIO_RES = sa.Column(sa.Integer, nullable = False)
+	FECHA_SINTOMAS = sa.Column(sa.Date, nullable = False)
+	FECHA_DEF = sa.Column(sa.Date, nullable = True)
+	GRUPO_EDAD = sa.Column(sa.String, nullable = True)
+
+# ------------------------------ Modelos de Kafka -----------------------------
+
+class RegistroKafka:
+	''' Modelo que representa un registro del tópico de Kafka de casos de COVID-19 '''
+
+	FECHA_ACTUALIZACION: Optional[str] = None
+	ID_REGISTRO: Optional[str] = None
+	ORIGEN: Optional[int] = None
+	SECTOR: Optional[int] = None
+	ENTIDAD_UM: Optional[int] = None
+	SEXO: Optional[int] = None
+	ENTIDAD_NAC: Optional[int] = None
+	ENTIDAD_RES: Optional[int] = None
+	MUNICIPIO_RES: Optional[int] = None
+	TIPO_PACIENTE: Optional[int] = None
+	FECHA_INGRESO: Optional[str] = None
+	FECHA_SINTOMAS: Optional[str] = None
+	FECHA_DEF: Optional[str] = None
+	INTUBADO: Optional[int] = None
+	NEUMONIA: Optional[int] = None
+	EDAD: Optional[int] = None
+	NACIONALIDAD: Optional[int] = None
+	EMBARAZO: Optional[int] = None
+	HABLA_LENGUA_INDIG: Optional[int] = None
+	INDIGENA: Optional[int] = None
+	DIABETES: Optional[int] = None
+	EPOC: Optional[int] = None
+	ASMA: Optional[int] = None
+	INMUSUPR: Optional[int] = None
+	HIPERTENSION: Optional[int] = None
+	OTRA_COM: Optional[int] = None
+	CARDIOVASCULAR: Optional[int] = None
+	OBESIDAD: Optional[int] = None
+	RENAL_CRONICA: Optional[int] = None
+	TABAQUISMO: Optional[int] = None
+	OTRO_CASO: Optional[int] = None
+	TOMA_MUESTRA_LAB: Optional[int] = None
+	RESULTADO_LAB: Optional[int] = None
+	TOMA_MUESTRA_ANTIGENO: Optional[int] = None
+	RESULTADO_ANTIGENO: Optional[int] = None
+	CLASIFICACION_FINAL: Optional[int] = None
+	MIGRANTE: Optional[int] = None
+	PAIS_NACIONALIDAD: Optional[str] = None
+	PAIS_ORIGEN: Optional[str] = None
+	UCI: Optional[int] = None
+	TYPE: Optional[str] = None
+
+	def __init__(self, **kwargs):
+		for campo, valor in kwargs.items():
+			setattr(self, campo, valor)
+
+	@property
+	def fecha_actualizacion(self):
+		return pd.Timestamp(self.FECHA_ACTUALIZACION)
+
+	@staticmethod
+	def desde_diccionario(diccionario: dict, _):
+		''' Método requerido por Kafka para poder deserializar un objeto de este tipo '''
+
+		if diccionario is None:
+			return None
+		return RegistroKafka(**diccionario)
+
+# -------------------- Inicialización de la base de datos ---------------------
+
+def cargar_tipos_valor(conexion: Conexion):
+	''' Carga los tipos de valor en la base de datos '''
+
+	conexion.sesion.execute(sa.insert(TipoValor), TIPO_VALOR.obtener_datos())
+
+def cargar_grupos_edad(conexion: Conexion):
+	''' Carga los grupos de edad en la base de datos '''
+
+	conexion.sesion.execute(sa.insert(GrupoEdad), GRUPO_EDAD.obtener_datos())
+
+def cargar_estados(conexion: Conexion):
+	''' Carga los estados desde el diccionario de datos en la base de datos '''
+
+	# Se cargan los estados desde el diccionario de datos de la DGE
+	estados = pd.read_excel(ARCHIVO_DICCIONARIO, sheet_name = HOJA_ENTIDADES)
+	estados.rename(columns = {'CLAVE_ENTIDAD': 'id_dge'}, inplace = True)
+	estados['nombre'] = estados['ENTIDAD_FEDERATIVA'].str.title()
+
+	# Se ordenan los estados por su clave de la DGE
+	estados.sort_values(by = 'id_dge', inplace = True)
+
+	# Se actualizan los estados en la base de datos
+	conexion.sesion.execute(sa.insert(Estado), estados.to_dict(orient = 'records'))
+
+def cargar_municipios(conexion: Conexion):
+	''' Carga los municipios desde el diccionario de datos en la base de datos '''
+
+	# Se cargan los municipios desde el diccionario de datos de la DGE
+	municipios = pd.read_excel(ARCHIVO_DICCIONARIO, sheet_name = HOJA_MUNICIPIOS)
+	municipios.rename(columns = {'CLAVE_MUNICIPIO': 'id_dge'}, inplace = True)
+	municipios['nombre'] = municipios['MUNICIPIO'].str.title()
+
+	# Se obtienen las llaves foráneas de los estados
+	estados = pd.read_sql(conexion.sesion.query(Estado).statement, conexion.enlace)
+	estados = estados[['id', 'id_dge']]
+	estados.columns = ['fk_estado', 'CLAVE_ENTIDAD']
+	municipios = pd.merge(municipios, estados, on = 'CLAVE_ENTIDAD', how = 'left')
+
+	# Se ordenan los municipios por clave de la DGE
+	municipios.sort_values(by = ['fk_estado', 'id_dge'], inplace = True)
+
+	# Se actualizan los municipios en la base de datos
+	conexion.sesion.execute(sa.insert(Municipio), municipios.to_dict(orient = 'records'))
+
+def inicializar_base_remota(conexion: Conexion):
+	''' Inicializa la base de datos remota con los datos iniciales '''
+
+	if conexion.sesion.query(TipoValor).count() == 0:
+		cargar_tipos_valor(conexion)
+		cargar_grupos_edad(conexion)
+		cargar_estados(conexion)
+		cargar_municipios(conexion)
+		conexion.sesion.commit()
\ No newline at end of file

requirements.txt

+pandas==2.0.2
+fastavro==1.7.4
+openpyxl==3.1.2
+SQLAlchemy==2.0.16
+GeoAlchemy2==0.13.3
+psycopg2-binary==2.9.6
+confluent-kafka==2.1.1
+python-daemon==3.0.1
+pid==3.0.4
\ No newline at end of file