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 = """
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>SH Channels Schema Editor</title>
<style>
:root {
--bg: #0f1115;
--panel: #181c23;
--panel-2: #202632;
--text: #e8eef7;
--muted: #9fb0c7;
--accent: #4da3ff;
--accent-2: #2d7fd6;
--danger: #d86161;
--ok: #5fbc7a;
--border: #2f3745;
--input: #121720;
--warning: #d8aa52;
}
* { box-sizing: border-box; }
body {
margin: 0;
font-family: Arial, sans-serif;
background: var(--bg);
color: var(--text);
}
.wrap {
max-width: 1180px;
margin: 0 auto;
padding: 24px;
}
.panel {
background: var(--panel);
border: 1px solid var(--border);
border-radius: 14px;
padding: 18px;
margin-bottom: 18px;
}
h1, h2, h3 { margin-top: 0; }
.muted { color: var(--muted); }
.grid {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 14px;
}
.grid-4 {
display: grid;
grid-template-columns: repeat(4, minmax(0, 1fr));
gap: 12px;
}
label {
display: block;
font-size: 14px;
margin-bottom: 6px;
color: var(--muted);
}
input[type=text], input[type=number] {
width: 100%;
background: var(--input);
color: var(--text);
border: 1px solid var(--border);
border-radius: 10px;
padding: 10px 12px;
outline: none;
}
input[type=checkbox] {
transform: scale(1.2);
}
.btn-row {
display: flex;
gap: 10px;
flex-wrap: wrap;
margin-top: 14px;
}
button {
background: var(--accent);
color: white;
border: 0;
border-radius: 10px;
padding: 10px 14px;
cursor: pointer;
font-weight: 700;
}
button.secondary {
background: #445067;
}
button.warn {
background: var(--warning);
color: #111;
}
button:hover { background: var(--accent-2); }
.channel-card {
background: var(--panel-2);
border: 1px solid var(--border);
border-radius: 14px;
padding: 14px;
margin-bottom: 14px;
}
.channel-header {
display: flex;
align-items: center;
justify-content: space-between;
gap: 10px;
margin-bottom: 12px;
}
.channel-title {
font-size: 18px;
font-weight: 700;
}
.tag {
display: inline-block;
padding: 4px 8px;
border-radius: 999px;
background: #273140;
color: var(--muted);
font-size: 12px;
border: 1px solid var(--border);
}
.msg {
padding: 12px 14px;
border-radius: 10px;
margin-bottom: 14px;
border: 1px solid var(--border);
}
.msg.ok { background: rgba(95,188,122,0.14); color: #dff4e6; }
.msg.err { background: rgba(216,97,97,0.14); color: #ffdede; }
.msg.info { background: rgba(77,163,255,0.14); color: #e2f0ff; }
.mono {
font-family: Consolas, monospace;
white-space: pre-wrap;
word-break: break-word;
}
.topline {
display: flex;
justify-content: space-between;
gap: 12px;
align-items: center;
flex-wrap: wrap;
}
.small { font-size: 13px; }
.flags-row {
display: flex;
gap: 18px;
align-items: center;
flex-wrap: wrap;
margin-top: 8px;
}
.inline-check {
display: flex;
align-items: center;
gap: 8px;
color: var(--text);
}
.footer-note {
color: var(--muted);
font-size: 13px;
margin-top: 12px;
}
@media (max-width: 900px) {
.grid, .grid-4 { grid-template-columns: 1fr; }
}
</style>
</head>
<body>
<div class="wrap">
<div class="topline">
<div>
<h1>SH Channels Schema Editor</h1>
<div class="muted">Port 8001. Reads and writes <span class="mono">/channels_schema</span> and <span class="mono">/set_channels_schema</span>.</div>
</div>
{% if device_ip %}
<div class="tag">Device: {{ device_ip }}</div>
{% endif %}
</div>
{% if message %}
<div class="msg {{ message_type }}">{{ message }}</div>
{% endif %}
<div class="panel">
<h2>Connection</h2>
<form method="post" action="{{ url_for('connect') }}">
<div class="grid">
<div>
<label for="device_ip">Device IP or host</label>
<input id="device_ip" name="device_ip" type="text" value="{{ device_ip }}" placeholder="192.168.1.50">
</div>
<div>
<label for="token">Authorization token</label>
<input id="token" name="token" type="text" value="{{ token }}" placeholder="Bearer token value only">
</div>
</div>
<div class="btn-row">
<button type="submit">Load schema from device</button>
</div>
</form>
<div class="footer-note">In setup mode the token may be empty. In normal mode the device usually requires a valid Bearer token.</div>
</div>
{% if schema %}
<div class="panel">
<h2>Schema editor</h2>
<form method="post" action="{{ url_for('save_schema') }}">
<input type="hidden" name="device_ip" value="{{ device_ip }}">
<input type="hidden" name="token" value="{{ token }}">
{% for channel in channels %}
<div class="channel-card">
<div class="channel-header">
<div class="channel-title">Channel {{ channel.index }}</div>
<div class="tag">Bytes {{ channel.base }}..{{ channel.base + 3 }}</div>
</div>
<div class="grid-4">
<div>
<label for="pin_{{ channel.index }}">Pin</label>
<input id="pin_{{ channel.index }}" name="pin_{{ channel.index }}" type="number" min="0" max="255" value="{{ channel.pin }}">
</div>
<div>
<label for="indicator_{{ channel.index }}">Indicator</label>
<input id="indicator_{{ channel.index }}" name="indicator_{{ channel.index }}" type="number" min="0" max="255" value="{{ channel.indicator }}">
</div>
<div>
<label for="feedback_{{ channel.index }}">Feedback</label>
<input id="feedback_{{ channel.index }}" name="feedback_{{ channel.index }}" type="number" min="0" max="255" value="{{ channel.feedback }}">
</div>
<div>
<label for="flags_{{ channel.index }}">Flags (raw byte)</label>
<input id="flags_{{ channel.index }}" name="flags_{{ channel.index }}" type="number" min="0" max="255" value="{{ channel.flags }}">
</div>
</div>
<div class="flags-row">
<label class="inline-check">
<input type="checkbox" name="invert_{{ channel.index }}" {% if channel.invert %}checked{% endif %}>
Invert channel (bit 0)
</label>
<button class="secondary" type="submit" name="fill_unused" value="{{ channel.index }}">Set this channel unused</button>
</div>
</div>
{% endfor %}
<div class="btn-row">
<button type="submit">Save schema to device</button>
<button class="secondary" type="submit" name="normalize" value="1">Normalize invert flags from checkboxes</button>
<button class="warn" type="submit" name="fill_all_unused" value="1">Fill all with 255/0</button>
</div>
</form>
</div>
<div class="panel">
<h2>Raw schema</h2>
<div class="mono small">{{ schema | join(', ') }}</div>
</div>
{% endif %}
</div>
</body>
</html>
"""
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)
