Discovers all Shelly devices on IoT networks via HTTP API and configures them to use the local HA NTP server (ntp.renner.uno). Supports: - Gen1 devices (via /settings endpoint) - Gen2/Gen3/Gen4 devices (via /rpc/Sys.SetConfig endpoint) Usage: ansible-playbook plays/configure-shelly-ntp.yml 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
68 lines
2.0 KiB
Python
Executable File
68 lines
2.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Discover Shelly devices on the network via HTTP API
|
|
Returns JSON array of discovered devices
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
import ipaddress
|
|
import concurrent.futures
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
def check_shelly(ip):
|
|
"""Check if IP is a Shelly device and return info"""
|
|
try:
|
|
url = f"http://{ip}/shelly"
|
|
req = urllib.request.Request(url, headers={'User-Agent': 'Ansible-Shelly-Discovery'})
|
|
with urllib.request.urlopen(req, timeout=2) as response:
|
|
data = json.loads(response.read().decode())
|
|
# Gen1 devices don't have 'gen' field, default to 1
|
|
gen = data.get('gen', 1)
|
|
if gen is None:
|
|
gen = 1
|
|
return {
|
|
'ip': str(ip),
|
|
'type': data.get('type', 'unknown'),
|
|
'mac': data.get('mac', ''),
|
|
'gen': gen,
|
|
'name': data.get('name', data.get('id', str(ip))),
|
|
'fw_id': data.get('fw_id', data.get('fw', 'unknown'))
|
|
}
|
|
except (urllib.error.URLError, urllib.error.HTTPError, OSError, json.JSONDecodeError):
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
print("Usage: discover_shelly.py <network1> [network2] ...")
|
|
print("Example: discover_shelly.py 10.2.30.0/24 10.2.20.0/24")
|
|
sys.exit(1)
|
|
|
|
networks = sys.argv[1:]
|
|
devices = []
|
|
|
|
# Collect all IPs to scan
|
|
all_ips = []
|
|
for network in networks:
|
|
try:
|
|
all_ips.extend(ipaddress.IPv4Network(network).hosts())
|
|
except ValueError as e:
|
|
print(f"Invalid network: {network} - {e}", file=sys.stderr)
|
|
continue
|
|
|
|
# Scan in parallel
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
|
|
results = executor.map(check_shelly, all_ips)
|
|
for result in results:
|
|
if result:
|
|
devices.append(result)
|
|
|
|
# Output JSON
|
|
print(json.dumps(devices, indent=2))
|
|
|
|
if __name__ == "__main__":
|
|
main()
|