Search the docs, videos, blog, apps, teams, skills and release notes.
At a glance
What it is
Guides design and deployment of mesh sensor networks including data collection, power management, communication protocols, and fleet management.
A skill is a written procedure a bot loads when a job calls for it. This one is a single file, SKILL.md, and the whole file is on this page.
The whole file, exactly as it downloads. Read it before you rely on it.
SKILL.md
535 lines · 2,163 words
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.
Use this skill when:
User asks about iot sensor network techniques or best practices
User needs guidance on iot sensor network concepts
Guides design and deployment of mesh sensor networks including data collection, power management, communication protocols, and fleet management.
Nothing. It ships with Murage, which is open source. Your bots run on the AI plan you already have.
Yes. Open it in Settings → Skills to read, edit or duplicate it. Skill Guard scans it again after every edit.
Download the app, connect the AI you already pay for, and tell your Chief of Staff what needs doing. Plan on about ten minutes from install to a working team.
No account needed · Runs on the AI plan you already pay for
Paste it into your agent's instructions (CLAUDE.md, AGENTS.md or a custom GPT), or add it to a bot in Murage.
Ask Ember for the bot, then switch the skill on in its settings.
Step 1
Send your Chief of Staff this message. It proposes the bot and its role first, and creates it when you say go.
Send this to Ember
“Create a bot that uses the IoT Sensor Network skill, and show me the plan first.”
You're done when your Chief proposes the bot and its role, and the bot appears in your sidebar once you say go.
Step 2
You're done when you see the bot’s skills and the library.
Step 3
You're done when the skill shows as on for that bot.
Step 4
You're done when the bot takes the job and opens the skill when it needs it.
Prefer to do it yourself? Choose + in the sidebar, then New Bot.Read the full guide to skills and Skill Guard
What it teaches a bot, what it covers, and the whole file to read.
The principle
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.
What it covers
When to Use
Network Topology Selection
ESP-NOW Mesh Network
Power Management
Data Pipeline Architecture
Fleet Management
Common Pitfalls
Exercises
Process
Output Format
Edge Cases
Example
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
| Topology | Range | Scalability | Power | Complexity | Reliability | Best For |
|---|---|---|---|---|---|---|
| Star | Short | Low (<20) | Low | Simple | Single point failure | Small indoor |
| Mesh | Extended | High (100+) | Medium | Complex | Self-healing | Large area |
| Tree | Medium | Medium | Medium | Medium | Branch failure | Hierarchical |
| Star-of-Stars | Extended | High | Low-Med | Medium | Gateway redundancy | Multi-room |
| Protocol | Range | Data Rate | Power | Topology | License | Nodes |
|---|---|---|---|---|---|---|
| WiFi | 50m | 54+ Mbps | High | Star | ISM | ~32 |
| BLE Mesh | 30m | 2 Mbps | Very Low | Mesh | ISM | 32K |
| Zigbee | 100m | 250 kbps | Low | Mesh/Star | ISM | 65K |
| Z-Wave | 100m | 100 kbps | Low | Mesh | Licensed | 232 |
| LoRa | 15km | 50 kbps | Very Low | Star | ISM | 1000s |
| Thread | 30m | 250 kbps | Low | Mesh | ISM | 250+ |
| ESP-NOW | 200m | 1 Mbps | Low | Star/Mesh | ISM | 20 |
/* sensor_node.c - Battery-powered sensor node using ESP-NOW */
#include <esp_now.h>
#include <esp_wifi.h>
#include <esp_sleep.h>
#include <nvs_flash.h>
#include <string.h>
#define GATEWAY_MAC {0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF}
#define SLEEP_DURATION_US (300 * 1000000ULL) /* 5 minutes */
#define SENSOR_PIN ADC1_CHANNEL_0
#define BATTERY_PIN ADC1_CHANNEL_3
typedef struct __attribute__((packed)) {
uint8_t node_id;
uint8_t msg_type; /* 0=data, 1=heartbeat, 2=alert */
uint16_t sequence;
float temperature;
float humidity;
float battery_v;
uint32_t uptime_ms;
} sensor_packet_t;
static uint16_t seq_num = 0;
static uint8_t gateway_mac[] = GATEWAY_MAC;
RTC_DATA_ATTR static uint16_t boot_count = 0;
static void on_data_sent(const uint8_t *mac, esp_now_send_status_t status) {
if (status != ESP_NOW_SEND_SUCCESS) {
/* Store in NVS for retry on next wake */
store_failed_packet();
}
}
static void init_espnow(void) {
wifi_init_config_t cfg = WIFI_INIT_CONFIG_DEFAULT();
esp_wifi_init(&cfg);
esp_wifi_set_mode(WIFI_MODE_STA);
esp_wifi_start();
esp_now_init();
esp_now_register_send_cb(on_data_sent);
esp_now_peer_info_t peer = {
.channel = 1,
.encrypt = false
};
memcpy(peer.peer_addr, gateway_mac, 6);
esp_now_add_peer(&peer);
}
static float read_battery_voltage(void) {
/* Voltage divider: 100K/100K, Vmax = 4.2V -> ADC max ~2.1V */
int raw = adc1_get_raw(BATTERY_PIN);
return (raw / 4095.0f) * 3.3f * 2.0f;
}
void app_main(void) {
boot_count++;
init_espnow();
sensor_packet_t pkt = {
.node_id = CONFIG_NODE_ID,
.msg_type = 0,
.sequence = boot_count,
.temperature = read_temperature_sensor(),
.humidity = read_humidity_sensor(),
.battery_v = read_battery_voltage(),
.uptime_ms = boot_count * (SLEEP_DURATION_US / 1000)
};
/* Retry failed packets from NVS first */
retry_stored_packets();
esp_now_send(gateway_mac, (uint8_t *)&pkt, sizeof(pkt));
vTaskDelay(pdMS_TO_TICKS(100)); /* Wait for send callback */
/* Deep sleep */
esp_wifi_stop();
esp_deep_sleep(SLEEP_DURATION_US);
}#!/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)
class SensorGateway:
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}
def process_packet(self, raw_data: bytes, rssi: int):
"""Parse and forward a sensor packet."""
if len(raw_data) != PACKET_SIZE:
self.stats["errors"] += 1
return
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 registry
with self.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"] += 1
self.stats["forwarded"] += 1
# Battery alert
if 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)
def get_fleet_status(self) -> dict:
"""Return status of all known nodes."""
now = time.time()
with self.lock:
status = {}
for nid, info in self.node_registry.items():
age = now - info["last_seen"]
status[nid] = {
**info,
"status": "online" if age < 600 else "offline",
"last_seen_ago": int(age)
}
return status| 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 |
/* 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 static int failed_sends = 0;
RTC_DATA_ATTR static float last_temp = 0;
/* Adaptive sleep: longer intervals when data is stable */
static uint64_t calculate_sleep_duration(float current_temp) {
float delta = fabsf(current_temp - last_temp);
last_temp = current_temp;
if (delta > 2.0f) return 60 * 1000000ULL; /* 1 min if changing fast */
if (delta > 0.5f) return 300 * 1000000ULL; /* 5 min if moderate */
return 900 * 1000000ULL; /* 15 min if stable */
}
/* Disable unused peripherals before sleep */
static void prepare_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);
}def calculate_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
}class EdgeProcessor:
"""Process sensor data locally before transmitting."""
def __init__(self, window_size: int = 12):
self.window = []
self.window_size = window_size
self.last_sent = None
self.threshold = 1.0 # Only send if delta > threshold
def add_reading(self, value: float) -> dict | None:
"""Add reading, return packet only if transmission warranted."""
self.window.append(value)
if len(self.window) > self.window_size:
self.window.pop(0)
if len(self.window) < self.window_size:
return None # 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 change
if self.last_sent is not None:
if abs(stats["mean"] - self.last_sent) < self.threshold:
return None
self.last_sent = stats["mean"]
self.window.clear()
return statsfrom influxdb_client import InfluxDBClient, Point, WritePrecision
from datetime import datetime
class SensorStorage:
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
def store_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)
def query_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
]import time
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class NodeHealth:
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 = ""
@property
def is_online(self) -> bool:
return (time.time() - self.last_seen) < (self.expected_interval * 2)
@property
def battery_percent(self) -> int:
# LiPo discharge curve approximation
if self.battery_voltage >= 4.2: return 100
if self.battery_voltage <= 3.0: return 0
return int((self.battery_voltage - 3.0) / 1.2 * 100)
@property
def packet_loss_rate(self) -> float:
if self.packet_count == 0: return 0
return self.error_count / (self.packet_count + self.error_count)
class FleetManager:
def __init__(self):
self.nodes: dict[int, NodeHealth] = {}
def update_node(self, node_id: int, **kwargs):
if node_id not in self.nodes:
self.nodes[node_id] = NodeHealth(node_id=node_id)
node = self.nodes[node_id]
for key, value in kwargs.items():
if hasattr(node, key):
setattr(node, key, value)
node.last_seen = time.time()
node.packet_count += 1
def get_alerts(self) -> list[dict]:
alerts = []
for nid, node in self.nodes.items():
if not 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| 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 |
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
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
## Iot Sensor Network Analysis
### Assessment
[Key findings and observations]
### Recommendations
1. [Primary recommendation]
2. [Secondary recommendation]
3. [Additional suggestions]
### Action Items
- [ ] [First action step]
- [ ] [Second action step]
- [ ] [Follow-up task]Incomplete information: Ask clarifying questions before proceeding with recommendations
Conflicting requirements: Prioritize the most critical constraint and note trade-offs
Out of scope requests: Redirect to appropriate specialized skill or professional resource
Beginner vs advanced: Adjust depth and terminology based on user's experience level
Input: "Help me with iot sensor network for my current situation"
Output:
Based on your situation, here is a structured approach to iot sensor network:
Assessment: Evaluate your current state and identify key areas for improvement
Strategy: Develop a targeted plan based on best practices
Implementation: Execute the plan with specific, measurable steps
Review: Monitor progress and adjust as needed
SKILL.md
535 lines · 2,163 words
The text of SKILL.md is shown as it downloads, under the Apache-2.0 license, by Ferrox Labs. The license is the one in the file's own frontmatter.
Your new bot
IoT Sensor Network bot
Waiting in your sidebar once you say go.
No account needed · Runs on the AI plan you already pay for