Sesión 5 · Persistencia de datos

Objetivos de la sesión

• Explicar por qué las lecturas de los sensores se guardan en una base de datos de series temporales.
• Guardar las magnitudes del invernadero en InfluxDB con measurements, tags y fields bien elegidos.
• Controlar cuánto se guarda: frecuencia de muestreo, escritura por lotes y solo los cambios de estado.
• Consultar el histórico con Flux: rango de tiempo, filtros, media, mínimo, máximo y ventanas.
• Publicar el histórico en una API y en una gráfica del panel con rango seleccionable, sin que un fallo de la base de datos pare el sistema.

Conexión con el sistema del invernadero

Hasta ahora el sistema solo conoce el presente: global.sensores guarda el último valor de cada magnitud, y las gráficas de la sesión 4 se vacían al reiniciar Node-RED. No puedes saber cuánto bajó la temperatura anoche ni cuántas veces se regó ayer.

En esta sesión añades la memoria del sistema: una pestaña que guarda en InfluxDB lo que entra por la capa de comunicaciones de la sesión 3, y lo consulta para el panel y para una API nueva, /api/invernadero/historico. En la sesión 7, el asistente de IA usará esa API para responder preguntas como "¿cuál fue la temperatura máxima de las últimas 6 horas?".

InfluxDB ya está en tu entorno desde la sesión 1: es el contenedor influxdb del docker-compose.yml, que hasta hoy no usaba nadie.

Conceptos clave

Por qué una base de datos de series temporales

Una serie temporal es una sucesión de valores con su instante: la temperatura del DHT22 cada 5 segundos. El invernadero produce unas 2 lecturas por segundo, que son 170 000 al día y más de 60 millones al año. Con datos así:

Lo que pasa	Qué necesita la base de datos
Se escribe sin parar y siempre al final, casi nunca se modifica	Escrituras muy rápidas y en lote
Casi todas las consultas son "entre tal hora y tal hora"	Datos ordenados e indexados por tiempo
Para una gráfica de una semana no hacen falta 120 000 puntos	Agregar por ventanas: una media cada hora
Los datos viejos valen menos	Retención: borrar o resumir lo antiguo automáticamente

Una base de datos relacional (PostgreSQL, MySQL) puede hacer todo esto, pero InfluxDB está construida para ello: comprime las series, tiene la agregación por ventanas como operación básica y gestiona la retención sola. Al final de la sesión verás cuándo tiene sentido la alternativa SQL.

El modelo de datos de InfluxDB

Cada dato guardado es un punto, con cuatro partes:

Parte	Qué es	En el invernadero
measurement	Qué se mide; como el nombre de una tabla	temperatura, humedad, peso…
tags	Etiquetas que describen el origen. Son texto y se indexan: sirven para filtrar y agrupar	sensor=dht22
fields	Los valores medidos. No se indexan	valor=24.3
timestamp	El instante	la hora de recepción, en milisegundos

Los puntos se guardan en un bucket: el del curso se llama sensores, dentro de la organización invernadero. Así se traduce el modelo de datos común de la sesión 3:

¿Tag o field? Pregúntate si vas a filtrar o agrupar por ese dato y si tiene pocos valores distintos. sensor cumple las dos cosas: filtras por dht22 y solo hay siete sensores. El valor medido no: cambia en cada lectura. Cada combinación de measurement y tags es una serie (temperatura,sensor=dht22 es una; temperatura,sensor=dht11, otra). Si pones como tag algo que cambia siempre, como el valor o la hora, creas una serie nueva por punto y la base de datos se degrada. Es el error de diseño más típico.

Flux: una consulta es un flujo

InfluxDB 2 se consulta con Flux, un lenguaje que se lee como un flujo de Node-RED: cada función recibe el resultado de la anterior por el operador |> (pipe forward).

from(bucket: "sensores")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "temperatura" and r.sensor == "dht22")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)

Función	Qué hace
from(bucket:)	De dónde se leen los datos
range(start:, stop:)	El intervalo de tiempo. -1h es "desde hace una hora"; stop es "ahora" si no se indica. Es obligatorio
filter(fn:)	Se queda con las filas que cumplen la condición. r es cada fila; sus columnas _measurement, _field, _value, _time y los tags
aggregateWindow(every:, fn:)	Parte el tiempo en ventanas y aplica a cada una mean, min, max, count… Es el downsampling
mean(), min(), max(), count()	Lo mismo, pero sobre todo el rango: un único valor

Documentación: Get started with Flux.

El reloj de InfluxDB es el real

InfluxDB guarda la hora real de cada lectura. Con el simulador a la velocidad por defecto (SIM_VELOCIDAD=60), una hora real son 60 horas simuladas: en la gráfica de la última hora verás pasar dos días y medio del invernadero. Con SIM_VELOCIDAD=1 (o con el invernadero real), las dos horas coinciden.

Práctica guiada paso a paso

Cada paso indica para qué sirve, qué hacer, qué debes ver y si no funciona. Lo nuevo de esta sesión son los nodos de InfluxDB, en la sección storage de la paleta (color marrón claro), y tres nodos del núcleo que sirven para controlar cuánto se escribe: delay, join y filter.

Paso 0 · Comprobar el entorno (5 min)

Para qué sirve. Node-RED necesita los nodos de InfluxDB y saber dónde está la base de datos. Las dos cosas vienen en el entorno del curso: los nodos, instalados en la imagen de Node-RED, y la conexión, en variables de entorno.

Qué hacer.

1. Actualiza tu entorno. Descarga de nuevo entorno-docker.zip y descomprímelo en el mismo sitio que en la sesión 1 (la carpeta que contiene docker, por ejemplo el Escritorio), aceptando reemplazar los ficheros. No pierdes nada: tu .env no va dentro del zip y tus flujos están en el volumen node-red-data.

   Por qué hay que descargarlo

   Los nodos de InfluxDB se instalan al construir la imagen de Node-RED, a partir de los ficheros de tu carpeta docker. docker compose up -d --build reconstruye con lo que tienes en disco, pero no descarga nada: si tu carpeta es de una versión anterior del curso, reconstruye la imagen antigua, sin InfluxDB.

2. En una terminal dentro de tu carpeta docker, reconstruye el entorno y comprueba el bucket:

   docker compose up -d --build
   docker compose exec node-red printenv INFLUX_BUCKET

Qué debes ver.

• printenv escribe sensores, el bucket donde guardarás las lecturas.
• En el editor (recarga la página), la paleta tiene una sección storage con influxdb in, influxdb out e influxdb batch.

Si no funciona. No escribe nada, o no aparece la sección storage: el zip no ha reemplazado tus ficheros. Comprueba que tu docker-compose.yml tiene un servicio influxdb; si no, probablemente lo descomprimiste en otro sitio (por ejemplo, dentro de la propia carpeta docker, que crea docker/docker). Repite el punto 1 en la carpeta correcta y luego los dos comandos.

Paso 1 · Conocer InfluxDB (10 min)

Para qué sirve. InfluxDB tiene su propia interfaz web. La usarás para comprobar qué se ha guardado y para probar consultas antes de llevarlas a Node-RED.

Qué hacer.

1. Abre http://localhost:8086 e inicia sesión con el usuario admin y la contraseña invernadero (los valores de INFLUX_USER e INFLUX_PASS en tu .env).
2. En el menú lateral, Load Data (icono de flecha hacia arriba) → Buckets: está sensores, creado al arrancar el contenedor.
3. Load Data → API Tokens: hay un token de administración. Su valor es token-curso-node-red (INFLUX_TOKEN en el .env), y es lo que Node-RED enviará en cada petición para identificarse.
4. En el menú lateral, Data Explorer (icono de gráfica): de momento, el bucket sensores no tiene nada que mostrar.

Qué debes ver. La organización invernadero, con el bucket sensores vacío.

Las credenciales del curso son de juguete

Usuario, contraseña y token están escritos en el .env y son iguales para todos: valen para un contenedor en tu PC, no para un servidor accesible desde la red. En la sesión 6 verás cómo se protegen las credenciales en Node-RED.

Paso 2 · Sacar los datos de la sesión 3 (10 min)

Para qué sirve. La nueva pestaña recibe lo mismo que ya procesa la sesión 3, sin volver a suscribirse a MQTT ni repetir la normalización. Se usa la misma técnica que en el panel: nodos link.

Qué hacer.

1. Crea una pestaña S5 · Histórico. Añade dos link in (sección common), llamados datos normalizados (S3) y estado actuadores (S3), y une un debug a cada uno.
2. En tu pestaña de la sesión 3, añade dos link out:
   • a histórico (datos), unido a la salida 1 de normalizar (la de los datos válidos). Ábrelo y marca en la lista datos normalizados (S3).
   • a histórico (actuadores), unido a la salida de estado actuadores. Márcale estado actuadores (S3).
3. Deploy.

Qué debes ver. En la barra de depuración, los objetos del modelo común, como { sensor: "dht22", magnitud: "temperatura", valor: 24.3, … }, y, al desplegar, el estado de los dos Shelly: { equipo: "bomba", estado: "off" }.

Cuando lo veas, desactiva los dos debug (botón de su derecha): llega un mensaje cada medio segundo y llenarían la barra.

Si no funciona. No llega nada: el link out no tiene marcado su link in (ábrelo y comprueba la casilla), o lo uniste a otra salida de normalizar.

Paso 3 · Del modelo común a un punto (25 min)

Para qué sirve. Escribir en InfluxDB, por primera vez, las lecturas del invernadero, con la traducción del modelo común a punto de los conceptos clave.

Qué hacer.

1. Detrás de datos normalizados (S3) (en lugar de su debug), un function llamado a punto InfluxDB:

   // Modelo común (S3) -> punto de InfluxDB
   const p = msg.payload;
   // Un valor retenido es una lectura vieja que el broker entrega al
   // suscribirse (por ejemplo, al hacer Deploy): no es una medida nueva
   if (p.retenido) return null;
   
   const punto = {
       measurement: p.magnitud,          // qué se mide: temperatura, humedad...
       tags: { sensor: p.sensor },       // quién lo mide: se indexa
       fields: { valor: p.valor },       // el número medido
       timestamp: new Date(p.ts)         // cuándo
   };
   // El nodo de escritura espera una lista de puntos: de momento, de uno
   msg.payload = [punto];
   return msg;

   El punto descarta los mensajes retenidos. Al hacer Deploy, el broker entrega de golpe la última lectura de cada topic, que puede tener horas, con retenido: true; como ts es la hora de recepción, guardarla pondría un valor viejo con la hora de ahora.

2. Detrás, un influxdb batch (sección storage). Ábrelo; en Server aparece Add new influxdb...: pulsa el lápiz para crear la conexión:

   

   Campo	Valor	Por qué
   Name	InfluxDB invernadero
   Version	2.0	Cambia los campos del diálogo: con 2.0 se usa URL y token
   URL	${INFLUX_URL}	http://influxdb:8086: el nombre del contenedor en la red de Docker, no localhost
   Token	${INFLUX_TOKEN}	Se lee de la variable de entorno, como la contraseña del broker

   Pulsa Add. De vuelta en el influxdb batch:

   

   Campo	Valor
   Organization	${INFLUX_ORG}
   Bucket	${INFLUX_BUCKET}
   Time Precision	Milliseconds (ms)
   Name	escribir en InfluxDB

3. Deploy.

4. En InfluxDB, Data Explorer: en FROM elige sensores; en el primer Filter (_measurement), marca temperatura. Pulsa SUBMIT.

Qué debes ver. Una gráfica con dos líneas que crecen hacia la derecha a medida que llegan lecturas: la temperatura del DHT22 y la del DHT11.

La línea en escalones es el DHT11, que mide en grados enteros. Tener el histórico permite ver cosas así, que en un gauge pasan desapercibidas.

Si no funciona.

• En la barra de depuración aparece un error con unauthorized: el token no es correcto. Escribe ${INFLUX_TOKEN} tal cual, con el dólar y las llaves.
• bucket not found u organization not found: revisa Organization y Bucket.
• Los puntos aparecen en 1970 o en el futuro: Time Precision no está en milisegundos.

Paso 4 · Frecuencia de muestreo (15 min)

Para qué sirve. Decidir cuánto guardar. Ahora mismo se guarda todo lo que publican los sensores: unas 2 lecturas por segundo. Para ver la evolución de la temperatura durante horas, una cada 10 segundos por magnitud es de sobra, y ocupa la mitad.

Qué hacer.

1. Arrastra un delay (sección function) y suéltalo sobre el cable entre datos normalizados (S3) y a punto InfluxDB: el cable se resalta y el nodo queda insertado.

2. Ábrelo:

   

   Campo	Valor	Por qué
   Action	Rate Limit · For each msg.topic	Un límite por topic: cada magnitud tiene su propio ritmo
   Rate	1 msg(s) per 10 Seconds	Una muestra cada 10 s
   	Drop intermediate messages · Send all topics	Cada 10 s sale el último valor de cada topic; los anteriores se descartan
   Name	1 cada 10 s por topic

3. Deploy.

Qué debes ver. Debajo del delay, un número: los topics que tiene retenidos esperando su turno (unos 10). En el Data Explorer, activa View Raw Data: los puntos de cada sensor están separados unos 10 s.

¿Cada cuánto muestrear? Depende de lo rápido que cambia la magnitud y de para qué se usa. La temperatura de un invernadero real cambia en minutos, así que una muestra por minuto bastaría. Con el simulador acelerado 60 veces, 10 s reales son 10 minutos simulados. Muestrear sin pensar en el uso solo llena el disco.

Paso 5 · Escribir por lotes (15 min)

Para qué sirve. Cada escritura es una petición HTTP a InfluxDB, con su conexión y su respuesta. Agrupar 20 puntos en una sola petición reduce 20 veces el trabajo de los dos lados.

Qué hacer.

1. Arrastra un join (sección sequence) sobre el cable entre a punto InfluxDB y escribir en InfluxDB. Ábrelo:

   

   Campo	Valor	Por qué
   Mode	manual	Tú decides cuándo se envía el grupo
   Combine each / to create	msg.payload / an Array	Junta los payloads en una lista
   After a number of message parts	20	Un lote cada 20 puntos
   After a timeout following the first message	30	Si llegan pocos, no esperar más de 30 s
   Name	lote	Debajo se verá lote 20

2. Ahora la lista la forma el join. En a punto InfluxDB, cambia la última asignación por:

   msg.payload = punto;

3. Deploy.

Qué debes ver. En el Data Explorer, pulsando SUBMIT cada poco, los datos llegan a saltos, unos 20 puntos de golpe cada 20 s. Pon un debug temporal tras el join: cada mensaje es un array de 20 puntos.

Si no funciona. Error msg.payload.forEach is not a function: el influxdb batch recibe un punto suelto en vez de una lista. Revisa que el join esté delante y en modo an Array.

El precio de los lotes

Un punto puede esperar hasta 30 s en el join antes de llegar a la base de datos, y si Node-RED se detiene en ese momento, el lote pendiente se pierde. Para un histórico es un precio razonable; para un sistema de facturación, no.

Paso 6 · Guardar los actuadores (15 min)

Para qué sirve. Registrar cuándo se encienden y apagan la bomba y el ventilador. Los Shelly publican su estado cada 30 s aunque no cambie, pero lo que interesa son los cambios.

Qué hacer.

1. Detrás de estado actuadores (S3) (en lugar de su debug), un filter (sección function; en versiones antiguas se llamaba rbe):

   

   Campo	Valor	Por qué
   Mode	block unless value changes	Solo pasa el mensaje si el valor es distinto del anterior
   Property	msg.payload.estado	Lo que se compara: on u off
   Apply mode separately for each	✔ msg.topic	Un "valor anterior" por Shelly; si no, la bomba y el ventilador se mezclarían
   Name	solo si cambia

2. Detrás, un function a punto actuador, cuya salida va a la entrada del join lote:

   // Estado confirmado de un Shelly (S3) -> punto de InfluxDB
   msg.payload = {
       measurement: "actuador",
       tags: { equipo: msg.payload.equipo },
       // 1 encendido, 0 apagado: como número se puede dibujar y contar
       fields: { encendido: msg.payload.estado === "on" ? 1 : 0 },
       timestamp: new Date()
   };
   return msg;

3. Deploy. Enciende la bomba desde el panel de la sesión 4 y apágala un minuto después.

Qué debes ver. En el Data Explorer, measurement actuador, field encendido: un escalón que sube a 1 y baja a 0. Con View Raw Data, solo hay puntos en los cambios, más uno por Shelly cada vez que despliegas (el filter empieza sin "valor anterior", así que deja pasar el primero; es el estado en ese momento, y está bien guardarlo).

Paso 7 · Tu primera consulta (20 min)

Para qué sirve. Leer el histórico desde Node-RED, con una consulta Flux.

Qué hacer.

1. En el Data Explorer, con temperatura seleccionada, pulsa SCRIPT EDITOR: verás la consulta Flux que ha generado la interfaz. Es una buena forma de aprender Flux: construyes la consulta con clics y lees el código.

2. En Node-RED: un inject llamado consultar → un function consulta Flux:

   // Media de la temperatura interior en ventanas de 5 minutos, última hora
   const bucket = env.get("INFLUX_BUCKET");
   msg.query = `
   from(bucket: "${bucket}")
     |> range(start: -1h)
     |> filter(fn: (r) => r._measurement == "temperatura" and r.sensor == "dht22")
     |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
   `;
   return msg;

   La consulta va en msg.query, con comillas invertidas (`) para poder escribirla en varias líneas e insertar el nombre del bucket con ${...}.

3. Detrás, un influxdb in (sección storage):

   

   • Server: InfluxDB invernadero (el que ya creaste).
   • Organization: ${INFLUX_ORG}.
   • Query: vacía. Si escribes algo aquí, se ignora msg.query.
   • Name: leer de InfluxDB.

4. Un debug resultado consulta. Deploy y pulsa consultar.

Qué debes ver. Un array con una fila por cada ventana de 5 minutos que tenga datos: con una hora de histórico, 12 o 13.

Columna	Qué es
_time	El final de la ventana, en hora UTC (la Z final): en España, una o dos horas menos que la hora local
_value	La media de esa ventana
_measurement, _field, sensor	De qué serie es la fila
_start, _stop	El rango de la consulta
table	Cada serie distinta sale en una "tabla" numerada

Si no funciona. error: ... expected ...: error de sintaxis en Flux. Copia la consulta en el SCRIPT EDITOR de InfluxDB, que marca la línea del error. Array vacío: aún no hay datos en la última hora, o el filtro no coincide (las mayúsculas cuentan).

Paso 8 · Una API para el histórico (25 min)

Para qué sirve. Que cualquier programa, y en la sesión 7 el asistente de IA, pueda preguntar por el histórico: media, mínimo y máximo de una magnitud en las últimas horas.

Qué hacer.

1. Un http in con Method GET y URL /api/invernadero/historico.

2. Un function validar y consultar con 2 salidas:

   // GET /api/invernadero/historico?magnitud=temperatura&sensor=dht22&horas=6
   const MAGNITUDES = {
       temperatura: ["dht22", "dht11"], humedad: ["dht22", "dht11", "suelo"],
       distancia: ["ultrasonidos"], peso: ["balanza"], tds: ["tds"]
   };
   const q = msg.req.query;
   const magnitud = q.magnitud || "temperatura";
   const sensor = q.sensor || (MAGNITUDES[magnitud] || [])[0];
   const horas = Number(q.horas || 1);
   
   // Validar ANTES de construir la consulta: nunca se pega en ella texto
   // del usuario sin comprobarlo (es la misma idea que la inyección SQL)
   if (!MAGNITUDES[magnitud] || !MAGNITUDES[magnitud].includes(sensor) ||
       !Number.isInteger(horas) || horas < 1 || horas > 168) {
       msg.statusCode = 400;
       msg.payload = { error: "Parámetros no válidos",
                       magnitudes: MAGNITUDES, horas: "entero de 1 a 168" };
       return [null, msg];
   }
   msg.consulta = { magnitud, sensor, horas };
   msg.query = `
   datos = from(bucket: "${env.get("INFLUX_BUCKET")}")
     |> range(start: -${horas}h)
     |> filter(fn: (r) => r._measurement == "${magnitud}" and r.sensor == "${sensor}")
   union(tables: [
     datos |> mean()  |> set(key: "_field", value: "media"),
     datos |> min()   |> set(key: "_field", value: "minimo"),
     datos |> max()   |> set(key: "_field", value: "maximo"),
     datos |> count() |> toFloat() |> set(key: "_field", value: "muestras")
   ])
   `;
   return [msg, null];

   Lo nuevo en Flux:

   • datos = ... guarda el resultado del filtro en una variable para usarlo cuatro veces.
   • Cada línea del union calcula un resumen y, con set, le pone nombre en la columna _field. union junta las cuatro filas en un único resultado. toFloat() hace falta porque count() devuelve un entero y union necesita el mismo tipo en todas las filas.

3. Salida 1 → un influxdb in leer resumen (mismo servidor y organización, Query vacía) → un function a respuesta JSON:

   // Filas de InfluxDB -> respuesta JSON
   const r = { ...msg.consulta };
   for (const fila of msg.payload) {
       r[fila._field] = fila._field === "muestras"
           ? fila._value
           : Math.round(fila._value * 10) / 10;
   }
   if (!r.muestras) r.aviso = "No hay datos en ese periodo";
   msg.payload = r;
   return msg;

4. La salida de a respuesta JSON y la salida 2 de validar y consultar, a un único http response.

5. Deploy.

Qué debes ver. Abre http://localhost:1880/api/invernadero/historico?magnitud=temperatura&sensor=dht22&horas=1:

Prueba también ?magnitud=humedad&sensor=suelo&horas=2 y un valor no válido, como ?horas=500: responde con el código 400 y explica qué parámetros acepta.

Nunca construyas una consulta con texto sin validar

Si validar y consultar pegara en la consulta lo que llega en la URL sin comprobarlo, alguien podría escribir en magnitud un trozo de Flux que cerrara el filtro y leyera otra cosa. Es la inyección de siempre, la misma que en SQL. Por eso solo se aceptan magnitudes y sensores de una lista cerrada y horas que sean un entero en un rango.

Paso 9 · El histórico en el panel (30 min)

Para qué sirve. Una gráfica que no se pierde al recargar ni al reiniciar, con un rango que elige el operador: de los últimos 15 minutos a la última semana. Muestra, para cada ventana de tiempo, la media, la mínima y la máxima.

Qué hacer.

1. Un grupo nuevo. En la barra lateral Dashboard 2.0, añade a la página Invernadero un grupo Histórico de ancho 12, el último.

2. El selector. Un dropdown (sección dashboard 2) en el grupo Histórico:

   

   • Label: Rango.
   • Options: con + option, cinco opciones de tipo string. El valor es lo que se envía y la etiqueta lo que ve el operador: 15m · Últimos 15 minutos, 1h · Última hora, 6h · Últimas 6 horas, 24h · Últimas 24 horas y 7d · Últimos 7 días.
   • Topic: rango.

3. El valor inicial. Un inject rango por defecto, con Inject once after 1 segundo y msg.payload string 1h, unido a la entrada del dropdown. Así el selector aparece con Última hora elegido.

4. La consulta. Un function consulta de la gráfica que recibe la salida del dropdown y un inject cada minuto (Repeat cada 60 segundos, Inject once after 3 segundos):

   // Rango elegido en el panel -> consulta con media, mínima y máxima por ventana
   const RANGOS = {            // rango -> ventana: unos 60-170 puntos por serie
       "15m": "15s", "1h": "1m", "6h": "5m", "24h": "15m", "7d": "1h"
   };
   if (msg.topic === "rango") flow.set("rango", msg.payload);
   const rango = flow.get("rango") || "1h";
   const ventana = RANGOS[rango];
   if (!ventana) return null;
   
   const ventanas = (fn, nombre) =>
       `datos |> aggregateWindow(every: ${ventana}, fn: ${fn}, createEmpty: false) |> set(key: "_field", value: "${nombre}")`;
   msg.query = `
   datos = from(bucket: "${env.get("INFLUX_BUCKET")}")
     |> range(start: -${rango})
     |> filter(fn: (r) => r._measurement == "temperatura" and r.sensor == "dht22")
   union(tables: [
     ${ventanas("mean", "media")},
     ${ventanas("min", "mínima")},
     ${ventanas("max", "máxima")}
   ])
   `;
   return msg;

   • El rango elegido se guarda en el contexto de flujo: el inject de cada minuto repite la consulta con el último rango, y así la gráfica sigue avanzando.
   • La ventana crece con el rango: para 7 días, una ventana de 1 hora da 168 puntos por serie; con ventanas de 1 minuto serían 10 000, demasiados para el navegador e imposibles de distinguir en la pantalla.
   • El rango nunca se pega en la consulta sin comprobarlo: si no está en RANGOS, return null.

5. Detrás, un influxdb in leer histórico (Query vacía) y un function a puntos de la gráfica:

   // Filas de InfluxDB -> puntos de la gráfica (una serie por _field)
   msg.payload = msg.payload.map(fila => ({
       serie: fila._field,
       hora: new Date(fila._time).getTime(),
       valor: Math.round(fila._value * 10) / 10
   }));
   node.status({ text: `${msg.payload.length} puntos` });
   return msg;

6. La gráfica. Un chart en el grupo Histórico, de ancho 12 y alto 6, con Label Temperatura interior (ºC):

   

   Campo	Valor	Por qué
   Action	Replace	Cada mensaje trae la gráfica completa y sustituye a la anterior. En la sesión 4 era Append: cada mensaje añadía un punto
   Point Style	None	Con muchos puntos, la línea sola se lee mejor
   X-Axis Type	Timescale
   X-Axis Limit	last 7 Days	La gráfica borra lo más antiguo que este límite. Con 1 hora, el rango de 7 días se cortaría
   Series	tipo key: serie	Una línea por cada valor distinto: media, mínima, máxima
   X	tipo key: hora	La hora de cada punto la trae el propio punto, no es la de llegada
   Y	tipo key: valor

7. Deploy y abre el panel.

Qué debes ver. El grupo Histórico con el selector en Última hora y la gráfica con tres líneas: la media entre la mínima y la máxima de cada minuto. Debajo del function a puntos de la gráfica, el número de puntos. Cambia el rango: la gráfica se redibuja. Recarga la página del panel: la gráfica sigue ahí.

Las tres líneas casi se tocan cuando la temperatura apenas cambia dentro de cada ventana, y se separan cuando cambia rápido o con rangos largos, cuyas ventanas son más anchas.

Si no funciona. La gráfica sale vacía y el estado dice 0 puntos: no hay datos en ese rango; prueba con 15 minutos. Sale una sola línea con todo mezclado: Series no es de tipo key con el valor serie.

Paso 10 · Si InfluxDB se cae (10 min)

Para qué sirve. Una base de datos caída no debe tumbar el sistema de control. Como con la API externa de la sesión 2, los errores se recogen con un catch y se hacen visibles.

Qué hacer.

1. Un catch (sección common). En Catch errors from, elige selected nodes y, con Select nodes, marca los cuatro nodos de InfluxDB de la pestaña. Name errores de InfluxDB.

2. Detrás, un function contar errores y un debug errores InfluxDB:

   // Un fallo de InfluxDB no debe parar el resto del sistema
   const n = (context.get("errores") || 0) + 1;
   context.set("errores", n);
   node.status({ fill: "red", shape: "ring", text: `${n} errores; último: ${msg.error.message}` });
   return msg;

3. Deploy. Para la base de datos: docker compose stop influxdb.

Qué debes ver.

• Los nodos de lectura fallan al momento: pulsa consultar o cambia el rango del panel y aparecerá un error connect ECONNREFUSED, contado bajo contar errores.
• Las escrituras no fallan enseguida: el cliente de InfluxDB guarda cada lote y lo reintenta, con esperas cada vez más largas, durante unos 3 minutos. En el log (docker compose logs node-red) verás Write to InfluxDB failed (attempt: 1), attempt: 2… Pasado ese tiempo, el lote se descarta y su error llega al catch.
• Mientras tanto, el resto sigue funcionando: el panel muestra los valores actuales y el riego automático decide.

Espera unos 5 minutos y vuelve a arrancarla con docker compose start influxdb. Las escrituras siguen solas, y los lotes que aún se estaban reintentando llegan a guardarse; los más antiguos se descartaron. Si la caída dura menos de unos 3 minutos, no se pierde nada.

Ahora busca el hueco. En la gráfica del panel, con el rango de 15 minutos, no se ve: la línea une el último punto de antes de la caída con el primero de después, como si la temperatura hubiera cambiado en línea recta. En el Data Explorer, con View Raw Data, sí se ven los minutos sin puntos. Una gráfica puede esconder que faltan datos; si importa, la consulta debe usar createEmpty: true y la gráfica tiene que saber dibujar los huecos.

El flujo completo de la práctica guiada:

Flujo de referencia: s5-historico.json. Necesita los de las sesiones 3 y 4, que también se han actualizado: el de la S3 trae los dos link out del paso 2. Al importarlo, abre InfluxDB invernadero y escribe el token ${INFLUX_TOKEN}: las credenciales no se exportan.

Node-RED como plataforma

Tres nodos para controlar el caudal

En esta sesión has usado tres nodos del núcleo que no son de InfluxDB, sino de control de caudal. Sirven igual delante de una base de datos, una API con cuota o un servicio de mensajería:

Nodo	Para qué	Uso en la sesión
delay en modo Rate Limit	Limitar cuántos mensajes pasan por unidad de tiempo, en total o por topic	Una muestra cada 10 s por magnitud
join en modo manual	Agrupar varios mensajes en uno: por número, por tiempo o al llegar una marca	Lotes de 20 puntos
filter (antes rbe, Report By Exception)	Dejar pasar solo lo que cambia, o lo que cambia más de un margen (deadband)	Solo los cambios de los actuadores

El filter tiene otro modo útil para sensores: block unless value change is greater or equal to un margen. Con un margen de 0,5 ºC, la temperatura solo se guardaría cuando cambia de verdad, y no por el ruido del sensor. El inconveniente es que, si la temperatura no cambia en horas, no se guarda nada y la gráfica no distingue "estable" de "sensor muerto". El muestreo periódico del paso 4 no tiene ese problema.

El opuesto del join es el split, que parte un array en mensajes sueltos. batch agrupa mensajes por número o por intervalo, sin juntarlos: les añade msg.parts para que un join en modo automático los una después.

Retención y downsampling

El bucket sensores se creó sin límite: guarda todo para siempre. En un sistema real se combinan dos medidas:

• Retención: en la configuración del bucket (Load Data → Buckets → Settings), borrar automáticamente lo que tenga más de, por ejemplo, 30 días.
• Downsampling: una tarea de InfluxDB (Tasks) que cada hora calcula medias horarias y las guarda en otro bucket de retención larga. Es la misma aggregateWindow de esta sesión, ejecutada de forma programada.

Así se conserva el detalle reciente y la tendencia de años sin que el disco crezca sin límite.

La alternativa SQL

Si el sistema ya tiene una base de datos relacional, o los datos se cruzan con otros (lotes de cultivo, operarios, pedidos), guardar las lecturas en SQL puede ser mejor que añadir InfluxDB. Lo habitual es una tabla estrecha:

CREATE TABLE lecturas (
    ts        TIMESTAMPTZ NOT NULL,
    sensor    TEXT        NOT NULL,
    magnitud  TEXT        NOT NULL,
    valor     DOUBLE PRECISION
);
CREATE INDEX ON lecturas (magnitud, sensor, ts);

-- Media por ventanas de 5 minutos de la última hora (PostgreSQL)
SELECT date_bin('5 minutes', ts, TIMESTAMPTZ '2000-01-01') AS ventana, avg(valor)
FROM lecturas
WHERE magnitud = 'temperatura' AND sensor = 'dht22' AND ts > now() - interval '1 hour'
GROUP BY ventana ORDER BY ventana;

	InfluxDB	SQL (PostgreSQL, SQLite…)
Modelo	Measurement, tags, fields, sin esquema fijo	Tablas con esquema
Ventanas, retención, compresión	Incluidas	A mano (o con la extensión TimescaleDB)
Cruzar con otros datos	Limitado	Lo natural: JOIN
En Node-RED	node-red-contrib-influxdb	node-red-node-sqlite, node-red-contrib-postgresql

El flujo de Node-RED apenas cambia: la función que construye el punto construiría un INSERT con parámetros y la consulta sería SQL en lugar de Flux. El muestreo, los lotes, la validación y el catch serían los mismos.

Entregable y evidencias para el informe

Qué debe funcionar

1. Histórico de todas las magnitudes: las lecturas del invernadero se guardan en InfluxDB con el modelo del paso 3, muestreadas y por lotes, y los cambios de los actuadores con el measurement actuador.

2. API del histórico GET /api/invernadero/historico, con validación de parámetros y respuesta 400 si no son válidos.

3. Gráficas del histórico en el panel, con rango seleccionable:

   • la de temperatura interior de la práctica guiada;
   • una segunda gráfica, Humedad del suelo (%), en el mismo grupo y con el mismo selector de rango, con el eje Y de 0 a 100. Con un solo selector, las dos gráficas deben mostrar siempre el mismo rango.

   Decide qué haces con el grupo Tendencias (última hora) de la sesión 4 (quitarlo, mantenerlo o cambiarlo) y justifícalo en el informe.

4. Robustez: con InfluxDB parado, el resto del sistema sigue funcionando y los errores se cuentan y se ven.

Reto opcional (elige uno):

• Riegos de hoy: añade a la API GET /api/invernadero/riegos?horas=24, que devuelva cuántas veces se encendió la bomba en ese periodo. Pista: filter con r._measurement == "actuador" and r.equipo == "bomba" and r._value == 1 y después count().
• Retención y downsampling: pon al bucket sensores una retención de 7 días y crea en InfluxDB una tarea que guarde cada hora la media horaria de cada serie en un bucket sensores_1h sin límite. Explica en el informe qué consultas harías a cada bucket.

Cómo se comprueba

• En el Data Explorer de InfluxDB aparecen todos los measurements del invernadero, con puntos cada 10 s, y actuador con un punto por cada encendido y apagado.
• /api/invernadero/historico responde con media, minimo, maximo y muestras, y con 400 ante ?horas=0 o una magnitud inventada.
• El panel conserva las gráficas del histórico al recargar la página y al reiniciar Node-RED (docker compose restart node-red).
• Con docker compose stop influxdb, el panel sigue mostrando los valores actuales y aparecen errores contados; al arrancarla, se recupera sola.

Criterios de calidad:

• Tags con pocos valores distintos; ningún valor ni hora como tag.
• Ninguna consulta construida con texto del usuario sin validar.
• No se guardan los mensajes retenidos.
• Nodos con nombre y flujo organizado en grupos.

Evidencias que debes guardar

• [ ] Captura del Data Explorer de InfluxDB con varias magnitudes.
• [ ] Captura del panel con las dos gráficas del histórico en dos rangos distintos.
• [ ] Captura del navegador con la respuesta de la API, válida y con error 400.
• [ ] Captura de los errores contados con InfluxDB parado.
• [ ] Captura del flujo de la pestaña S5 en el editor.
• [ ] Flujo exportado: s5-<tu-apellido>.json.
• [ ] En el informe, respuesta breve a:
  • ¿Por qué sensor es un tag y valor un field? ¿Qué pasaría si guardaras valor como tag?
  • ¿Qué frecuencia de muestreo elegirías para el invernadero real y por qué? ¿Cuántos puntos al año supone?
  • ¿Por qué se descartan los mensajes retenidos en lugar de guardarlos?
  • ¿Qué se pierde y qué no cuando InfluxDB está parado un minuto? ¿Y una hora?

Recursos adicionales

• InfluxDB 2 — Conceptos clave (measurements, tags, fields)
• InfluxDB 2 — Get started with Flux
• Flux — aggregateWindow()
• InfluxDB 2 — Line protocol
• node-red-contrib-influxdb
• Node-RED — nodos delay, join y filter (Cookbook)