152 lines
6.4 KiB
Python
152 lines
6.4 KiB
Python
"""ENV.TXT for the sensorbox SD card.
|
|
|
|
Parsing mirrors EnvironmentService::read_env_var in the firmware exactly:
|
|
lines read with a 512-byte buffer, split at the FIRST '=', key compared
|
|
verbatim (no trimming), CR/LF removed from the value only, first match wins.
|
|
So the checks below report what would actually break a box - e.g. a UTF-8
|
|
BOM (Windows Notepad) hides the first key, and "KEY = value" is not "KEY".
|
|
|
|
The firmware FAT driver has no long-file-name support: the file must be
|
|
the 8.3 name ENV.TXT (case-insensitive) in the card root.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
ENV_NAME = "ENV.TXT"
|
|
BUFFER_NAME = "BUFFER.TXT" # offline measurements waiting for upload
|
|
LINE_BUF = 512 # char line[512] in the firmware
|
|
REQUIRED = ("INFLUX_URL", "INFLUX_TOKEN")
|
|
OPTIONAL = ("SSID", "PASSWORD")
|
|
_BOM = b"\xef\xbb\xbf"
|
|
|
|
|
|
@dataclass
|
|
class Issue:
|
|
level: str # "error" | "warning" | "info"
|
|
text: str
|
|
|
|
|
|
@dataclass
|
|
class EnvConfig:
|
|
influx_url: str = ""
|
|
influx_token: str = ""
|
|
ssid: str = ""
|
|
password: str = ""
|
|
|
|
@property
|
|
def local_wifi(self) -> bool:
|
|
return bool(self.ssid or self.password)
|
|
|
|
def lines(self) -> list[tuple[str, str]]:
|
|
out = [("INFLUX_TOKEN", self.influx_token), ("INFLUX_URL", self.influx_url)]
|
|
if self.local_wifi:
|
|
out += [("SSID", self.ssid), ("PASSWORD", self.password)]
|
|
return out
|
|
|
|
def render(self) -> bytes:
|
|
# LF only, no BOM, no quoting: exactly what the firmware parser expects.
|
|
return "".join(f"{k}={v}\n" for k, v in self.lines()).encode("utf-8")
|
|
|
|
def problems(self) -> list[Issue]:
|
|
issues: list[Issue] = []
|
|
for key, value in self.lines():
|
|
if any(c in value for c in "\r\n"):
|
|
issues.append(Issue("error", f"{key} contains a line break."))
|
|
if value != value.strip():
|
|
issues.append(Issue("error", f"{key} starts or ends with a space; the box would use the space too."))
|
|
if len(f"{key}={value}".encode()) >= LINE_BUF - 1:
|
|
issues.append(Issue("error", f"{key} is too long for the box (max {LINE_BUF - 2} bytes per line)."))
|
|
url = self.influx_url.strip()
|
|
if not url:
|
|
issues.append(Issue("error", "INFLUX_URL is empty."))
|
|
elif not re.match(r"^https?://", url):
|
|
issues.append(Issue("error", "INFLUX_URL must start with https://"))
|
|
elif not ("/api/v2/write" in url and "bucket=" in url and "org=" in url):
|
|
issues.append(Issue("error", "INFLUX_URL must be a write URL: .../api/v2/write?bucket=...&org=..."))
|
|
elif url.startswith("http://"):
|
|
issues.append(Issue("warning", "INFLUX_URL uses http:// (unencrypted)."))
|
|
if not self.influx_token:
|
|
issues.append(Issue("error", "INFLUX_TOKEN is empty."))
|
|
elif re.search(r"\s", self.influx_token):
|
|
issues.append(Issue("error", "INFLUX_TOKEN contains spaces."))
|
|
elif len(self.influx_token) < 20:
|
|
issues.append(Issue("warning", "INFLUX_TOKEN looks too short."))
|
|
if self.local_wifi:
|
|
if not self.ssid or not self.password:
|
|
issues.append(Issue("error", "Wi-Fi needs both SSID and PASSWORD (the box ignores one without the other)."))
|
|
if len(self.ssid.encode()) > 32:
|
|
issues.append(Issue("error", "SSID is longer than 32 bytes."))
|
|
pw = self.password
|
|
if pw and not (8 <= len(pw) <= 63 or re.fullmatch(r"[0-9A-Fa-f]{64}", pw)):
|
|
issues.append(Issue("error", "Wi-Fi PASSWORD must be 8-63 characters (WPA2)."))
|
|
return issues
|
|
|
|
|
|
@dataclass
|
|
class EnvCheck:
|
|
values: dict[str, str]
|
|
issues: list[Issue]
|
|
|
|
@property
|
|
def ok(self) -> bool:
|
|
return not any(i.level == "error" for i in self.issues)
|
|
|
|
def masked(self) -> dict[str, str]:
|
|
out = {}
|
|
for k, v in self.values.items():
|
|
out[k] = (v[:4] + "..." + v[-4:] if len(v) > 12 else "***") if k in ("INFLUX_TOKEN", "PASSWORD") else v
|
|
return out
|
|
|
|
|
|
def parse_env(raw: bytes) -> EnvCheck:
|
|
"""Read ENV.TXT the way the firmware does and report what it would see."""
|
|
issues: list[Issue] = []
|
|
if raw.startswith(_BOM):
|
|
issues.append(Issue("error", "File starts with a UTF-8 BOM (saved by Notepad?): the box cannot read the first key. "
|
|
"Re-write the card with the tool."))
|
|
text = raw.decode("utf-8", errors="replace")
|
|
values: dict[str, str] = {}
|
|
for n, line in enumerate(text.split("\n"), 1):
|
|
if len(line.encode()) >= LINE_BUF - 1:
|
|
issues.append(Issue("error", f"Line {n} is longer than the box can read."))
|
|
if "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
value = value.replace("\r", "")
|
|
if key != key.strip():
|
|
issues.append(Issue("error", f"Line {n}: spaces around the key '{key.strip()}' - the box will not find it."))
|
|
if key in values:
|
|
issues.append(Issue("warning", f"{key} appears twice; the box uses the first one."))
|
|
continue
|
|
values[key] = value
|
|
for key in REQUIRED:
|
|
if key not in values:
|
|
issues.append(Issue("error", f"{key} is missing - the box stops at start-up (white LED)."))
|
|
elif not values[key]:
|
|
issues.append(Issue("error", f"{key} is empty."))
|
|
if values.get("HTTP_SERVER") == "true":
|
|
issues.append(Issue("warning", "HTTP_SERVER=true: a box with newer firmware starts in Wi-Fi setup mode."))
|
|
if bool(values.get("SSID")) != bool(values.get("PASSWORD")):
|
|
issues.append(Issue("warning", "Only one of SSID / PASSWORD is set; the box ignores Wi-Fi settings."))
|
|
elif values.get("SSID"):
|
|
issues.append(Issue("info", "Wi-Fi settings present: newer firmware uses this Wi-Fi instead of Matter "
|
|
"(the factory v1.0 firmware ignores them)."))
|
|
return EnvCheck(values, issues)
|
|
|
|
|
|
def load_template(path: Path) -> EnvConfig:
|
|
"""Settings from an existing ENV.TXT (e.g. 002-SD_card_format/ENV.txt)."""
|
|
v = parse_env(path.read_bytes()).values
|
|
return EnvConfig(v.get("INFLUX_URL", ""), v.get("INFLUX_TOKEN", ""), v.get("SSID", ""), v.get("PASSWORD", ""))
|
|
|
|
|
|
def find_case_insensitive(root: Path, name: str) -> Path | None:
|
|
for p in root.iterdir() if root.is_dir() else ():
|
|
if p.is_file() and p.name.lower() == name.lower():
|
|
return p
|
|
return None
|