From d06cb6d0de3ccf57b3d92dda2a3f5f9914604c60 Mon Sep 17 00:00:00 2001 From: Nova Date: Thu, 12 Feb 2026 01:42:09 +0000 Subject: [PATCH] Initial commit - Tasmota skill for OpenClaw --- SKILL.md | 251 +++++++++++++++++++++++++++++++++++ scripts/tasmota-control.py | 101 ++++++++++++++ scripts/tasmota-discovery.py | 166 +++++++++++++++++++++++ scripts/tasmota-status.py | 57 ++++++++ 4 files changed, 575 insertions(+) create mode 100644 SKILL.md create mode 100755 scripts/tasmota-control.py create mode 100755 scripts/tasmota-discovery.py create mode 100755 scripts/tasmota-status.py diff --git a/SKILL.md b/SKILL.md new file mode 100644 index 0000000..8de7bd7 --- /dev/null +++ b/SKILL.md @@ -0,0 +1,251 @@ +--- +name: tasmota +description: Discover, monitor, and control Tasmota smart home devices on local networks. Use when tasks involve finding Tasmota devices via network scanning, checking device status and power states, controlling devices (on-off, brightness, color), managing device inventory, or any other Tasmota management operations on ESP8266 or ESP32 devices running Tasmota firmware. +--- + +# Tasmota Device Management + +## Overview + +Automated discovery and control of Tasmota-powered smart home devices (ESP8266/ESP32) on local networks. Includes network scanning, status monitoring, power control, dimming, color control, and inventory management. + +## Quick Start + +**Scan network for Tasmota devices:** +```bash +python3 scripts/tasmota-discovery.py +``` + +**Check device status:** +```bash +python3 scripts/tasmota-control.py status 0 +``` + +**Control device:** +```bash +python3 scripts/tasmota-control.py power on|off|toggle +python3 scripts/tasmota-control.py brightness 0-100 +python3 scripts/tasmota-control.py color +``` + +## Discovery + +### Network Scan + +Run a full network scan to find all Tasmota devices: + +```bash +python3 scripts/tasmota-discovery.py +``` + +The script: +1. Performs ping sweep to identify live hosts +2. Scans HTTP (port 80) on live hosts +3. Detects Tasmota via Server header (e.g., "Tasmota/13.1.0") +4. Confirms via JSON API (`/cm?cmnd=status%200`) +5. Retrieves device names, versions, and hardware + +Output includes: +- IP address +- Device friendly name +- Tasmota version +- Hardware platform (ESP8266/ESP32) +- Response time + +### Identification Signals + +Tasmota devices are identified by: +- **Server header:** `Tasmota/ ()` +- **JSON API response:** `/cm?cmnd=status%200` returns structured status +- **Content keywords:** "Tasmota" in HTML + +## Device Control + +### Power Control + +Toggle or set power state: + +```bash +# Toggle +python3 scripts/tasmota-control.py power toggle + +# On/Off +python3 scripts/tasmota-control.py power on +python3 scripts/tasmota-control.py power off +``` + +### Brightness (Dimmers) + +Set brightness level (0-100): + +```bash +python3 scripts/tasmota-control.py brightness 50 +``` + +Works on devices with Dimmer support (check StatusSTS). + +### Color (RGB Lights) + +Set RGB color (hex or comma-separated): + +```bash +# Hex format +python3 scripts/tasmota-control.py color FF0000 # Red +python3 scripts/tasmota-control.py color 00FF00 # Green + +# RGB comma format +python3 scripts/tasmota-control.py color 255,0,0 +``` + +Works on devices with RGB support (AiYaTo-RGBCW, etc.). + +## Status Queries + +### Device Status + +Retrieve status information: + +```bash +# Basic status +python3 scripts/tasmota-control.py status 0 + +# All statuses +python3 scripts/tasmota-control.py status all +``` + +**Status codes:** +- `0 = Status` - Device info, friendly name, power state +- `1 = StatusPRM` - Parameters, uptime, MAC +- `2 = StatusFWR` - Firmware version, hardware +- `3 = StatusLOG` - Log settings +- `4 = StatusNET` - Network config (IP, gateway, WiFi) +- `5 = StatusMQT` - MQTT configuration +- `9 = StatusTIM` - Time, timezone, sunrise/sunset + +### Key Status Fields + +**StatusSTS (Status 0):** +- `POWER` - Current state (ON/OFF) +- `Dimmer` - Brightness level (0-100) +- `Wifi.RSSI` - Signal strength +- `Wifi.SSId` - Connected WiFi network + +**StatusNET:** +- `IPAddress` - Device IP +- `Hostname` - mDNS hostname +- `Mac` - MAC address + +## Bulk Operations + +### Get All Device Status + +```bash +python3 scripts/tasmota-status.py +``` + +Iterates through inventory file and shows power state for all devices. + +### Managing Inventory + +Devices are tracked in a CSV inventory file. Format: + +``` +IP Address,Device Name,Version,Hardware,Response Time (ms) +192.168.1.116,Office Hall Light,13.1.0,ESP8266EX,53 +``` + +After discovery, save output to inventory file for batch operations. + +## Common Tasks + +### Finding Labeled Devices + +```bash +# Scan and grep for specific device names +python3 scripts/tasmota-discovery.py | grep "Kitchen" +python3 scripts/tasmota-discovery.py | grep "Bulb" +``` + +### Checking All Lights + +```bash +# Get status of all devices +python3 scripts/tasmota-status.py +``` + +### Power Cycle a Device + +```bash +# Off, wait 2s, on +python3 scripts/tasmota-control.py 192.168.1.116 power off +sleep 2 +python3 scripts/tasmota-control.py 192.168.1.116 power on +``` + +## Tasmota API Reference + +### Command Format + +``` +http:///cm?cmnd= +``` + +### Common Commands + +| Command | Description | +|---------|-------------| +| `Power` | Toggle power | +| `Power ON` | Turn on | +| `Power OFF` | Turn off | +| `Power TOGGLE` | Toggle state | +| `Status 0` | Device status | +| `Status 4` | Network status | +| `Dimmer <0-100>` | Set brightness | +| `Color ` | Set RGB color | +| `Fade ` | Enable fade effects | + +## Troubleshooting + +### Device Not Found + +- Verify device is on same subnet +- Check device has HTTP server enabled (Webserver 2 in config) +- Ensure device is powered on and connected to WiFi +- Try direct HTTP request: `curl http:///cm?cmnd=Status%200` + +### Timeout Errors + +- Device may be power saving (WiFi sleep) +- Network latency or packet loss +- Check device uptime for recent restarts + +### Unknown Power State + +Some devices (BLE gateways, sensors) don't have power control. Check capability in StatusSTS. + +## Network Configuration + +Tasmota devices typically: +- Connect on WiFi channel 1-11 +- Use DHCP (check StatusNET for current IP) +- May respond to mDNS (hostname patterns: tasmota-XXXXXX) +- Use HTTP on port 80 (standard) + +## Best Practices + +- Scan during network maintenance windows (avoid peak usage) +- Cache inventory file to avoid repeated scans +- Use device friendly names for easier identification +- Set static IPs for critical devices (via Tasmota web UI or DHCP reservations) +- Group devices by location/function in inventory comments + +## Resources + +### scripts/tasmota-discovery.py +Network scanner that finds live hosts and identifies Tasmota devices via HTTP and JSON API. + +### scripts/tasmota-control.py +Device controller supporting power, brightness, color, and status queries via Tasmota JSON API. + +### scripts/tasmota-status.py +Bulk status checker that queries all devices in inventory and displays power states. \ No newline at end of file diff --git a/scripts/tasmota-control.py b/scripts/tasmota-control.py new file mode 100755 index 0000000..ab1cf6d --- /dev/null +++ b/scripts/tasmota-control.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +""" +Tasmota Device Controller +Control Tasmota devices via HTTP JSON API +""" + +import sys +import json +from urllib.request import urlopen, Request +from urllib.error import URLError, HTTPError + +def tasmota_command(ip, command, timeout=5): + """Send command to Tasmota device""" + try: + # Tasmota JSON API format: /cm?cmnd=COMMAND + cmd = command.replace(' ', '%20') + url = f"http://{ip}/cm?cmnd={cmd}" + + req = Request(url, headers={'User-Agent': 'Tasmota-Controller'}) + with urlopen(req, timeout=timeout) as response: + data = response.read() + return json.loads(data.decode('utf-8')) + except Exception as e: + return {"ERROR": str(e)} + +def toggle_power(ip): + """Toggle power on/off""" + return tasmota_command(ip, "Power") + +def set_power(ip, state): + """Set power state (ON/OFF/TOGGLE)""" + return tasmota_command(ip, f"Power {state.upper()}") + +def get_status(ip, status_type="0"): + """Get device status + status_type codes: + 0 = Status + 1 = StatusPRM (Parameters) + 2 = StatusFWR (Firmware) + 3 = StatusLOG (Log) + 4 = StatusNET (Network) + 5 = StatusMQT (MQTT) + 9 = StatusTIM (Time) + """ + return tasmota_command(ip, f"Status {status_type}") + +def set_brightness(ip, level): + """Set brightness (0-100)""" + return tasmota_command(ip, f"Dimmer {level}") + +def set_color(ip, color): + """Set RGB color (hex or r,g,b)""" + return tasmota_command(ip, f"Color {color}") + +def main(): + if len(sys.argv) < 3: + print("Usage:") + print(" python3 tasmota-control.py power [on|off|toggle]") + print(" python3 tasmota-control.py status [0-9|all]") + print(" python3 tasmota-control.py brightness <0-100>") + print(" python3 tasmota-control.py color ") + sys.exit(1) + + ip = sys.argv[1] + cmd = sys.argv[2].lower() + + if cmd == "power": + if len(sys.argv) > 3: + result = set_power(ip, sys.argv[3]) + else: + result = toggle_power(ip) + elif cmd == "status": + if len(sys.argv) > 3: + status_type = sys.argv[3] + if status_type == "all": + # Get all statuses + result = {"statuses": {}} + for st in ["0", "1", "2", "3", "4", "5", "9"]: + result["statuses"][st] = get_status(ip, st) + else: + result = get_status(ip, status_type) + else: + result = get_status(ip, "0") + elif cmd == "brightness": + if len(sys.argv) < 4: + print("Error: brightness requires level (0-100)") + sys.exit(1) + result = set_brightness(ip, sys.argv[3]) + elif cmd == "color": + if len(sys.argv) < 4: + print("Error: color requires hex value (e.g., FF0000)") + sys.exit(1) + result = set_color(ip, sys.argv[3]) + else: + # Raw command + result = tasmota_command(ip, ' '.join(sys.argv[2:])) + + print(json.dumps(result, indent=2)) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/scripts/tasmota-discovery.py b/scripts/tasmota-discovery.py new file mode 100755 index 0000000..23d56d6 --- /dev/null +++ b/scripts/tasmota-discovery.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 +""" +Tasmota Device Discovery +Scans network for Tasmota devices by: +1. HTTP-based detection (port 80) +2. Checking for Tasmota JSON API response +3. Looking for Tasmota user-agent in responses +""" + +import socket +import subprocess +import concurrent.futures +from urllib.request import urlopen, Request +from urllib.error import URLError, HTTPError +import json +import time + +# Network configuration +SUBNET = "192.168.1.0/24" +TIMEOUT = 2 # seconds + +def get_local_ip(): + """Get local IP to determine subnet""" + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("8.8.8.8", 80)) + ip = s.getsockname()[0] + s.close() + return ip + except: + return "192.168.1.148" + +def get_subnet_ips(): + """Generate list of IPs in subnet""" + local_ip = get_local_ip() + parts = local_ip.split('.') + base = f"{parts[0]}.{parts[1]}.{parts[2]}" + return [f"{base}.{i}" for i in range(1, 255)] + +def check_http(ip): + """Check if IP has HTTP service that might be Tasmota""" + try: + # Try basic HTTP request first + req = Request(f"http://{ip}/", headers={'User-Agent': 'Tasmota-Scanner'}) + start = time.time() + with urlopen(req, timeout=TIMEOUT) as response: + elapsed = time.time() - start + data = response.read(5000).decode('utf-8', errors='ignore') + headers = dict(response.headers) + + # Check for Tasmota indicators + is_tasmota = False + evidence = [] + + # Check for Tasmota-specific headers/content + if 'tasmota' in headers.get('Server', '').lower(): + is_tasmota = True + evidence.append(f"Server header: {headers.get('Server')}") + + # Check content for Tasmota patterns + if 'tasmota' in data.lower(): + is_tasmota = True + evidence.append("Tasmota keyword in content") + + # Check for known Tasmota page titles/elements + if 'Status' in data and 'tasmota' in data.lower(): + is_tasmota = True + evidence.append("Tasmota status page detected") + + # Try Tasmota JSON API + try: + req_api = Request(f"http://{ip}/cm?cmnd=status%200") + with urlopen(req_api, timeout=1) as api_response: + api_data = api_response.read(1000).decode('utf-8', errors='ignore') + if 'tasmota' in api_data.lower() or 'StatusSNS' in api_data or 'StatusNET' in api_data: + is_tasmota = True + evidence.append("Tasmota JSON API response") + except: + pass + + if is_tasmota: + try: + title = data.split('')[1].split('')[0].strip() + except: + title = "N/A" + + return { + 'ip': ip, + 'port': 80, + 'title': title, + 'server': headers.get('Server', 'Unknown'), + 'evidence': evidence, + 'elapsed': elapsed + } + return None + + except (URLError, HTTPError, socket.timeout, ConnectionResetError): + return None + except Exception as e: + return None + +def ping_scan(ip): + """Quick ping check to see if host is up""" + try: + result = subprocess.run( + ['ping', '-c', '1', '-W', '1', ip], + capture_output=True, + timeout=2 + ) + return result.returncode == 0 + except: + return False + +def discover_tasmota(): + """Main discovery function""" + print(f"šŸ” Scanning network for Tasmota devices...") + print(f"šŸ“ Local subnet: {SUBNET}") + + ips = get_subnet_ips() + print(f"šŸ“” Checking {len(ips)} IP addresses...") + + tasmota_devices = [] + + # First, quick ping sweep to find live hosts + print("\n1ļøāƒ£ Ping sweep (finding live hosts)...") + live_hosts = [] + with concurrent.futures.ThreadPoolExecutor(max_workers=50) as executor: + future_to_ip = {executor.submit(ping_scan, ip): ip for ip in ips} + for future in concurrent.futures.as_completed(future_to_ip): + ip = future_to_ip[future] + if future.result(): + live_hosts.append(ip) + print(f" āœ… {ip}") + + print(f"\nšŸ“ Found {len(live_hosts)} live hosts") + print(f"\n2ļøāƒ£ Scanning for Tasmota (HTTP) on live hosts...") + + # Then check HTTP on live hosts + with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor: + future_to_ip = {executor.submit(check_http, ip): ip for ip in live_hosts} + for future in concurrent.futures.as_completed(future_to_ip): + result = future.result() + if result: + tasmota_devices.append(result) + + return tasmota_devices + +if __name__ == "__main__": + devices = discover_tasmota() + + print(f"\n{'='*60}") + if devices: + print(f"āœ… Found {len(devices)} Tasmota device(s):\n") + for i, device in enumerate(devices, 1): + print(f"šŸ”Œ Device {i}:") + print(f" IP: {device['ip']}:{device['port']}") + print(f" Title: {device['title']}") + print(f" Server: {device['server']}") + print(f" Evidence:") + for ev in device['evidence']: + print(f" ā€¢ {ev}") + print(f" Response time: {device['elapsed']:.3f}s") + print() + else: + print("āŒ No Tasmota devices found") + print(f"{'='*60}") \ No newline at end of file diff --git a/scripts/tasmota-status.py b/scripts/tasmota-status.py new file mode 100755 index 0000000..82b354b --- /dev/null +++ b/scripts/tasmota-status.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +""" +Get summary of all Tasmota devices from inventory +""" + +import subprocess +import json +import time +from pathlib import Path + +CSV_FILE = Path.home() / ".openclaw/workspace/memory/tasmota-inventory.csv" + +def get_device_status(ip): + """Get power status from device""" + try: + result = subprocess.run( + ["python3", Path.home() / ".openclaw/workspace/scripts/tasmota-control.py", ip, "status", "0"], + capture_output=True, + timeout=3 + ) + if result.returncode == 0: + data = json.loads(result.stdout) + if "StatusSTS" in data: + return data["StatusSTS"].get("POWER", "UNKNOWN") + if "Status" in data: + return data["Status"].get("Power", "ON" if data["Status"].get("Power") == 1 else "OFF") + return "UNKNOWN" + except: + return "TIMEOUT" + +print("šŸ” Tasmota Device Summary") +print("="*80) + +devices = [] +with open(CSV_FILE) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + parts = line.split(',') + if len(parts) >= 3 and parts[0].strip().startswith('192'): + ip = parts[0].strip() + name = parts[1].strip().replace(' - Main Menu', '') + version = parts[2].strip() + devices.append((ip, name, version)) + +# Quick check of first few devices +print(f"Found {len(devices)} devices\n") +print(f"{'IP Address':<18} {'Device Name':<30} {'Version':<12} {'Power'}") +print("-"*80) + +for ip, name, version in devices: + power = get_device_status(ip) + print(f"{ip:<18} {name[:28]:<30} {version:<12} {power}") + time.sleep(0.1) # Don't overwhelm the network + +print("="*80) \ No newline at end of file