Files
location_managemet/app/models/tuya_device.py
ske087 1e89323035 Add Tuya Cloud integration; 2-step add-board wizard; DP value formatting
- Tuya Cloud Gateway driver (tuya-device-sharing-sdk):
  - QR-code auth flow: user-provided user_code, terminal_id/endpoint
    returned by Tuya login_result (mirrors HA implementation)
  - Device sync, toggle DP, rename, per-device detail page
  - category → kind mapping; detect_switch_dps helper
  - format_dp_value: temperature (÷10 + °C/°F), humidity (+ %)
    registered as 'tuya_dp' Jinja2 filter

- TuyaDevice model (tuya_devices table)

- Templates:
  - tuya/gateway.html: device grid with live-reading sensor cards
    (config/threshold keys hidden from card, shown on detail page)
  - tuya/device.html: full status table with formatted DP values
  - tuya/auth_settings.html: user_code input + QR scan flow

- Add-board wizard refactored to 2-step flow:
  - Step 1: choose board type (Cloud Gateways vs Hardware)
  - Step 2: type-specific fields; gateways skip IP/relay fields

- Layout builder: Tuya chip support (makeTuyaChip, tuya_update socket)

- requirements.txt: tuya-device-sharing-sdk, cryptography
2026-02-27 16:06:48 +02:00

79 lines
3.3 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""TuyaDevice model represents a single Tuya / Smart Life sub-device
belonging to a Tuya Cloud Gateway board.
Each board (gateway) has many TuyaDevice rows, one per cloud device.
Controllable "channels" are boolean switch Data-Points (DPs) identified
by dp_code strings like "switch", "switch_1", "switch_2", …
"""
import json
from datetime import datetime
from app import db
class TuyaDevice(db.Model):
__tablename__ = "tuya_devices"
id = db.Column(db.Integer, primary_key=True)
board_id = db.Column(db.Integer, db.ForeignKey("boards.id"), nullable=False)
# Tuya global device ID (e.g. "bf0123456789abc")
device_id = db.Column(db.String(64), nullable=False)
# User-visible name
name = db.Column(db.String(128), default="")
# Tuya product category code (e.g. "kg"=switch, "dj"=light)
category = db.Column(db.String(32), default="")
# Full product name from Tuya cloud
product_name = db.Column(db.String(128), default="")
# Derived kind label: switch / light / fan / sensor / cover
kind = db.Column(db.String(32), default="switch")
# Bootstrap icon class for this kind
kind_icon_cls = db.Column(db.String(64), default="bi-toggles")
# Number of controllable switch channels
num_channels = db.Column(db.Integer, default=1)
# JSON list of switch DP codes in channel order, e.g. ["switch_1","switch_2"]
switch_dps_json = db.Column(db.Text, default='["switch"]')
# Full device status as JSON dict, e.g. {"switch":true,"countdown":0}
status_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="tuya_devices")
# ── status helpers ────────────────────────────────────────────────────────
@property
def status(self) -> dict:
try:
return json.loads(self.status_json or "{}")
except (ValueError, TypeError):
return {}
@status.setter
def status(self, value: dict):
self.status_json = json.dumps(value)
# ── switch-DP helpers ─────────────────────────────────────────────────────
@property
def switch_dps(self) -> list[str]:
"""Ordered list of switch DP codes, e.g. ['switch_1', 'switch_2']."""
try:
v = json.loads(self.switch_dps_json or '["switch"]')
return v if v else ["switch"]
except (ValueError, TypeError):
return ["switch"]
def get_channel_state(self, idx: int) -> bool:
"""Return the on/off state of the switch channel at *idx*."""
dps = self.switch_dps
if not dps:
return False
dp_code = dps[idx] if idx < len(dps) else dps[0]
return bool(self.status.get(dp_code, False))
def dp_for_channel(self, idx: int) -> str:
"""Return the dp_code for channel at *idx*."""
dps = self.switch_dps
return dps[idx] if dps and idx < len(dps) else dps[0] if dps else "switch"