from flask import Flask, request, redirect, url_for, render_template_string import requests from typing import Any, Dict, List, Optional, Tuple app = Flask(__name__) app.config["SECRET_KEY"] = "sh-schema-editor-dev" CHANNEL_COUNT = 8 CHANNEL_BYTES = 4 SCHEMA_LEN = CHANNEL_COUNT * CHANNEL_BYTES SH_PIN_UNUSED = 255 FIELD_NAMES = ["pin", "indicator", "feedback", "flags"] FIELD_LABELS = { "pin": "Pin", "indicator": "Indicator", "feedback": "Feedback", "flags": "Flags", } HTML = """ SH Channels Schema Editor

SH Channels Schema Editor

Port 8001. Reads and writes /channels_schema and /set_channels_schema.

{% if device_ip %}

Device: {{ device_ip }}

{% endif %}

{% if message %}

{% endif %}

Connection

{% if schema %}

Schema editor

Raw schema

{% endif %}

""" def build_headers(token: str) -> Dict[str, str]: headers: Dict[str, str] = {"Content-Type": "application/json"} token = token.strip() if token: headers["Authorization"] = f"Bearer {token}" return headers def normalize_ip(device_ip: str) -> str: device_ip = device_ip.strip() if device_ip.startswith("http://") or device_ip.startswith("https://"): return device_ip.rstrip("/") return f"http://{device_ip}" def parse_device_schema(payload: Dict[str, Any]) -> List[int]: schema = payload.get("schema") if not isinstance(schema, list): raise ValueError("Device response does not contain a schema array") if len(schema) != SCHEMA_LEN: raise ValueError(f"Schema length is {len(schema)}, expected {SCHEMA_LEN}") normalized: List[int] = [] for value in schema: if not isinstance(value, int): raise ValueError("Schema contains a non-integer value") if value < 0 or value > 255: raise ValueError("Schema values must be in range 0..255") normalized.append(value) return normalized def get_schema(device_ip: str, token: str) -> List[int]: base_url = normalize_ip(device_ip) response = requests.get( f"{base_url}/channels_schema", headers=build_headers(token), timeout=5, ) response.raise_for_status() data = response.json() if data.get("status") == "error": raise ValueError(data.get("message") or data.get("error") or "Device returned an error") return parse_device_schema(data) def set_schema(device_ip: str, token: str, schema: List[int]) -> Dict[str, Any]: base_url = normalize_ip(device_ip) response = requests.post( f"{base_url}/set_channels_schema", headers=build_headers(token), json={"schema": schema}, timeout=8, ) response.raise_for_status() data = response.json() if data.get("status") == "error": raise ValueError(data.get("message") or data.get("error") or "Device returned an error") return data def split_schema(schema: List[int]) -> List[Dict[str, Any]]: channels: List[Dict[str, Any]] = [] for index in range(CHANNEL_COUNT): base = index * CHANNEL_BYTES flags = schema[base + 3] channels.append( { "index": index, "base": base, "pin": schema[base + 0], "indicator": schema[base + 1], "feedback": schema[base + 2], "flags": flags, "invert": (flags & 0x01) != 0, } ) return channels def coerce_byte(value: Optional[str], field_name: str) -> int: if value is None: raise ValueError(f"Missing value for {field_name}") value = value.strip() if value == "": raise ValueError(f"Empty value for {field_name}") try: parsed = int(value) except ValueError as exc: raise ValueError(f"Invalid integer for {field_name}") from exc if parsed < 0 or parsed > 255: raise ValueError(f"Value for {field_name} must be 0..255") return parsed def schema_from_form(form: Any) -> List[int]: schema: List[int] = [] for ch in range(CHANNEL_COUNT): pin = coerce_byte(form.get(f"pin_{ch}"), f"pin_{ch}") indicator = coerce_byte(form.get(f"indicator_{ch}"), f"indicator_{ch}") feedback = coerce_byte(form.get(f"feedback_{ch}"), f"feedback_{ch}") flags = coerce_byte(form.get(f"flags_{ch}"), f"flags_{ch}") if form.get(f"invert_{ch}"): flags |= 0x01 else: flags &= 0xFE schema.extend([pin, indicator, feedback, flags]) return schema def render_page( *, device_ip: str = "", token: str = "", schema: Optional[List[int]] = None, message: str = "", message_type: str = "info", ) -> str: return render_template_string( HTML, device_ip=device_ip, token=token, schema=schema, channels=split_schema(schema) if schema else [], message=message, message_type=message_type, ) @app.get("/") def index() -> str: return render_page() @app.post("/connect") def connect() -> str: device_ip = request.form.get("device_ip", "").strip() token = request.form.get("token", "").strip() if not device_ip: return render_page(device_ip=device_ip, token=token, message="Device IP is required.", message_type="err") try: schema = get_schema(device_ip, token) except Exception as exc: return render_page(device_ip=device_ip, token=token, message=f"Failed to load schema: {exc}", message_type="err") return render_page(device_ip=device_ip, token=token, schema=schema, message="Schema loaded successfully.", message_type="ok") @app.post("/save") def save_schema() -> str: device_ip = request.form.get("device_ip", "").strip() token = request.form.get("token", "").strip() if not device_ip: return render_page(device_ip=device_ip, token=token, message="Device IP is required.", message_type="err") try: schema = schema_from_form(request.form) fill_unused_channel = request.form.get("fill_unused") if fill_unused_channel is not None: ch = int(fill_unused_channel) base = ch * CHANNEL_BYTES schema[base + 0] = SH_PIN_UNUSED schema[base + 1] = SH_PIN_UNUSED schema[base + 2] = SH_PIN_UNUSED schema[base + 3] = 0 return render_page(device_ip=device_ip, token=token, schema=schema, message=f"Channel {ch} marked as unused. Not saved yet.", message_type="info") if request.form.get("fill_all_unused"): schema = [] for _ in range(CHANNEL_COUNT): schema.extend([SH_PIN_UNUSED, SH_PIN_UNUSED, SH_PIN_UNUSED, 0]) return render_page(device_ip=device_ip, token=token, schema=schema, message="All channels marked as unused. Not saved yet.", message_type="info") if request.form.get("normalize"): return render_page(device_ip=device_ip, token=token, schema=schema, message="Flags normalized from checkboxes. Not saved yet.", message_type="info") result = set_schema(device_ip, token, schema) return render_page(device_ip=device_ip, token=token, schema=schema, message=result.get("message", "Schema saved."), message_type="ok") except Exception as exc: try: schema = schema_from_form(request.form) except Exception: schema = None return render_page(device_ip=device_ip, token=token, schema=schema, message=f"Failed to save schema: {exc}", message_type="err") if __name__ == "__main__": app.run(host="0.0.0.0", port=8001, debug=True)