Modbus TCP → host 127.0.0.1 puerto 5020 ·
un unit_id por pozo ·
CSV de import generado en simulators/variables_import.csv
"""
def build_api_payload():
"""Arma el snapshot agregado para el dashboard."""
wells = []
for unit in sorted(BANKS):
well_vars = BANKS[unit]["HR"] + BANKS[unit]["IR"]
hf = sum(1 for v in well_vars if v.is_hf)
wells.append({
"code": well_vars[0].well_code if well_vars else f"unit-{unit}",
"unit_id": unit,
"tags": len(well_vars),
"hf": hf,
"lf": len(well_vars) - hf,
})
sample = []
for v in random.sample(VARS, min(14, len(VARS))):
sample.append({
"well": v.well_code,
"alias": v.alias,
"address": f"{v.area}:{v.address}",
"value": int(round(v.value)),
"unit": v.unit,
})
with state.lock:
ts = state.last_update
changes = state.changes_last_cycle
running = state.modbus_running
hf_total = sum(1 for v in VARS if v.is_hf)
return {
"timestamp": ts,
"summary": {
"total": len(VARS),
"hf": hf_total,
"lf": len(VARS) - hf_total,
"wells": len(BANKS),
"changes_last_cycle": changes,
"modbus_running": running,
},
"wells": wells,
"sample": sample,
}
class SimulatorHTTPHandler(BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass
def do_GET(self):
clean_path = self.path.split('?', 1)[0]
if clean_path in ('/', '/dashboard', '/dashboard/'):
self.send_response(200)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(HTML_TEMPLATE.encode('utf-8'))
elif clean_path == '/api/data':
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(build_api_payload()).encode('utf-8'))
else:
self.send_response(404)
self.end_headers()
self.wfile.write(b"404 Not Found")
def start_http_server(host="0.0.0.0", port=HTTP_PORT):
server = ThreadingHTTPServer((host, port), SimulatorHTTPHandler)
state.http_running = True
log.info(f"🌐 [HTTP] Panel de monitoreo activo en: http://localhost:{port}")
threading.Thread(target=server.serve_forever, daemon=True).start()
return server
# ---------------------------------------------------------
# MODBUS SERVER (un slave / unit_id por pozo)
# ---------------------------------------------------------
async def run_modbus_server():
global modbus_context
if not pymodbus_installed:
return
slaves = {}
for unit in sorted(BANKS):
hr_vars = BANKS[unit]["HR"]
ir_vars = BANKS[unit]["IR"]
hr_size = max((v.address for v in hr_vars), default=-1)
ir_size = max((v.address for v in ir_vars), default=-1)
hr_values = [0] * (hr_size + 1)
ir_values = [0] * (ir_size + 1)
for v in hr_vars:
hr_values[v.address] = int(v.value)
for v in ir_vars:
ir_values[v.address] = int(v.value)
# zero_mode=True + bloque base 0 => wire address == índice == address de setValues
slaves[unit] = ModbusSlaveContext(
hr=ModbusSequentialDataBlock(0, hr_values),
ir=ModbusSequentialDataBlock(0, ir_values),
zero_mode=True,
)
modbus_context = ModbusServerContext(slaves=slaves, single=False)
log.info(f"🚀 [MODBUS] Servidor en {MODBUS_HOST}:{MODBUS_PORT} con {len(slaves)} pozos (unit_id {min(slaves)}..{max(slaves)})")
state.modbus_running = True
try:
await StartAsyncTcpServer(modbus_context, address=(MODBUS_HOST, MODBUS_PORT))
except Exception as e:
state.modbus_running = False
log.error(f"❌ [MODBUS] Error iniciando servidor: {e}")
# ---------------------------------------------------------
# LOOP DE SIMULACIÓN
# ---------------------------------------------------------
async def update_simulation_loop():
log.info(f"📊 [DATA LOOP] Generando {len(VARS)} variables (tick base {BASE_TICK_S}s)...")
cycle = 0
while True:
try:
now = time.time()
changes = 0
dirty_banks = set() # (unit, area) con al menos un cambio este tick
for v in VARS:
if now - v.last_step >= v.freq_ms / 1000.0:
v.last_step = now
v.step()
changes += 1
dirty_banks.add((v.unit_id, v.area))
# Volcar al datastore (setValues síncrono, base 0, por pozo/área)
if state.modbus_running and modbus_context and dirty_banks:
fc = {"HR": 3, "IR": 4}
for unit, area in dirty_banks:
bank = BANKS[unit][area]
values = [0] * (max(v.address for v in bank) + 1)
for v in bank:
values[v.address] = int(round(v.value))
try:
modbus_context[unit].setValues(fc[area], 0, values)
except Exception as e:
log.warning(f"setValues falló (unit {unit} {area}): {e}")
with state.lock:
state.last_update = time.strftime('%Y-%m-%d %H:%M:%S')
state.changes_last_cycle = changes
ts = state.last_update
running = state.modbus_running
cycle += 1
if cycle % 5 == 0: # resumen cada ~5 ticks
hf = sum(1 for v in VARS if v.is_hf)
status = "ON" if running else "OFF (pymodbus ausente)"
print(f"[{ts}] Modbus {status} | {len(VARS)} vars ({hf} HF / {len(VARS)-hf} LF) "
f"en {len(BANKS)} pozos | cambios este tick: {changes}")
except Exception as e:
log.error(f"Error en loop de simulación: {e}")
await asyncio.sleep(BASE_TICK_S)
# ---------------------------------------------------------
# MAIN
# ---------------------------------------------------------
def print_summary():
hf = sum(1 for v in VARS if v.is_hf)
print("\n" + "=" * 68)
print(" SIMULADOR DE CARGA OMNIOIL")
print("=" * 68)
print(f" Pozos (unit_id) : {WELL_COUNT} ({min(BANKS)}..{max(BANKS)})")
print(f" Variables totales : {len(VARS)}")
print(f" Alta frecuencia (HF) : {hf} cada {HF_INTERVAL_MS} ms")
print(f" Baja frecuencia (LF) : {len(VARS)-hf} cada {LF_INTERVAL_MS} ms")
print(f" Modbus TCP : 127.0.0.1:{MODBUS_PORT} (un unit_id por pozo)")
print(f" CSV de importación : {CSV_PATH}")
print(f" Panel de monitoreo : http://localhost:{HTTP_PORT}")
print("=" * 68 + "\n")
async def main():
log.info("✨ Iniciando Simulador de Carga Modbus OmniOil ✨")
write_import_csv()
print_summary()
http_server = start_http_server(port=HTTP_PORT)
tasks = [update_simulation_loop()]
if pymodbus_installed:
tasks.append(run_modbus_server())
else:
log.warning("Modbus desactivado: el panel y el CSV funcionan, pero no hay servidor TCP para el edge.")
try:
await asyncio.gather(*tasks)
except Exception as e:
log.error(f"Error en orquestador de tareas: {e}")
finally:
log.info("Deteniendo servidor HTTP...")
http_server.shutdown()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
log.info("👋 Simulador detenido por el usuario.")
except Exception as e:
log.error(f"💥 Error fatal en la ejecución: {e}")
sys.exit(1)