diff --git a/server/SHServ/Controllers/DevicesRESTAPIController.php b/server/SHServ/Controllers/DevicesRESTAPIController.php index a76041a..175b929 100644 --- a/server/SHServ/Controllers/DevicesRESTAPIController.php +++ b/server/SHServ/Controllers/DevicesRESTAPIController.php @@ -29,7 +29,11 @@ public function setup_new_device($device_ip, $alias, $name, $description) { $devices_model = new \SHServ\Models\Devices(); - if(!filter_var($device_ip, FILTER_VALIDATE_IP, FILTER_FLAG_IPV4)) { + $host = $device_ip; + if(strpos($device_ip, ':') !== false) { + list($host, ) = explode(':', $device_ip, 2); + } + if(!filter_var($host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV4)) { return $this -> utils() -> response_error("invalid_ip", ["device_ip"]); } diff --git a/tools/virtual_devices/.gitignore b/tools/virtual_devices/.gitignore new file mode 100644 index 0000000..760e420 --- /dev/null +++ b/tools/virtual_devices/.gitignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +*.pyc +*.pyo +*.pid diff --git a/tools/virtual_devices/README.md b/tools/virtual_devices/README.md new file mode 100644 index 0000000..0d01358 --- /dev/null +++ b/tools/virtual_devices/README.md @@ -0,0 +1,103 @@ +# Virtual Device Emulator + +Эмуляторы устройств умного дома на Python + Flask. Полностью повторяют REST-контракт реальных ESP8266/ESP32 устройств. + +## Типы устройств + +- **relay** — 4 канала реле. Управление `set_state` (`on`/`off`). +- **button** — 4 канала кнопок. При клике шлёт событие `POST /events/new` на сервер. + +## Установка + +```bash +cd tools/virtual_devices +python3 -m venv .venv +source .venv/bin/activate # или .venv\Scripts\activate на Windows +pip install -r requirements.txt +``` + +## CLI + +```bash +# Создать виртуальное реле +python cli.py create --type relay --alias virt_relay --name "Virtual Relay" --port 9001 + +# Создать виртуальные кнопки +python cli.py create --type button --alias virt_btn --name "Virtual Buttons" --port 9002 + +# Запустить эмулятор +python cli.py start --alias virt_relay +python cli.py start --alias virt_btn + +# Список устройств +python cli.py list + +# Остановить +python cli.py stop --alias virt_relay + +# Статус (JSON) +python cli.py status --alias virt_relay + +# Нажать кнопку (для button-типа) +python cli.py click --alias virt_btn --channel 0 + +# Удалить устройство +python cli.py remove --alias virt_relay +``` + +## Регистрация в PHP-сервере + +```bash +# Убедитесь, что сервер доступен по нужному URL +python cli.py register --alias virt_relay --server-url http://smart-home-serv.local +``` + +Команда `register`: +1. Проверяет `/about` эмулятора (должен быть `setup`) +2. Отправляет `POST /api/v1/devices/setup/new-device` на сервер +3. Сервер сам устанавливает токен и переводит устройство в `normal` + +## Web UI + +Каждый эмулятор предоставляет UI на корневом URL: +``` +http://127.0.0.1:9001/ +``` + +- Для реле — тумблеры on/off +- Для кнопок — кнопки Click и лог событий +- Обновление каждые 2 секунды + +## Эндпоинты эмулятора + +| Method | Path | Auth | Описание | +|--------|------|------|----------| +| GET | `/about` | — | Информация об устройстве | +| GET | `/status` | Bearer | Состояние каналов | +| POST | `/action` | Bearer | Управление (`set_state`, `simulate_click`) | +| POST | `/set_token` | — (setup) / Bearer (normal) | Установка токена | +| POST | `/reset` | Bearer | Сброс в setup | +| GET | `/channels_schema` | Bearer (или — в setup) | Схема каналов | +| POST | `/simulate-event` | — | Debug: ручной триггер события | + +## Структура + +``` +virtual_devices/ +├── cli.py # Управление устройствами +├── emulator.py # Flask-приложение эмулятора +├── state.py # Хранение состояния (JSON) +├── device/ +│ ├── base.py # Базовый класс устройства +│ ├── relay.py # Реле +│ └── button.py # Кнопки +├── templates/ +│ └── control.html # UI (inline в emulator.py) +├── devices/ # Состояния (*.json, .gitignore) +└── pids/ # PID-файлы запущенных процессов +``` + +## Изменения в сервере + +Для поддержки `IP:port` в `device_ip` при регистрации: +- `server/SHServ/Controllers/DevicesRESTAPIController.php` — валидация `setup_new_device` теперь разбирает `host:port` и валидирует только host. diff --git a/tools/virtual_devices/cli.py b/tools/virtual_devices/cli.py new file mode 100644 index 0000000..7cfe84b --- /dev/null +++ b/tools/virtual_devices/cli.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +import uuid +from pathlib import Path + +import click +import requests + +from state import DeviceState, list_all, load, remove, save + +PID_DIR = Path(__file__).parent / "pids" +PID_DIR.mkdir(exist_ok=True) + + +def _pid_file(alias: str) -> Path: + return PID_DIR / f"{alias}.pid" + + +def _is_running(alias: str) -> bool: + pid_file = _pid_file(alias) + if not pid_file.exists(): + return False + pid = int(pid_file.read_text().strip()) + try: + os.kill(pid, 0) + return True + except OSError: + pid_file.unlink(missing_ok=True) + return False + + +def _start_emulator(alias: str, host: str = "0.0.0.0", port: int | None = None) -> None: + state = load(alias) + if state is None: + click.echo(f"Device '{alias}' not found.", err=True) + sys.exit(1) + + if _is_running(alias): + click.echo(f"Device '{alias}' is already running (PID {int(_pid_file(alias).read_text().strip())}).") + sys.exit(1) + + cmd = [sys.executable, str(Path(__file__).parent / "emulator.py"), "--alias", alias, "--host", host] + if port is not None: + cmd += ["--port", str(port)] + + proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + time.sleep(0.5) + if proc.poll() is not None: + click.echo("Emulator failed to start.", err=True) + sys.exit(1) + + _pid_file(alias).write_text(str(proc.pid)) + actual_port = port or state.port + click.echo(f"Started '{alias}' on http://{host}:{actual_port} (PID {proc.pid})") + + +def _stop_emulator(alias: str) -> None: + pid_file = _pid_file(alias) + if not pid_file.exists(): + click.echo(f"Device '{alias}' is not running.") + sys.exit(1) + + pid = int(pid_file.read_text().strip()) + try: + os.kill(pid, 9) + click.echo(f"Stopped '{alias}' (PID {pid}).") + except ProcessLookupError: + click.echo(f"Process {pid} not found, cleaning up.") + finally: + pid_file.unlink(missing_ok=True) + + +@click.group() +def cli(): + """Virtual Smart Home Device Manager.""" + pass + + +@cli.command() +@click.option("--type", "device_type", required=True, type=click.Choice(["relay", "button"])) +@click.option("--alias", required=True, help="Unique alias (e.g. virt_relay)") +@click.option("--name", default=None, help="Human-readable name") +@click.option("--channels", default=4, type=int, help="Number of channels") +@click.option("--port", default=9001, type=int, help="HTTP port for emulator") +@click.option("--server-url", default="http://localhost", help="Server URL to send events to") +@click.option("--ip", default="127.0.0.1", help="IP address reported in /about") +def create(device_type, alias, name, channels, port, server_url, ip): + """Create a new virtual device state file.""" + if load(alias) is not None: + click.echo(f"Device '{alias}' already exists. Use a different alias or remove it first.", err=True) + sys.exit(1) + + device_id = f"virt-{uuid.uuid4().hex[:12]}" + mac = f"VIRTUAL:{uuid.uuid4().hex[:6].upper()}:{uuid.uuid4().hex[:6].upper()}" + state = DeviceState( + alias=alias, + device_type=device_type, + device_id=device_id, + device_name=name or alias, + port=port, + server_url=server_url, + mac_address=mac, + channels=[{} for _ in range(channels)], + ) + state.channels_schema = [] + for ch in range(8): + base = ch * 4 + state.channels_schema.extend([base, ch, 255, 0]) + save(state) + click.echo(f"Created {device_type} '{alias}' (ID {device_id}) on port {port}.") + click.echo(f"State file: devices/{alias}.json") + + +@cli.command() +@click.option("--alias", required=True) +@click.option("--host", default="0.0.0.0") +@click.option("--port", type=int, default=None) +def start(alias, host, port): + """Start the emulator for a device.""" + _start_emulator(alias, host, port) + + +@cli.command() +@click.option("--alias", required=True) +def stop(alias): + """Stop the emulator for a device.""" + _stop_emulator(alias) + + +@cli.command("list") +def list_devices(): + """List all virtual devices and their running status.""" + aliases = list_all() + if not aliases: + click.echo("No virtual devices found. Use 'create' to make one.") + return + + click.echo(f"{'Alias':<20} {'Type':<10} {'Port':<8} {'Status'}") + click.echo("-" * 50) + for alias in aliases: + state = load(alias) + running = "running" if _is_running(alias) else "stopped" + click.echo(f"{alias:<20} {state.device_type:<10} {state.port:<8} {running}") + + +@cli.command() +@click.option("--alias", required=True) +def status(alias): + """Show device state and current channels.""" + state = load(alias) + if state is None: + click.echo(f"Device '{alias}' not found.", err=True) + sys.exit(1) + + click.echo(json.dumps(state.to_dict(), indent=2, ensure_ascii=False)) + + +@cli.command("click") +@click.option("--alias", required=True) +@click.option("--channel", default=0, type=int) +def click_cmd(alias, channel): + """Simulate a button click (for button-type devices).""" + state = load(alias) + if state is None: + click.echo(f"Device '{alias}' not found.", err=True) + sys.exit(1) + + url = f"http://127.0.0.1:{state.port}/simulate-event" + try: + resp = requests.post(url, json={"event_name": "click", "channel": channel}, timeout=5) + click.echo(resp.json()) + except requests.RequestException as e: + click.echo(f"Failed to reach emulator: {e}", err=True) + sys.exit(1) + + +@cli.command() +@click.option("--alias", required=True) +@click.option("--server-url", default="http://localhost", help="PHP server base URL") +def register(alias, server_url): + """Register the virtual device in the PHP server (setup flow).""" + state = load(alias) + if state is None: + click.echo(f"Device '{alias}' not found.", err=True) + sys.exit(1) + + # Step 1: call /about to verify setup mode + about_url = f"http://127.0.0.1:{state.port}/about" + try: + resp = requests.get(about_url, timeout=5) + about = resp.json() + except requests.RequestException as e: + click.echo(f"Emulator not reachable at {about_url}: {e}", err=True) + sys.exit(1) + + if about.get("status") != "setup": + click.echo("Device is not in setup mode. Reset it first.", err=True) + sys.exit(1) + + # Step 2: register in PHP server via API + setup_url = f"{server_url.rstrip('/')}/api/v1/devices/setup/new-device" + payload = { + "device_ip": f"127.0.0.1:{state.port}", + "alias": alias, + "name": state.device_name, + "description": "Virtual device (emulator)", + } + try: + resp = requests.post(setup_url, json=payload, timeout=10) + result = resp.json() + except requests.RequestException as e: + click.echo(f"Server request failed: {e}", err=True) + sys.exit(1) + + if result.get("status") == "ok": + click.echo(f"Device '{alias}' registered successfully.") + click.echo(json.dumps(result.get("data", {}).get("device", {}), indent=2, ensure_ascii=False)) + else: + click.echo(f"Registration failed: {result}", err=True) + sys.exit(1) + + +@cli.command("remove") +@click.option("--alias", required=True) +def remove_cmd(alias): + """Remove a virtual device state file.""" + if _is_running(alias): + click.echo(f"Device '{alias}' is running. Stop it first.", err=True) + sys.exit(1) + + if remove(alias): + click.echo(f"Removed device '{alias}'.") + else: + click.echo(f"Device '{alias}' not found.", err=True) + sys.exit(1) + + +if __name__ == "__main__": + cli() diff --git a/tools/virtual_devices/device/__init__.py b/tools/virtual_devices/device/__init__.py new file mode 100644 index 0000000..1d8faab --- /dev/null +++ b/tools/virtual_devices/device/__init__.py @@ -0,0 +1,5 @@ +from device.base import BaseDevice +from device.relay import RelayDevice +from device.button import ButtonDevice + +__all__ = ["BaseDevice", "RelayDevice", "ButtonDevice"] diff --git a/tools/virtual_devices/device/base.py b/tools/virtual_devices/device/base.py new file mode 100644 index 0000000..a16ee90 --- /dev/null +++ b/tools/virtual_devices/device/base.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import time +from typing import Any, Dict, List + +from state import DeviceState, save + + +class BaseDevice: + def __init__(self, state: DeviceState): + self.state = state + + # ── helpers ────────────────────────────────────────────── + + def _uptime(self) -> int: + return int(time.time() - self.state.uptime_start) + + def _is_setup(self) -> bool: + return self.state.status == "setup" + + def _check_auth(self, headers: Dict[str, str]) -> bool: + if self._is_setup(): + return True + auth = headers.get("Authorization", "") + if not auth.startswith("Bearer "): + return False + token = auth[7:] + return token == (self.state.token or "") + + def _ensure_channels_schema(self, channels_count: int = 8) -> None: + if not self.state.channels_schema: + schema: List[int] = [] + for ch in range(channels_count): + base = ch * 4 + schema.extend([base, ch, 255, 0]) + self.state.channels_schema = schema + + def _ensure_channels(self, count: int) -> None: + while len(self.state.channels) < count: + self.state.channels.append({"state": "off"}) + for ch in self.state.channels: + ch.setdefault("state", "off") + + # ── common endpoints ───────────────────────────────────── + + def about(self, ip_address: str) -> Dict[str, Any]: + return { + "device_name": self.state.device_name, + "device_type": self.state.device_type, + "firmware_version": self.state.firmware_version, + "device_id": self.state.device_id, + "server": self.state.server_url, + "status": self.state.status, + "ip_address": ip_address, + "mac_address": self.state.mac_address, + "uptime": self._uptime(), + } + + def set_token(self, token: str) -> Dict[str, Any]: + self.state.token = token + self.state.status = "normal" + save(self.state) + return { + "status": "ok", + "message": "Token set. Device mode: normal", + } + + def reset(self) -> Dict[str, Any]: + self.state.token = None + self.state.status = "setup" + self.state.uptime_start = time.time() + for ch in self.state.channels: + ch["state"] = "off" + save(self.state) + return { + "status": "ok", + "message": "Device reset to factory settings. Entering setup mode.", + } + + def get_channels_schema(self) -> Dict[str, Any]: + self._ensure_channels_schema() + return { + "status": "ok", + "schema": self.state.channels_schema, + } + + def set_channels_schema(self, schema: List[int]) -> Dict[str, Any]: + self.state.channels_schema = schema + save(self.state) + return { + "status": "ok", + "message": "Channels schema updated", + } + + # ── to be overridden ───────────────────────────────────── + + def status(self) -> Dict[str, Any]: + raise NotImplementedError + + def action(self, action_name: str, params: Dict[str, Any]) -> Dict[str, Any]: + raise NotImplementedError diff --git a/tools/virtual_devices/device/button.py b/tools/virtual_devices/device/button.py new file mode 100644 index 0000000..f9f702a --- /dev/null +++ b/tools/virtual_devices/device/button.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import time +from typing import Any, Dict + +import requests + +from device.base import BaseDevice +from state import save + + +class ButtonDevice(BaseDevice): + def __init__(self, state): + super().__init__(state) + self._ensure_channels(self._channel_count()) + for ch in self.state.channels: + ch["state"] = "idle" + + def _channel_count(self) -> int: + return 4 + + def status(self) -> Dict[str, Any]: + return {"channels": self.state.channels} + + def action(self, action_name: str, params: Dict[str, Any]) -> Dict[str, Any]: + if action_name == "simulate_click": + channel = params.get("channel", 0) + return self.trigger_click(channel) + + return { + "status": "error", + "error": "IllegalActionOrParams", + "message": "Device does not support this action or params", + } + + def trigger_click(self, channel: int) -> Dict[str, Any]: + if channel < 0 or channel >= len(self.state.channels): + return { + "status": "error", + "error": "IllegalActionOrParams", + "message": "Invalid channel index", + } + + now = time.strftime("%Y-%m-%d %H:%M:%S") + self.state.channels[channel]["last_event"] = "click" + self.state.channels[channel]["last_event_time"] = now + save(self.state) + + # Send event to server asynchronously (best-effort) + self._send_event("click", channel) + + return {"status": "ok", "message": f"Click triggered on channel {channel}"} + + def _send_event(self, event_name: str, channel: int) -> None: + server_url = self.state.server_url.rstrip("/") + url = f"{server_url}/events/new" + payload = { + "event_name": event_name, + "device_id": self.state.device_id, + "data": {"channel": channel}, + } + try: + requests.post(url, json=payload, timeout=3) + except Exception: + pass # Best-effort; server may be offline diff --git a/tools/virtual_devices/device/relay.py b/tools/virtual_devices/device/relay.py new file mode 100644 index 0000000..f9c6336 --- /dev/null +++ b/tools/virtual_devices/device/relay.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from typing import Any, Dict + +from device.base import BaseDevice +from state import save + + +class RelayDevice(BaseDevice): + def __init__(self, state): + super().__init__(state) + self._ensure_channels(self._channel_count()) + + def _channel_count(self) -> int: + # Default to 4 if not specified, but schema implies 8 max + # We use active channels based on schema or default to 4 + return 4 + + def status(self) -> Dict[str, Any]: + return {"channels": self.state.channels} + + def action(self, action_name: str, params: Dict[str, Any]) -> Dict[str, Any]: + if action_name == "set_state": + channel = params.get("channel", 0) + new_state = params.get("state", "off") + if channel < 0 or channel >= len(self.state.channels): + return { + "status": "error", + "error": "IllegalActionOrParams", + "message": "Invalid channel index", + } + if new_state not in ("on", "off"): + return { + "status": "error", + "error": "IllegalActionOrParams", + "message": "State must be 'on' or 'off'", + } + self.state.channels[channel]["state"] = new_state + save(self.state) + return {"status": "ok", "message": "State changed"} + + return { + "status": "error", + "error": "IllegalActionOrParams", + "message": "Device does not support this action or params", + } diff --git a/tools/virtual_devices/devices/.gitignore b/tools/virtual_devices/devices/.gitignore new file mode 100644 index 0000000..a6c57f5 --- /dev/null +++ b/tools/virtual_devices/devices/.gitignore @@ -0,0 +1 @@ +*.json diff --git a/tools/virtual_devices/emulator.py b/tools/virtual_devices/emulator.py new file mode 100644 index 0000000..60a4d7d --- /dev/null +++ b/tools/virtual_devices/emulator.py @@ -0,0 +1,340 @@ +from __future__ import annotations + +import argparse +import sys +from typing import Any, Dict + +from flask import Flask, jsonify, request, render_template_string + +from device import RelayDevice, ButtonDevice +from state import load + +app = Flask(__name__) + +# Global set by create_app() +device_instance = None + + +# ── helpers ────────────────────────────────────────────────── + +def _client_ip() -> str: + if request.environ.get("HTTP_X_FORWARDED_FOR") is None: + return request.environ["REMOTE_ADDR"] + else: + return request.environ["HTTP_X_FORWARDED_FOR"] + + +def _unauthorized() -> tuple: + return jsonify({ + "status": "error", + "error": "Unauthorized", + "message": "Missing or invalid token", + }), 401 + + +def _not_available() -> tuple: + return jsonify({ + "status": "error", + "error": "NotAvailable", + "message": "Setup mode is not active", + }), 403 + + +# ── about (always public) ────────────────────────────────── + +@app.get("/about") +def about(): + data = device_instance.about(_client_ip()) + data["ip_address"] = f"{_client_ip()}:{device_instance.state.port}" + return jsonify(data) + + +# ── status ────────────────────────────────────────────────── + +@app.get("/status") +def status(): + if not device_instance._check_auth(request.headers): + return _unauthorized() + return jsonify(device_instance.status()) + + +# ── action ────────────────────────────────────────────────── + +@app.post("/action") +def action(): + if not device_instance._check_auth(request.headers): + return _unauthorized() + data = request.get_json(silent=True) or {} + action_name = data.get("action", "") + params = data.get("params", {}) + result = device_instance.action(action_name, params) + return jsonify(result) + + +# ── set_token ────────────────────────────────────────────── + +@app.post("/set_token") +def set_token(): + data = request.get_json(silent=True) or {} + token = data.get("token", "") + + if device_instance._is_setup(): + if not token: + return jsonify({"status": "error", "error": "InvalidParams", "message": "Token required"}), 400 + return jsonify(device_instance.set_token(token)) + + # Normal mode — require auth + if not device_instance._check_auth(request.headers): + return _unauthorized() + + if not token: + return jsonify({"status": "error", "error": "InvalidParams", "message": "Token required"}), 400 + + return jsonify(device_instance.set_token(token)) + + +# ── reset ────────────────────────────────────────────────── + +@app.post("/reset") +def reset(): + if not device_instance._check_auth(request.headers): + return _unauthorized() + return jsonify(device_instance.reset()) + + +# ── reboot (stub) ────────────────────────────────────────── + +@app.post("/reboot") +def reboot(): + if not device_instance._check_auth(request.headers): + return _unauthorized() + return jsonify({"status": "ok", "message": "Device will reboot now"}) + + +# ── set_device_name ──────────────────────────────────────── + +@app.post("/set_device_name") +def set_device_name(): + if not device_instance._check_auth(request.headers): + return _unauthorized() + data = request.get_json(silent=True) or {} + name = data.get("device_name", "") + if name: + device_instance.state.device_name = name + from state import save + save(device_instance.state) + return jsonify({"status": "ok", "message": "Device name updated"}) + + +# ── channels schema ──────────────────────────────────────── + +@app.get("/channels_schema") +def get_channels_schema(): + if device_instance._is_setup(): + return jsonify(device_instance.get_channels_schema()) + if not device_instance._check_auth(request.headers): + return _unauthorized() + return jsonify(device_instance.get_channels_schema()) + + +@app.post("/set_channels_schema") +def set_channels_schema(): + if device_instance._is_setup(): + pass # allowed + elif not device_instance._check_auth(request.headers): + return _unauthorized() + data = request.get_json(silent=True) or {} + schema = data.get("schema", []) + if not isinstance(schema, list): + return jsonify({"status": "error", "error": "InvalidParams"}), 400 + return jsonify(device_instance.set_channels_schema(schema)) + + +# ── setup (stub) ─────────────────────────────────────────── + +@app.get("/setup") +def setup_get(): + if not device_instance._is_setup(): + return _not_available() + return jsonify({"status": "ok", "message": "Setup mode active"}) + + +@app.post("/setup") +def setup_post(): + if not device_instance._is_setup(): + return _not_available() + return jsonify({"status": "ok", "message": "Wi-Fi configured. Connecting..."}) + + +# ── simulate-event (debug helper, no auth) ───────────────── + +@app.post("/simulate-event") +def simulate_event(): + data = request.get_json(silent=True) or {} + event_name = data.get("event_name", "click") + channel = data.get("channel", 0) + + if hasattr(device_instance, "trigger_click"): + result = device_instance.trigger_click(channel) + return jsonify(result) + + # For relay, just flip state + if device_instance.state.device_type == "relay": + current = device_instance.state.channels[channel].get("state", "off") + new_state = "on" if current == "off" else "off" + result = device_instance.action("set_state", {"channel": channel, "state": new_state}) + return jsonify(result) + + return jsonify({"status": "error", "message": "Unsupported device type"}), 400 + + +# ── web UI ───────────────────────────────────────────────── + +_UI_HTML = """ + + + + + + {{ alias }} — {{ device_type }} + + + +
+

{{ alias }} ({{ device_type }})

+
+ ID: {{ device_id }} · Port: {{ port }} · Mode: {{ status }} + {% if token %}· Token set{% endif %} +
+ 

+  

+
+  

+    
Channels

+    

+  

+
+  

+    
Raw Status

+    

+  

+
+
+
+
+"""
+
+
+@app.get("/")
+def index():
+    return render_template_string(
+        _UI_HTML,
+        alias=device_instance.state.alias,
+        device_type=device_instance.state.device_type,
+        device_id=device_instance.state.device_id,
+        port=device_instance.state.port,
+        status=device_instance.state.status,
+        token=bool(device_instance.state.token),
+    )
+
+
+# ── app factory ──────────────────────────────────────────────
+
+def create_app(alias: str) -> Flask:
+    global device_instance
+    state = load(alias)
+    if state is None:
+        print(f"Device '{alias}' not found. Create it first with cli.py", file=sys.stderr)
+        sys.exit(1)
+
+    if state.device_type == "relay":
+        device_instance = RelayDevice(state)
+    elif state.device_type == "button":
+        device_instance = ButtonDevice(state)
+    else:
+        print(f"Unknown device type: {state.device_type}", file=sys.stderr)
+        sys.exit(1)
+
+    return app
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Virtual SH device emulator")
+    parser.add_argument("--alias", required=True, help="Device alias (must be created via cli.py)")
+    parser.add_argument("--host", default="0.0.0.0", help="Bind host")
+    parser.add_argument("--port", type=int, default=None, help="Override port (default from device state)")
+    args = parser.parse_args()
+
+    flask_app = create_app(args.alias)
+    port = args.port or device_instance.state.port
+    flask_app.run(host=args.host, port=port, debug=False)
diff --git a/tools/virtual_devices/pids/.gitignore b/tools/virtual_devices/pids/.gitignore
new file mode 100644
index 0000000..9fd4f9b
--- /dev/null
+++ b/tools/virtual_devices/pids/.gitignore
@@ -0,0 +1 @@
+*.pid
diff --git a/tools/virtual_devices/requirements.txt b/tools/virtual_devices/requirements.txt
new file mode 100644
index 0000000..56432e0
--- /dev/null
+++ b/tools/virtual_devices/requirements.txt
@@ -0,0 +1,4 @@
+Flask==3.1.0
+requests==2.32.3
+click==8.1.7
+Werkzeug==3.1.3
diff --git a/tools/virtual_devices/state.py b/tools/virtual_devices/state.py
new file mode 100644
index 0000000..62e2761
--- /dev/null
+++ b/tools/virtual_devices/state.py
@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass, field, asdict
+from typing import Any, Dict, List
+
+
+STATE_DIR = os.path.join(os.path.dirname(__file__), "devices")
+
+
+def _path(alias: str) -> str:
+    return os.path.join(STATE_DIR, f"{alias}.json")
+
+
+@dataclass
+class ChannelState:
+    state: str = "off"  # "on" | "off" for relay; "idle" | "clicked" for button
+    last_event: str = ""
+    last_event_time: str = ""
+
+
+@dataclass
+class DeviceState:
+    alias: str
+    device_type: str
+    device_id: str
+    device_name: str
+    port: int
+    server_url: str
+    mac_address: str
+    firmware_version: str = "virtual-1.0.0"
+    status: str = "setup"  # "setup" | "normal"
+    token: str | None = None
+    channels: List[Dict[str, Any]] = field(default_factory=list)
+    channels_schema: List[int] = field(default_factory=list)
+    uptime_start: float = field(default_factory=lambda: __import__("time").time())
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "DeviceState":
+        return cls(**data)
+
+
+def ensure_dir() -> None:
+    os.makedirs(STATE_DIR, exist_ok=True)
+
+
+def load(alias: str) -> DeviceState | None:
+    ensure_dir()
+    path = _path(alias)
+    if not os.path.exists(path):
+        return None
+    with open(path, "r", encoding="utf-8") as f:
+        return DeviceState.from_dict(json.load(f))
+
+
+def save(state: DeviceState) -> None:
+    ensure_dir()
+    path = _path(state.alias)
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(state.to_dict(), f, indent=2, ensure_ascii=False)
+
+
+def remove(alias: str) -> bool:
+    path = _path(alias)
+    if os.path.exists(path):
+        os.remove(path)
+        return True
+    return False
+
+
+def list_all() -> List[str]:
+    ensure_dir()
+    names = []
+    for fname in os.listdir(STATE_DIR):
+        if fname.endswith(".json"):
+            names.append(fname[:-5])
+    return sorted(names)