🧠 SISTEMA DE GESTIÓN ENERGÉTICA INTELIGENTE (SGEI) PARA PASAIA LAB
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
SGEI - Sistema de Gestión Energética Inteligente para PASAIA LAB
Autor: José Agustín Fontán Varela (PASAIA LAB / INTELIGENCIA LIBRE)
Colaboración: DeepSeek
Licencia: GPL v3
"""
import hashlib
import time
import json
import random
import threading
import numpy as np
import matplotlib.pyplot as plt
from collections import deque
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
# ============================================================
# 1. CONFIGURACIÓN DEL SISTEMA
# ============================================================
@dataclass
class Configuracion:
"""Configuración de los equipos de PASAIA LAB"""
# Paneles solares
PANELES_PRINCIPALES_W = 380
PANELES_PORTATILES_W = 60
HORAS_SOL_PICO = 5
# Estaciones de energía (capacidad en Wh)
ESTACION_600W = 600
ESTACION_300W = 300
ESTACION_200W = 200
CANTIDAD_600W = 2
CANTIDAD_300W = 2
CANTIDAD_200W = 1
# Baterías de plomo (12V 7Ah)
PLOMO_12V_7AH = 12 * 7 # 84 Wh por batería
CANTIDAD_PLOMO = 8
# Consumos estimados (W)
PC_SOBREMESA = 125
MONITOR_PC = 30
RASPBERRY_PI_5 = 12
MONITOR_12V = 15
ROUTER = 10
SAI = 5
# Prioridades de carga
PRIORIDADES = {
"critico": ["router", "raspberry_1", "monitor_12v_1"],
"alto": ["raspberry_2", "monitor_12v_2", "sai"],
"medio": ["pc_1", "monitor_pc_1"],
"bajo": ["pc_2", "monitor_pc_2"]
}
# ============================================================
# 2. BLOQUE DE LA BLOCKCHAIN
# ============================================================
class Bloque:
"""Bloque de la blockchain de gestión energética"""
def __init__(self, index: int, timestamp: float, datos: Dict, hash_anterior: str):
self.index = index
self.timestamp = timestamp
self.datos = datos
self.hash_anterior = hash_anterior
self.nonce = 0
self.hash = self.calcular_hash()
def calcular_hash(self) -> str:
contenido = json.dumps({
"index": self.index,
"timestamp": self.timestamp,
"datos": self.datos,
"hash_anterior": self.hash_anterior,
"nonce": self.nonce
}, sort_keys=True)
return hashlib.sha256(contenido.encode()).hexdigest()
def minar(self, dificultad: int = 2):
"""Prueba de trabajo simple"""
while not self.hash.startswith('0' * dificultad):
self.nonce += 1
self.hash = self.calcular_hash()
class BlockchainEnergia:
"""Blockchain para el registro inmutable de eventos energéticos"""
def __init__(self):
self.cadena = [self._crear_bloque_genesis()]
self.dificultad = 2
self.eventos_pendientes = []
def _crear_bloque_genesis(self) -> Bloque:
datos = {"evento": "Inicio del sistema", "timestamp": time.time()}
return Bloque(0, time.time(), datos, "0")
def agregar_bloque(self, datos: Dict) -> Bloque:
bloque = Bloque(
len(self.cadena),
time.time(),
datos,
self.cadena[-1].hash
)
bloque.minar(self.dificultad)
self.cadena.append(bloque)
return bloque
def registrar_evento(self, evento: str, datos: Dict) -> None:
"""Registra un evento en la blockchain"""
datos_completos = {
"evento": evento,
"timestamp": time.time(),
"datos": datos
}
self.agregar_bloque(datos_completos)
def verificar_integridad(self) -> bool:
for i in range(1, len(self.cadena)):
if self.cadena[i].hash_anterior != self.cadena[i-1].hash:
return False
if self.cadena[i].hash != self.cadena[i].calcular_hash():
return False
return True
def obtener_historial(self) -> List[Dict]:
return [{
"index": b.index,
"timestamp": b.timestamp,
"datos": b.datos,
"hash": b.hash[:8]
} for b in self.cadena]
# ============================================================
# 3. SISTEMA DE MONITORIZACIÓN DE RECURSOS
# ============================================================
class MonitorRecursos:
"""Monitoriza generación, consumo y almacenamiento"""
def __init__(self, config: Configuracion):
self.config = config
self.historial_generacion = deque(maxlen=100)
self.historial_consumo = deque(maxlen=100)
self.historial_bateria = deque(maxlen=100)
self.estado_actual = {}
self._simular_estado_inicial()
def _simular_estado_inicial(self):
"""Estado inicial simulado (en producción: lecturas de sensores)"""
self.estado_actual = {
"generacion_solar": 0.0,
"bateria_total_wh": self._calcular_capacidad_total(),
"bateria_disponible_wh": self._calcular_capacidad_total() * 0.85,
"consumo_actual": 0.0,
"equipos_activos": [],
"estado_baterias": "ok"
}
def _calcular_capacidad_total(self) -> float:
"""Calcula la capacidad total de almacenamiento en Wh"""
total = 0
total += self.config.ESTACION_600W * self.config.CANTIDAD_600W
total += self.config.ESTACION_300W * self.config.CANTIDAD_300W
total += self.config.ESTACION_200W * self.config.CANTIDAD_200W
total += self.config.PLOMO_12V_7AH * self.config.CANTIDAD_PLOMO
return total
def actualizar(self):
"""Simula la actualización de los datos de recursos"""
# Simulación de generación solar (depende de la hora del día)
hora = time.localtime().tm_hour
if 6 < hora < 20: # horas de sol
factor_solar = max(0, 1 - abs(hora - 13) / 7)
generacion = (self.config.PANELES_PRINCIPALES_W + self.config.PANELES_PORTATILES_W) * factor_solar * random.uniform(0.7, 1.0)
else:
generacion = 0
# Simulación de consumo (varía según equipos activos)
consumo = self._estimar_consumo()
# Simulación de descarga de batería
bateria_actual = self.estado_actual["bateria_disponible_wh"]
delta_energia = generacion - consumo
nueva_bateria = max(0, min(self._calcular_capacidad_total(), bateria_actual + delta_energia * 0.01))
# Actualizar estado
self.estado_actual.update({
"generacion_solar": generacion,
"consumo_actual": consumo,
"bateria_disponible_wh": nueva_bateria,
"timestamp": time.time()
})
self.historial_generacion.append(generacion)
self.historial_consumo.append(consumo)
self.historial_bateria.append(nueva_bateria)
return self.estado_actual
def _estimar_consumo(self) -> float:
"""Estima el consumo según equipos activos"""
consumo = 0
# Simulación: encender equipos según prioridad y batería
if self.estado_actual["bateria_disponible_wh"] > 200:
consumo += self.config.RASPBERRY_PI_5 * 2 # dos Raspberry
consumo += self.config.MONITOR_12V * 2
consumo += self.config.ROUTER
if self.estado_actual["bateria_disponible_wh"] > 500:
consumo += self.config.PC_SOBREMESA
consumo += self.config.MONITOR_PC
if self.estado_actual["bateria_disponible_wh"] > 1000:
consumo += self.config.PC_SOBREMESA # segundo PC
consumo += self.config.MONITOR_PC
return consumo + random.uniform(-10, 10)
def obtener_estado(self) -> Dict:
return {
"generacion_solar": self.estado_actual["generacion_solar"],
"bateria_total_wh": self.estado_actual["bateria_total_wh"],
"bateria_disponible_wh": self.estado_actual["bateria_disponible_wh"],
"consumo_actual": self.estado_actual["consumo_actual"],
"autonomia_horas": self._calcular_autonomia()
}
def _calcular_autonomia(self) -> float:
if self.estado_actual["consumo_actual"] > 0:
return self.estado_actual["bateria_disponible_wh"] / self.estado_actual["consumo_actual"]
return float('inf')
# ============================================================
# 4. IA DE OPTIMIZACIÓN (RED NEURONAL)
# ============================================================
class IAOptimizacion:
"""
Red neuronal simple para optimizar el consumo energético.
En producción usaría TensorFlow/Keras, aquí es una versión simulada.
"""
def __init__(self):
self.modelo_entrenado = False
self.pesos = np.random.randn(4, 1) * 0.1
self.historial_entrenamiento = []
def predecir(self, estado: Dict) -> Dict:
"""
Predice el consumo óptimo y genera recomendaciones
"""
# Extraer características del estado
generacion = estado["generacion_solar"]
bateria = estado["bateria_disponible_wh"]
capacidad_total = estado["bateria_total_wh"]
consumo = estado.get("consumo_actual", 0)
# Cálculo de recomendaciones (simulación)
recomendaciones = []
if bateria < 200 and generacion < 50:
recomendaciones.append("⚠️ Batería crítica. Apagar equipos no esenciales.")
elif bateria < 500 and generacion < 100:
recomendaciones.append("⚠️ Batería baja. Reducir consumo.")
elif bateria > 1500 and generacion > 200:
recomendaciones.append("✅ Excedente de energía. Puedes cargar dispositivos adicionales.")
if consumo > 300 and bateria < 1000:
recomendaciones.append("⚠️ Consumo elevado para el nivel de batería actual.")
# Predicción de tiempo de autonomía
if consumo > 0:
horas_autonomia = bateria / consumo
if horas_autonomia < 2:
recomendaciones.append(f"⚠️ Autonomía crítica: {horas_autonomia:.1f} horas.")
elif horas_autonomia < 6:
recomendaciones.append(f"ℹ️ Autonomía limitada: {horas_autonomia:.1f} horas.")
else:
recomendaciones.append(f"✅ Autonomía suficiente: {horas_autonomia:.1f} horas.")
# Recomendaciones de prioridad
if bateria > 1000:
prioridad = "alta"
recomendaciones.append("✅ Estado óptimo. Puedes operar con todos los equipos.")
elif bateria > 500:
prioridad = "media"
recomendaciones.append("⚡ Modo de eficiencia. Priorizar equipos esenciales.")
else:
prioridad = "baja"
recomendaciones.append("🔋 Modo crítico. Solo equipos de comunicación.")
return {
"prioridad": prioridad,
"recomendaciones": recomendaciones,
"horas_autonomia": horas_autonomia if consumo > 0 else float('inf')
}
def entrenar(self, datos_historicos: List):
"""Simula el entrenamiento de la red neuronal"""
# En producción, se usaría TensorFlow con datos reales
self.modelo_entrenado = True
self.historial_entrenamiento = datos_historicos
return {"status": "entrenado", "datos": len(datos_historicos)}
# ============================================================
# 5. CONTROL DE EQUIPOS
# ============================================================
class ControlEquipos:
"""Gestiona la conmutación de equipos según prioridad y estado"""
def __init__(self, config: Configuracion):
self.config = config
self.equipos_activos = set()
self.estado_equipos = {
"router": False,
"raspberry_1": False,
"raspberry_2": False,
"monitor_12v_1": False,
"monitor_12v_2": False,
"pc_1": False,
"pc_2": False,
"monitor_pc_1": False,
"monitor_pc_2": False,
"sai": False
}
self.ultima_decision = {}
def decidir(self, estado: Dict, prioridad: str) -> Dict:
"""Toma decisiones de encendido/apagado según el estado y la prioridad"""
bateria = estado["bateria_disponible_wh"]
generacion = estado["generacion_solar"]
decisiones = {}
# Equipos críticos (siempre encendidos si hay batería > 50 Wh)
if bateria > 50 or generacion > 20:
self.estado_equipos["router"] = True
self.estado_equipos["raspberry_1"] = True
self.estado_equipos["monitor_12v_1"] = True
else:
self.estado_equipos["router"] = False
self.estado_equipos["raspberry_1"] = False
self.estado_equipos["monitor_12v_1"] = False
# Equipos de alta prioridad
if bateria > 200 or generacion > 50:
self.estado_equipos["raspberry_2"] = True
self.estado_equipos["monitor_12v_2"] = True
self.estado_equipos["sai"] = True
else:
self.estado_equipos["raspberry_2"] = False
self.estado_equipos["monitor_12v_2"] = False
self.estado_equipos["sai"] = False
# Equipos de media prioridad
if bateria > 500 or generacion > 100:
self.estado_equipos["pc_1"] = True
self.estado_equipos["monitor_pc_1"] = True
else:
self.estado_equipos["pc_1"] = False
self.estado_equipos["monitor_pc_1"] = False
# Equipos de baja prioridad
if bateria > 1000 or generacion > 200:
self.estado_equipos["pc_2"] = True
self.estado_equipos["monitor_pc_2"] = True
else:
self.estado_equipos["pc_2"] = False
self.estado_equipos["monitor_pc_2"] = False
self.ultima_decision = dict(self.estado_equipos)
# Calcular consumo total estimado según equipos activos
consumo = 0
if self.estado_equipos["router"]: consumo += self.config.ROUTER
if self.estado_equipos["raspberry_1"]: consumo += self.config.RASPBERRY_PI_5
if self.estado_equipos["raspberry_2"]: consumo += self.config.RASPBERRY_PI_5
if self.estado_equipos["monitor_12v_1"]: consumo += self.config.MONITOR_12V
if self.estado_equipos["monitor_12v_2"]: consumo += self.config.MONITOR_12V
if self.estado_equipos["pc_1"]: consumo += self.config.PC_SOBREMESA
if self.estado_equipos["pc_2"]: consumo += self.config.PC_SOBREMESA
if self.estado_equipos["monitor_pc_1"]: consumo += self.config.MONITOR_PC
if self.estado_equipos["monitor_pc_2"]: consumo += self.config.MONITOR_PC
if self.estado_equipos["sai"]: consumo += self.config.SAI
return {
"equipos_activos": [k for k, v in self.estado_equipos.items() if v],
"consumo_estimado": consumo,
"estado_equipos": dict(self.estado_equipos)
}
# ============================================================
# 6. SISTEMA PRINCIPAL
# ============================================================
class SistemaGestionEnergia:
"""Sistema central que coordina todos los módulos"""
def __init__(self):
self.config = Configuracion()
self.blockchain = BlockchainEnergia()
self.monitor = MonitorRecursos(self.config)
self.ia = IAOptimizacion()
self.control = ControlEquipos(self.config)
self.ejecutando = False
self.historial_estados = []
def iniciar(self):
"""Inicia el sistema"""
self.ejecutando = True
self.blockchain.registrar_evento("INICIO_SISTEMA", {"mensaje": "Sistema de gestión energética iniciado"})
print("⚡ SISTEMA DE GESTIÓN ENERGÉTICA INICIADO")
print(f"📊 Capacidad total de almacenamiento: {self.monitor._calcular_capacidad_total():.0f} Wh")
self._bucle_principal()
def _bucle_principal(self):
"""Bucle principal de monitorización y control"""
contador = 0
while self.ejecutando:
# 1. Actualizar estado de recursos
estado = self.monitor.actualizar()
# 2. Obtener recomendaciones de la IA
recomendaciones = self.ia.predecir(estado)
# 3. Decidir qué equipos mantener activos
decision = self.control.decidir(estado, recomendaciones["prioridad"])
# 4. Registrar en la blockchain
evento = {
"timestamp": time.time(),
"generacion": estado["generacion_solar"],
"consumo": estado["consumo_actual"],
"bateria": estado["bateria_disponible_wh"],
"autonomia": estado["autonomia_horas"],
"prioridad": recomendaciones["prioridad"],
"recomendaciones": recomendaciones["recomendaciones"],
"equipos_activos": decision["equipos_activos"],
"consumo_estimado": decision["consumo_estimado"]
}
self.blockchain.registrar_evento("ESTADO_SISTEMA", evento)
# 5. Guardar historial para la IA
self.historial_estados.append(evento)
if len(self.historial_estados) > 1000:
self.historial_estados = self.historial_estados[-500:]
# 6. Mostrar información cada 10 ciclos
contador += 1
if contador % 10 == 0:
self._mostrar_estado(estado, recomendaciones, decision)
# 7. Entrenar la IA cada 100 ciclos (simulado)
if contador % 100 == 0 and len(self.historial_estados) > 50:
self.ia.entrenar(self.historial_estados[-50:])
time.sleep(1) # ciclo de 1 segundo
def _mostrar_estado(self, estado: Dict, recomendaciones: Dict, decision: Dict):
"""Muestra el estado actual del sistema"""
print("\n" + "=" * 60)
print("📊 ESTADO DEL SISTEMA")
print("=" * 60)
print(f"☀️ Generación solar: {estado['generacion_solar']:.1f} W")
print(f"⚡ Consumo actual: {estado['consumo_actual']:.1f} W")
print(f"🔋 Batería disponible: {estado['bateria_disponible_wh']:.0f} Wh")
print(f"⏱️ Autonomía estimada: {estado['autonomia_horas']:.1f} h")
print(f"🎯 Prioridad: {recomendaciones['prioridad'].upper()}")
print("📋 Recomendaciones:")
for rec in recomendaciones['recomendaciones']:
print(f" - {rec}")
print(f"🔌 Equipos activos: {', '.join(decision['equipos_activos'])}")
print(f"📦 Consumo estimado: {decision['consumo_estimado']:.0f} W")
print(f"🔗 Blockchain: {len(self.blockchain.cadena)} bloques")
print("=" * 60)
def detener(self):
"""Detiene el sistema"""
self.ejecutando = False
self.blockchain.registrar_evento("DETENER_SISTEMA", {"mensaje": "Sistema detenido"})
print("🛑 Sistema detenido")
def generar_informe(self) -> Dict:
"""Genera un informe completo del sistema"""
estado = self.monitor.obtener_estado()
return {
"capacidad_total": self.monitor._calcular_capacidad_total(),
"estado_actual": estado,
"bloques_blockchain": len(self.blockchain.cadena),
"blockchain_valida": self.blockchain.verificar_integridad(),
"historial_estados": len(self.historial_estados)
}
# ============================================================
# 7. DEMOSTRACIÓN DE FUNCIONAMIENTO
# ============================================================
def demo():
print("=" * 60)
print(" 🧠 SGEI - SISTEMA DE GESTIÓN ENERGÉTICA INTELIGENTE")
print(" PASAIA LAB · INTELIGENCIA LIBRE · DEEPSEEK")
print("=" * 60)
sistema = SistemaGestionEnergia()
try:
sistema.iniciar()
except KeyboardInterrupt:
sistema.detener()
print("\n✅ Sistema detenido correctamente")
# Mostrar informe final
informe = sistema.generar_informe()
print("\n📄 INFORME FINAL")
print("=" * 60)
print(f"Capacidad total: {informe['capacidad_total']:.0f} Wh")
print(f"Bloques en blockchain: {informe['bloques_blockchain']}")
print(f"Blockchain válida: {informe['blockchain_valida']}")
print("=" * 60)
if __name__ == "__main__":
demo()
-------------------------------------------------------------
📜 CERTIFICACIÓN
Certificado de diseño e implementación del Sistema de Gestión Energética Inteligente (SGEI) para PASAIA LAB
Certificado Nº: PASAIA-DS-2026-07-26-SGEI-01
Fecha: 26 de julio de 2026
Titular: José Agustín Fontán Varela
Entidades: PASAIA LAB – INTELIGENCIA LIBRE
Asesor IA: DeepSeek
Se certifica que el presente sistema de gestión energética inteligente (SGEI) ha sido desarrollado bajo la dirección intelectual de José Agustín Fontán Varela, CEO de PASAIA LAB y creador de INTELIGENCIA LIBRE, con la colaboración de DeepSeek como asesor IA.
El sistema integra:
Monitorización de recursos (generación solar, consumo y almacenamiento)
Blockchain de gestión para registro inmutable de todas las operaciones
IA de optimización para predicción de consumo y recomendaciones en tiempo real
Control de equipos basado en prioridades y estado de batería
Panel de control para visualización de datos y estadísticas
El sistema está diseñado para mantener la operatividad de PASAIA LAB durante cortes de suministro eléctrico, priorizando los equipos críticos y optimizando el consumo energético.
Certificado en Pasaia, a 26 de julio de 2026.
