274 lines
10 KiB
Python
274 lines
10 KiB
Python
# Agregar al inicio del archivo, después de los imports existentes
|
|
import cv2
|
|
from pyzbar import pyzbar
|
|
|
|
import os
|
|
import json
|
|
from datetime import datetime
|
|
import requests # Para webhooks si se necesitan
|
|
import time
|
|
import logging
|
|
import sys
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple, List
|
|
from dataclasses import dataclass
|
|
from contextlib import contextmanager
|
|
from src.db_handler import DatabaseHandler, DatabaseError
|
|
|
|
@dataclass
|
|
class QRData:
|
|
"""Clase para almacenar información del código QR"""
|
|
data: str
|
|
code_type: str
|
|
coordinates: Tuple[int, int, int, int]
|
|
timestamp: float
|
|
|
|
class QRDetectorError(Exception):
|
|
"""Clase base para excepciones del detector QR"""
|
|
pass
|
|
|
|
class CameraError(QRDetectorError):
|
|
"""Error relacionado con la cámara"""
|
|
pass
|
|
|
|
class QRDecodeError(QRDetectorError):
|
|
"""Error en la decodificación del QR"""
|
|
pass
|
|
|
|
class ConfigError(QRDetectorError):
|
|
"""Error en la configuración"""
|
|
pass
|
|
|
|
class QRDetector:
|
|
def __init__(self, config_path: str = 'config.yaml'):
|
|
self.config = self.load_config(config_path)
|
|
self.setup_logging()
|
|
self.camera_id = self.config['camera']['id']
|
|
self.min_delay = self.config['camera']['min_delay']
|
|
self.camera = None
|
|
self.last_code = None
|
|
self.last_time = 0
|
|
|
|
# Inicializar base de datos si está habilitada
|
|
if self.config['database']['enabled']:
|
|
try:
|
|
self.db = DatabaseHandler(self.config['database'])
|
|
self.logger.info("Base de datos inicializada correctamente")
|
|
except DatabaseError as e:
|
|
self.logger.error(f"Error inicializando base de datos: {e}")
|
|
raise
|
|
|
|
def load_config(self, config_path: str) -> dict:
|
|
"""Carga la configuración desde el archivo YAML"""
|
|
try:
|
|
with open(config_path, 'r') as file:
|
|
return yaml.safe_load(file)
|
|
except Exception as e:
|
|
raise ConfigError(f"Error cargando configuración: {e}")
|
|
|
|
def setup_logging(self):
|
|
"""Configura el sistema de logging basado en la configuración"""
|
|
log_config = self.config['logging']
|
|
handlers = []
|
|
|
|
if log_config['output']['console']:
|
|
handlers.append(logging.StreamHandler(sys.stdout))
|
|
|
|
if log_config['output']['file']['enabled']:
|
|
handlers.append(logging.FileHandler(log_config['output']['file']['path']))
|
|
|
|
logging.basicConfig(
|
|
level=getattr(logging, log_config['level']),
|
|
format=log_config['format'],
|
|
handlers=handlers
|
|
)
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
@contextmanager
|
|
def initialize_camera(self):
|
|
"""Inicializa la cámara de manera segura usando context manager"""
|
|
try:
|
|
self.camera = cv2.VideoCapture(self.camera_id)
|
|
if not self.camera.isOpened():
|
|
raise CameraError("No se pudo abrir la cámara")
|
|
yield self.camera
|
|
finally:
|
|
if self.camera:
|
|
self.camera.release()
|
|
cv2.destroyAllWindows()
|
|
self.logger.info("Recursos de la cámara liberados")
|
|
|
|
def process_frame(self, frame) -> List[QRData]:
|
|
"""Procesa un frame y retorna la lista de códigos QR detectados"""
|
|
try:
|
|
qr_codes = pyzbar.decode(frame)
|
|
current_time = time.time()
|
|
results = []
|
|
|
|
for qr_code in qr_codes:
|
|
try:
|
|
data = qr_code.data.decode('utf-8')
|
|
code_type = qr_code.type
|
|
coordinates = qr_code.rect
|
|
|
|
if (data != self.last_code or
|
|
(current_time - self.last_time) > self.min_delay):
|
|
|
|
qr_data = QRData(
|
|
data=data,
|
|
code_type=code_type,
|
|
coordinates=coordinates,
|
|
timestamp=current_time
|
|
)
|
|
results.append(qr_data)
|
|
|
|
self.last_code = data
|
|
self.last_time = current_time
|
|
|
|
# Guardar en base de datos si está habilitada
|
|
if self.config['database']['enabled']:
|
|
try:
|
|
qr_dict = {
|
|
'data': data,
|
|
'code_type': code_type,
|
|
'coordinates': coordinates,
|
|
'timestamp': current_time
|
|
}
|
|
self.db.save_qr_code(qr_dict)
|
|
scan_count = self.db.get_scan_count(data)
|
|
self.logger.info(f"QR guardado en DB. Total escaneos: {scan_count}")
|
|
except DatabaseError as e:
|
|
self.logger.error(f"Error guardando en base de datos: {e}")
|
|
|
|
except UnicodeDecodeError as e:
|
|
self.logger.error(f"Error decodificando QR: {e}")
|
|
raise QRDecodeError("Error al decodificar el contenido del QR")
|
|
|
|
return results
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error procesando frame: {e}")
|
|
raise QRDetectorError(f"Error en el procesamiento del frame: {e}")
|
|
|
|
def draw_qr_info(self, frame, qr_data: QRData):
|
|
"""Dibuja la información del QR en el frame con efectos visuales mejorados"""
|
|
try:
|
|
display_config = self.config['display']
|
|
x, y, w, h = qr_data.coordinates
|
|
|
|
# Dibujar rectángulo principal
|
|
cv2.rectangle(
|
|
frame,
|
|
(x, y),
|
|
(x + w, y + h),
|
|
tuple(display_config['rectangle']['color']),
|
|
display_config['rectangle']['thickness']
|
|
)
|
|
|
|
# Dibujar esquinas del QR
|
|
l = 30 # longitud de las líneas de las esquinas
|
|
# Esquina superior izquierda
|
|
cv2.line(frame, (x, y), (x + l, y), (0, 255, 255), 2)
|
|
cv2.line(frame, (x, y), (x, y + l), (0, 255, 255), 2)
|
|
# Esquina superior derecha
|
|
cv2.line(frame, (x + w, y), (x + w - l, y), (0, 255, 255), 2)
|
|
cv2.line(frame, (x + w, y), (x + w, y + l), (0, 255, 255), 2)
|
|
# Esquina inferior izquierda
|
|
cv2.line(frame, (x, y + h), (x + l, y + h), (0, 255, 255), 2)
|
|
cv2.line(frame, (x, y + h), (x, y + h - l), (0, 255, 255), 2)
|
|
# Esquina inferior derecha
|
|
cv2.line(frame, (x + w, y + h), (x + w - l, y + h), (0, 255, 255), 2)
|
|
cv2.line(frame, (x + w, y + h), (x + w, y + h - l), (0, 255, 255), 2)
|
|
|
|
# Crear fondo semi-transparente para el texto
|
|
overlay = frame.copy()
|
|
cv2.rectangle(
|
|
overlay,
|
|
(x, y - 40),
|
|
(x + w, y),
|
|
(0, 0, 0),
|
|
-1
|
|
)
|
|
frame = cv2.addWeighted(overlay, 0.5, frame, 0.5, 0)
|
|
|
|
# Mostrar información
|
|
display_text = f"{qr_data.code_type}: {qr_data.data}"
|
|
cv2.putText(
|
|
frame,
|
|
display_text,
|
|
(x + 5, y - 15),
|
|
cv2.FONT_HERSHEY_SIMPLEX,
|
|
0.6,
|
|
(255, 255, 255), # Color blanco para mejor contraste
|
|
2
|
|
)
|
|
|
|
# Contador de escaneos si está disponible
|
|
if self.config['database']['enabled']:
|
|
try:
|
|
scan_count = self.db.get_scan_count(qr_data.data)
|
|
count_text = f"Escaneos: {scan_count}"
|
|
cv2.putText(
|
|
frame,
|
|
count_text,
|
|
(x + w - 100, y + h + 25),
|
|
cv2.FONT_HERSHEY_SIMPLEX,
|
|
0.5,
|
|
(0, 255, 255),
|
|
1
|
|
)
|
|
except DatabaseError:
|
|
pass
|
|
|
|
self.logger.info(f"QR detectado - Tipo: {qr_data.code_type}, Datos: {qr_data.data}")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error dibujando información del QR: {e}")
|
|
raise QRDetectorError(f"Error al dibujar información: {e}")
|
|
|
|
def run(self):
|
|
"""Método principal para ejecutar el detector"""
|
|
self.logger.info("Iniciando detector de códigos QR...")
|
|
|
|
with self.initialize_camera() as camera:
|
|
while True:
|
|
try:
|
|
success, frame = camera.read()
|
|
if not success:
|
|
raise CameraError("Error capturando frame")
|
|
|
|
qr_codes = self.process_frame(frame)
|
|
|
|
for qr_data in qr_codes:
|
|
self.draw_qr_info(frame, qr_data)
|
|
|
|
cv2.imshow(self.config['display']['window_name'], frame)
|
|
|
|
if cv2.waitKey(1) & 0xFF == ord('q'):
|
|
self.logger.info("Cerrando aplicación...")
|
|
break
|
|
|
|
except CameraError as e:
|
|
self.logger.error(f"Error de cámara: {e}")
|
|
break
|
|
except QRDecodeError as e:
|
|
self.logger.error(f"Error de decodificación: {e}")
|
|
continue
|
|
except Exception as e:
|
|
self.logger.error(f"Error inesperado: {e}")
|
|
break
|
|
|
|
def main():
|
|
try:
|
|
detector = QRDetector()
|
|
detector.run()
|
|
except ConfigError as e:
|
|
print(f"Error en la configuración: {e}")
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Error crítico en la aplicación: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main() |