212 lines
9.3 KiB
Python
212 lines
9.3 KiB
Python
"""SonoffDevice model – represents an individual Sonoff sub-device
|
||
belonging to a Sonoff eWeLink Gateway board."""
|
||
import json
|
||
from datetime import datetime
|
||
from app import db
|
||
|
||
|
||
# -----------------------------------------------------------------
|
||
# Device-type → human label + channel count heuristics
|
||
# Based on UIIDs from SonoffLAN devices.py
|
||
# -----------------------------------------------------------------
|
||
UIID_INFO: dict[int, dict] = {
|
||
1: {"label": "Basic / BASICR2", "channels": 1, "kind": "switch"},
|
||
2: {"label": "Dual Switch", "channels": 2, "kind": "switch"},
|
||
3: {"label": "Triple Switch", "channels": 3, "kind": "switch"},
|
||
4: {"label": "Quad Switch (4CH)", "channels": 4, "kind": "switch"},
|
||
5: {"label": "POW (Power Meter)", "channels": 1, "kind": "switch", "has_power": True},
|
||
6: {"label": "Basic (3rd-party)", "channels": 1, "kind": "switch"},
|
||
7: {"label": "T1 2C", "channels": 2, "kind": "switch"},
|
||
8: {"label": "T1 3C", "channels": 3, "kind": "switch"},
|
||
9: {"label": "T1 4C", "channels": 4, "kind": "switch"},
|
||
14: {"label": "Basic (3rd-party)", "channels": 1, "kind": "switch"},
|
||
15: {"label": "Sonoff TH (Thermostat)", "channels": 1, "kind": "sensor",
|
||
"has_temp": True, "has_humidity": True},
|
||
22: {"label": "Slampher B1", "channels": 1, "kind": "light"},
|
||
24: {"label": "GSM Socket", "channels": 1, "kind": "switch"},
|
||
28: {"label": "RF Bridge 433", "channels": 0, "kind": "remote"},
|
||
29: {"label": "Dual Switch (2CH)", "channels": 2, "kind": "switch"},
|
||
30: {"label": "Triple Switch (3CH)", "channels": 3, "kind": "switch"},
|
||
31: {"label": "Quad Switch (4CH)", "channels": 4, "kind": "switch"},
|
||
32: {"label": "POWR2 (Power Meter)", "channels": 1, "kind": "switch", "has_power": True},
|
||
34: {"label": "iFan02/iFan03", "channels": 1, "kind": "fan"},
|
||
44: {"label": "D1 Dimmer", "channels": 1, "kind": "light"},
|
||
57: {"label": "Micro USB", "channels": 1, "kind": "switch"},
|
||
59: {"label": "LED Light Strip L1", "channels": 1, "kind": "light"},
|
||
77: {"label": "Micro Switch", "channels": 1, "kind": "switch"},
|
||
78: {"label": "Micro Switch 2CH", "channels": 2, "kind": "switch"},
|
||
81: {"label": "Switch (compact)", "channels": 1, "kind": "switch"},
|
||
82: {"label": "Switch 2CH (compact)", "channels": 2, "kind": "switch"},
|
||
83: {"label": "Switch 3CH (compact)", "channels": 3, "kind": "switch"},
|
||
84: {"label": "Switch 4CH (compact)", "channels": 4, "kind": "switch"},
|
||
102: {"label": "DW2 Door/Window Sensor", "channels": 0, "kind": "sensor",
|
||
"has_door": True},
|
||
107: {"label": "MINIR3", "channels": 1, "kind": "switch"},
|
||
109: {"label": "DUAL R3", "channels": 2, "kind": "switch"},
|
||
126: {"label": "DUAL R3 Lite", "channels": 2, "kind": "switch"},
|
||
130: {"label": "POWR3 (Power Meter)", "channels": 1, "kind": "switch", "has_power": True},
|
||
133: {"label": "ZBMINI (ZigBee)", "channels": 1, "kind": "switch"},
|
||
136: {"label": "B05-BL Light", "channels": 1, "kind": "light"},
|
||
138: {"label": "MINIR4", "channels": 1, "kind": "switch"},
|
||
160: {"label": "SwitchMan M5-1C", "channels": 1, "kind": "switch"},
|
||
161: {"label": "SwitchMan M5-2C", "channels": 2, "kind": "switch"},
|
||
162: {"label": "SwitchMan M5-3C", "channels": 3, "kind": "switch"},
|
||
190: {"label": "NSPanel", "channels": 2, "kind": "switch"},
|
||
195: {"label": "NSPanel Pro", "channels": 1, "kind": "switch"},
|
||
}
|
||
|
||
|
||
def _uiid_info(uiid: int) -> dict:
|
||
return UIID_INFO.get(uiid, {"label": f"Sonoff UIID-{uiid}", "channels": 1, "kind": "switch"})
|
||
|
||
|
||
class SonoffDevice(db.Model):
|
||
__tablename__ = "sonoff_devices"
|
||
|
||
id = db.Column(db.Integer, primary_key=True)
|
||
board_id = db.Column(db.Integer, db.ForeignKey("boards.id"), nullable=False)
|
||
|
||
# eWeLink device ID (10-char string, e.g. "1000abcdef")
|
||
device_id = db.Column(db.String(20), nullable=False)
|
||
# User-visible name
|
||
name = db.Column(db.String(128), default="")
|
||
# LAN address (host:port) — populated when device reachable on LAN
|
||
ip_address = db.Column(db.String(64), default="")
|
||
port = db.Column(db.Integer, default=8081)
|
||
# eWeLink UIID (determines device capabilities/type)
|
||
uiid = db.Column(db.Integer, default=1)
|
||
# Number of controllable channels
|
||
num_channels = db.Column(db.Integer, default=1)
|
||
# Firmware/model string
|
||
model = db.Column(db.String(64), default="")
|
||
firmware = db.Column(db.String(32), default="")
|
||
# AES encryption key (needed for non-DIY LAN control)
|
||
device_key = db.Column(db.String(128), default="")
|
||
# eWeLink device apikey (owner's apikey)
|
||
api_key = db.Column(db.String(128), default="")
|
||
# Current device state as JSON {"switch":"on"} or {"switches":[...]}
|
||
params_json = db.Column(db.Text, default="{}")
|
||
# Whether device is currently online
|
||
is_online = db.Column(db.Boolean, default=False)
|
||
last_seen = db.Column(db.DateTime, nullable=True)
|
||
|
||
board = db.relationship("Board", back_populates="sonoff_devices")
|
||
|
||
# ── param helpers ─────────────────────────────────────────────────────────
|
||
|
||
@property
|
||
def params(self) -> dict:
|
||
try:
|
||
return json.loads(self.params_json or "{}")
|
||
except (ValueError, TypeError):
|
||
return {}
|
||
|
||
@params.setter
|
||
def params(self, value: dict):
|
||
self.params_json = json.dumps(value)
|
||
|
||
# ── channel state helpers ─────────────────────────────────────────────────
|
||
|
||
def get_channel_state(self, channel: int = 0) -> bool:
|
||
"""Return True if channel is ON.
|
||
Detects format from stored params: if 'switches' key is present (or
|
||
num_channels > 1) uses the switches-array format; otherwise uses 'switch'.
|
||
"""
|
||
p = self.params
|
||
uses_switches_fmt = self.num_channels > 1 or "switches" in p
|
||
if uses_switches_fmt:
|
||
switches = p.get("switches", [])
|
||
for s in switches:
|
||
if s.get("outlet") == channel:
|
||
return s.get("switch") == "on"
|
||
return False
|
||
return p.get("switch") == "on"
|
||
|
||
def all_channel_states(self) -> list[dict]:
|
||
"""Return [{channel, label, state}, ...] for every channel."""
|
||
result = []
|
||
for ch in range(max(self.num_channels, 1)):
|
||
result.append({
|
||
"channel": ch,
|
||
"label": f"Channel {ch + 1}",
|
||
"state": self.get_channel_state(ch),
|
||
})
|
||
return result
|
||
|
||
# ── sensor helpers ────────────────────────────────────────────────────────
|
||
|
||
@property
|
||
def temperature(self) -> float | None:
|
||
p = self.params
|
||
for key in ("temperature", "currentTemperature"):
|
||
if key in p:
|
||
try:
|
||
return float(p[key])
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return None
|
||
|
||
@property
|
||
def humidity(self) -> float | None:
|
||
p = self.params
|
||
for key in ("humidity", "currentHumidity"):
|
||
if key in p:
|
||
try:
|
||
return float(p[key])
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return None
|
||
|
||
@property
|
||
def power(self) -> float | None:
|
||
v = self.params.get("power")
|
||
try:
|
||
return float(v) if v is not None else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@property
|
||
def voltage(self) -> float | None:
|
||
v = self.params.get("voltage")
|
||
try:
|
||
return float(v) if v is not None else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@property
|
||
def current(self) -> float | None:
|
||
v = self.params.get("current")
|
||
try:
|
||
return float(v) if v is not None else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
# ── display helpers ───────────────────────────────────────────────────────
|
||
|
||
@property
|
||
def device_info(self) -> dict:
|
||
return _uiid_info(self.uiid)
|
||
|
||
@property
|
||
def device_label(self) -> str:
|
||
return self.device_info.get("label", f"UIID-{self.uiid}")
|
||
|
||
@property
|
||
def kind(self) -> str:
|
||
return self.device_info.get("kind", "switch")
|
||
|
||
@property
|
||
def has_power_meter(self) -> bool:
|
||
return self.device_info.get("has_power", False)
|
||
|
||
@property
|
||
def has_temperature(self) -> bool:
|
||
return self.device_info.get("has_temp", False)
|
||
|
||
@property
|
||
def has_door_sensor(self) -> bool:
|
||
return self.device_info.get("has_door", False)
|
||
|
||
def __repr__(self):
|
||
return f"<SonoffDevice {self.device_id} {self.name!r}>"
|