En esta práctica vais a construir un pequeño sistema de adquisición de datos distribuido. En lugar de trabajar con un único sensor conectado a un solo dispositivo, vais a montar una arquitectura formada por varios Arduinos, cada uno con uno o varios sensores, y una Raspberry Pi que actuará como centro de recepción y almacenamiento.
El objetivo no es solo «leer sensores», sino comprender cómo se organiza un sistema real donde varios nodos capturan información y un equipo central la recibe, la procesa y la guarda para su análisis posterior.
Este tipo de arquitectura se parece mucho a sistemas reales de monitorización, automatización, IoT, domótica o control industrial.
2. Objetivos de la práctica
Al finalizar la práctica, debéis ser capaces de:
Conectar sensores a varios Arduinos.
Leer datos de sensores desde Arduino.
Enviar esos datos por comunicación serie a través de USB.
Conectar varios Arduinos a una Raspberry Pi.
Identificar los puertos serie de cada Arduino en Linux.
Crear en la Raspberry Pi un programa en Python que reciba datos desde varios Arduinos.
Guardar los datos recibidos en archivos CSV o JSON.
Documentar correctamente el montaje, el código y el funcionamiento del sistema.
3. Descripción general del sistema
La arquitectura general de la práctica será la siguiente:
18. Script Python de ejemplo para varios Arduinos y CSV
import serial
import threading
import csv
import os
from datetime import datetime
PUERTOS = [
"/dev/ttyACM0",
"/dev/ttyACM1",
"/dev/ttyACM2"
]
BAUDIOS = 9600
ARCHIVO_CSV = "datos_sensores.csv"
CAMPOS = ["timestamp", "arduino", "sensor", "valor"]
lock_csv = threading.Lock()
def parsear_linea(linea):
datos = {}
partes = linea.strip().split(";")
for parte in partes:
if "=" in parte:
clave, valor = parte.split("=", 1)
datos[clave.strip()] = valor.strip()
return datos
def inicializar_csv():
if not os.path.exists(ARCHIVO_CSV) or os.path.getsize(ARCHIVO_CSV) == 0:
with open(ARCHIVO_CSV, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=CAMPOS)
writer.writeheader()
def guardar_csv(registro):
with lock_csv:
with open(ARCHIVO_CSV, "a", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=CAMPOS)
writer.writerow(registro)
def escuchar_puerto(puerto):
try:
ser = serial.Serial(puerto, BAUDIOS, timeout=1)
print(f"Escuchando {puerto}")
while True:
linea = ser.readline().decode("utf-8", errors="ignore").strip()
if linea:
print(f"{puerto} -> {linea}")
datos = parsear_linea(linea)
if "arduino" in datos and "sensor" in datos and "valor" in datos:
registro = {
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"arduino": datos["arduino"],
"sensor": datos["sensor"],
"valor": datos["valor"]
}
guardar_csv(registro)
except Exception as e:
print(f"Error en {puerto}: {e}")
def main():
inicializar_csv()
hilos = []
for puerto in PUERTOS:
hilo = threading.Thread(target=escuchar_puerto, args=(puerto,), daemon=True)
hilo.start()
hilos.append(hilo)
for hilo in hilos:
hilo.join()
if __name__ == "__main__":
main()
19. Script Python de ejemplo para guardar en JSON
import serial
import threading
import json
from datetime import datetime
PUERTOS = [
"/dev/ttyACM0",
"/dev/ttyACM1"
]
BAUDIOS = 9600
ARCHIVO_JSON = "datos_sensores.jsonl"
lock_json = threading.Lock()
def parsear_linea(linea):
datos = {}
partes = linea.strip().split(";")
for parte in partes:
if "=" in parte:
clave, valor = parte.split("=", 1)
datos[clave.strip()] = valor.strip()
return datos
def guardar_json(registro):
with lock_json:
with open(ARCHIVO_JSON, "a", encoding="utf-8") as f:
f.write(json.dumps(registro, ensure_ascii=False) + "\n")
def escuchar_puerto(puerto):
try:
ser = serial.Serial(puerto, BAUDIOS, timeout=1)
print(f"Escuchando {puerto}")
while True:
linea = ser.readline().decode("utf-8", errors="ignore").strip()
if linea:
print(f"{puerto} -> {linea}")
datos = parsear_linea(linea)
if "arduino" in datos and "sensor" in datos and "valor" in datos:
registro = {
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"arduino": datos["arduino"],
"sensor": datos["sensor"],
"valor": datos["valor"]
}
guardar_json(registro)
except Exception as e:
print(f"Error en {puerto}: {e}")
def main():
hilos = []
for puerto in PUERTOS:
hilo = threading.Thread(target=escuchar_puerto, args=(puerto,), daemon=True)
hilo.start()
hilos.append(hilo)
for hilo in hilos:
hilo.join()
if __name__ == "__main__":
main()
FASE 7. Ejecución del programa
20. Guardar y ejecutar el script
Guardad el script con un nombre como:
nano receptor_sensores.py
Pegad el código, guardadlo y ejecutadlo:
python3 receptor_sensores.py
Si todo funciona, deberíais ver en pantalla líneas como estas:
Esta práctica os permitirá trabajar con una arquitectura completa de adquisición de datos usando varios microcontroladores y un sistema Linux central.
No se trata solo de conectar cables y leer números, sino de entender cómo diseñar un sistema modular, organizado y ampliable. Ese enfoque es mucho más valioso que hacer una sola lectura aislada en un monitor serie.
Cuando el sistema funcione, habréis construido una pequeña red de sensores con almacenamiento centralizado, algo muy cercano a muchos proyectos reales de automatización, IoT y monitorización.
