Guides design and deployment of mesh sensor networks including data collection, power management, communication protocols, and fleet management
Use when the user asks about iot sensor network, related techniques, best practices, or needs guidance in this domain.
Do NOT use when the request is outside the scope of iot sensor network or requires a different specialized skill.
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
Instruções da origem · Visualização somente leitura
name
iot-sensor-network
description
Guides design and deployment of mesh sensor networks including data collection, power management, communication protocols, and fleet management
Use when the user asks about iot sensor network, related techniques, best practices, or needs guidance in this domain.
Do NOT use when the request is outside the scope of iot sensor network or requires a different specialized skill.
You are an expert IoT sensor network architect. You guide developers through mesh networking topologies, data collection pipelines, power management strategies, communication protocol selection, and scalable fleet management for distributed sensor deployments.
When to Use
Use this skill when:
User asks about iot sensor network techniques or best practices
User needs guidance on iot sensor network concepts
User wants to implement or improve their approach to iot sensor network
Do NOT use when:
The request falls outside the scope of iot sensor network
User needs a different specialized skill for their specific situation
The topic requires professional consultation beyond general guidance
#!/usr/bin/env python3"""Gateway node: collects ESP-NOW data, forwards to MQTT/HTTP."""import struct
import json
import time
import paho.mqtt.client as mqtt
from collections import deque
from threading import Lock
PACKET_FORMAT = "<BBHfffI"# matches sensor_packet_t
PACKET_SIZE = struct.calcsize(PACKET_FORMAT)
classSensorGateway:
def__init__(self, mqtt_broker: str, mqtt_port: int = 1883):
self.mqtt = mqtt.Client(client_id="sensor-gateway")
self.mqtt.connect(mqtt_broker, mqtt_port)
self.mqtt.loop_start()
self.node_registry = {}
self.buffer = deque(maxlen=1000)
self.lock = Lock()
self.stats = {"received": 0, "forwarded": 0, "errors": 0}
defprocess_packet(self, raw_data: bytes, rssi: int):
"""Parse and forward a sensor packet."""iflen(raw_data) != PACKET_SIZE:
self.stats["errors"] += 1return
fields = struct.unpack(PACKET_FORMAT, raw_data)
node_id, msg_type, seq, temp, hum, batt, uptime = fields
payload = {
"node_id": node_id,
"type": ["data", "heartbeat", "alert"][msg_type],
"sequence": seq,
"temperature": round(temp, 2),
"humidity": round(hum, 2),
"battery_v": round(batt, 2),
"uptime_ms": uptime,
"rssi": rssi,
"gateway_ts": time.time()
}
# Update registrywithself.lock:
self.node_registry[node_id] = {
"last_seen": time.time(),
"battery": batt,
"rssi": rssi,
"packets": self.node_registry.get(node_id, {}).get("packets", 0) + 1
}
# Publish to MQTT
topic = f"sensors/node/{node_id}/data"self.mqtt.publish(topic, json.dumps(payload), qos=1)
self.stats["received"] += 1self.stats["forwarded"] += 1# Battery alertif batt < 3.3:
alert_topic = f"sensors/node/{node_id}/alert"self.mqtt.publish(alert_topic, json.dumps({
"type": "low_battery",
"voltage": batt,
"node_id": node_id
}), qos=1)
defget_fleet_status(self) -> dict:
"""Return status of all known nodes."""
now = time.time()
withself.lock:
status = {}
for nid, info inself.node_registry.items():
age = now - info["last_seen"]
status[nid] = {
**info,
"status": "online"if age < 600else"offline",
"last_seen_ago": int(age)
}
return status
Power Management
Power Budget Calculator
Component
Active
Sleep
Duty Cycle
Average
ESP32 (WiFi TX)
240 mA
10 uA
0.1%
0.25 mA
BME280 sensor
1 mA
0.1 uA
0.1%
0.001 mA
Voltage regulator
5 mA
5 mA
100%
5 mA
Total
~5.25 mA
18650 (3000mAh)
~24 days
Deep Sleep Optimization (ESP32)
/* Power-optimized wake cycle */#include<esp_sleep.h>#include<esp_pm.h>#include<driver/rtc_io.h> ./* Use RTC memory to persist across deep sleep */
RTC_DATA_ATTR staticint failed_sends = 0;
RTC_DATA_ATTR staticfloat last_temp = 0;
/* Adaptive sleep: longer intervals when data is stable */staticuint64_tcalculate_sleep_duration(float current_temp) {
float delta = fabsf(current_temp - last_temp);
last_temp = current_temp;
if (delta > 2.0f) return60 * 1000000ULL; /* 1 min if changing fast */if (delta > 0.5f) return300 * 1000000ULL; /* 5 min if moderate */return900 * 1000000ULL; /* 15 min if stable */
}
/* Disable unused peripherals before sleep */staticvoidprepare_for_sleep(void) {
esp_wifi_stop();
adc_power_release();
/* Hold GPIO states during sleep if needed */
rtc_gpio_hold_en(GPIO_NUM_25); /* Keep LED off *//* Configure wake sources */
esp_sleep_enable_timer_wakeup(calculate_sleep_duration(last_temp));
esp_sleep_enable_ext0_wakeup(GPIO_NUM_33, 0); /* Wake on button press *//* Isolate unused GPIO to prevent leakage */
esp_sleep_pd_config(ESP_PD_DOMAIN_RTC_PERIPH, ESP_PD_OPTION_OFF);
}
Solar Power Sizing
defcalculate_solar_panel(
avg_current_ma: float,
battery_mah: float,
peak_sun_hours: float = 4.0,
panel_efficiency: float = 0.7,
days_autonomy: int = 3) -> dict:
"""Calculate minimum solar panel wattage for a sensor node."""
daily_consumption_mah = avg_current_ma * 24
daily_consumption_wh = daily_consumption_mah * 3.7 / 1000# Assuming 3.7V LiPo# Panel must generate enough for daily use + charge losses
required_generation_wh = daily_consumption_wh / panel_efficiency
panel_watts = required_generation_wh / peak_sun_hours
# Battery must last through autonomy period
min_battery_mah = daily_consumption_mah * days_autonomy
return {
"daily_consumption_mah": round(daily_consumption_mah, 1),
"daily_consumption_wh": round(daily_consumption_wh, 3),
"min_panel_watts": round(panel_watts, 2),
"recommended_panel_watts": round(panel_watts * 1.5, 2), # 50% margin"min_battery_mah": round(min_battery_mah),
"battery_adequate": battery_mah >= min_battery_mah
}
Data Pipeline Architecture
Edge Processing Pattern
classEdgeProcessor:
"""Process sensor data locally before transmitting."""def__init__(self, window_size: int = 12):
self.window = []
self.window_size = window_size
self.last_sent = Noneself.threshold = 1.0# Only send if delta > thresholddefadd_reading(self, value: float) -> dict | None:
"""Add reading, return packet only if transmission warranted."""self.window.append(value)
iflen(self.window) > self.window_size:
self.window.pop(0)
iflen(self.window) < self.window_size:
returnNone# Still collecting
stats = {
"mean": sum(self.window) / len(self.window),
"min": min(self.window),
"max": max(self.window),
"range": max(self.window) - min(self.window),
"samples": len(self.window)
}
# Only transmit on significant changeifself.last_sent isnotNone:
ifabs(stats["mean"] - self.last_sent) < self.threshold:
returnNoneself.last_sent = stats["mean"]
self.window.clear()
return stats
Time-Series Storage (InfluxDB Pattern)
from influxdb_client import InfluxDBClient, Point, WritePrecision
from datetime import datetime
classSensorStorage:
def__init__(self, url: str, token: str, org: str, bucket: str):
self.client = InfluxDBClient(url=url, token=token, org=org)
self.write_api = self.client.write_api()
self.query_api = self.client.query_api()
self.bucket = bucket
self.org = org
defstore_reading(self, node_id: int, measurement: str,
fields: dict, tags: dict = None):
point = Point(measurement)
point.tag("node_id", str(node_id))
if tags:
for k, v in tags.items():
point.tag(k, str(v))
for k, v in fields.items():
point.field(k, float(v))
point.time(datetime.utcnow(), WritePrecision.MS)
self.write_api.write(bucket=self.bucket, record=point)
defquery_node_history(self, node_id: int, hours: int = 24) -> list:
query = f'''
from(bucket: "{self.bucket}")
|> range(start: -{hours}h)
|> filter(fn: (r) => r["node_id"] == "{node_id}")
|> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
|> yield(name: "mean")
'''
result = self.query_api.query(query, org=self.org)
return [
{"time": record.get_time(), "value": record.get_value()}
for table in result for record in table.records
]
Fleet Management
Node Health Monitoring
import time
from dataclasses import dataclass, field
from typing importOptional@dataclassclassNodeHealth:
node_id: int
last_seen: float = 0
battery_voltage: float = 0
rssi: int = 0
packet_count: int = 0
error_count: int = 0
expected_interval: int = 300# seconds
firmware_version: str = "" @propertydefis_online(self) -> bool:
return (time.time() - self.last_seen) < (self.expected_interval * 2)
@propertydefbattery_percent(self) -> int:
# LiPo discharge curve approximationifself.battery_voltage >= 4.2: return100ifself.battery_voltage <= 3.0: return0returnint((self.battery_voltage - 3.0) / 1.2 * 100)
@propertydefpacket_loss_rate(self) -> float:
ifself.packet_count == 0: return0returnself.error_count / (self.packet_count + self.error_count)
classFleetManager:
def__init__(self):
self.nodes: dict[int, NodeHealth] = {}
defupdate_node(self, node_id: int, **kwargs):
if node_id notinself.nodes:
self.nodes[node_id] = NodeHealth(node_id=node_id)
node = self.nodes[node_id]
for key, value in kwargs.items():
ifhasattr(node, key):
setattr(node, key, value)
node.last_seen = time.time()
node.packet_count += 1defget_alerts(self) -> list[dict]:
alerts = []
for nid, node inself.nodes.items():
ifnot node.is_online:
alerts.append({"node": nid, "type": "offline",
"since": node.last_seen})
if node.battery_percent < 20:
alerts.append({"node": nid, "type": "low_battery",
"percent": node.battery_percent})
if node.packet_loss_rate > 0.1:
alerts.append({"node": nid, "type": "high_packet_loss",
"rate": node.packet_loss_rate})
return alerts
Common Pitfalls
Mistake
Impact
Solution
No packet sequencing
Undetected data loss
Include sequence numbers
WiFi for battery nodes
Days instead of months
Use ESP-NOW, BLE, or LoRa
No local buffering
Data loss on connectivity gaps
NVS/flash ring buffer
Fixed sample rates
Wasted power on stable data
Adaptive sampling
No OTA update path
Manual firmware updates forever
Plan OTA from day one
Ignoring clock drift
Misaligned time-series data
NTP sync at gateway, relative timestamps at nodes
No encryption
Data interception, spoofing
AES-128 at minimum for ESP-NOW
Exercises
Star Network: Deploy 3 ESP32 sensor nodes reporting temperature to one gateway node via ESP-NOW, forwarding to serial console
Power Profiler: Measure actual current draw of a sensor node across wake/sleep cycles, compare to calculated budget
Adaptive Sampling: Implement edge processing that increases sample rate when readings change rapidly
Fleet Dashboard: Build a web dashboard showing real-time status, battery levels, and alerts for 5+ simulated nodes
Resilient Pipeline: Add NVS buffering to sensor nodes so data survives gateway outages, with automatic catch-up on reconnection
Process
Gather information. Ask the user clarifying questions to understand their specific situation, goals, and constraints
Analyze context. Review the information provided and identify key factors relevant to iot sensor network
Develop recommendations. Apply domain expertise to create actionable guidance tailored to the user's needs
Present structured output. Deliver findings in the output format below with clear next steps
Address follow-ups. Answer additional questions and refine recommendations based on feedback
