forked from viewit/KX-Bridge-Release
b39577ad4d
2-column layout: settings toggle switches dashboard between 1-col (current behaviour) and 2-col (Camera|Progress, Temps|AMS side by side, motion cards in 2×2). Preference stored in localStorage only (per-browser, not printer config). Mobile always stays 1-col via CSS. Local timelapse: new TimelapseSpool class captures JPEG frames from the camera stream during a print at a configurable interval (default 4s) and encodes them to MP4 with ffmpeg on print end. Stored in data/timelapses/, indexed in SQLite. New /kx/timelapses API + browser tab in the file store panel with download and delete. Also wires the undocumented printer-native timelapse MQTT field as an experimental opt-in setting. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
277 lines
11 KiB
Python
277 lines
11 KiB
Python
"""
|
||
config_loader.py – lädt Verbindungsparameter aus config/config.ini (primär)
|
||
oder .env (Fallback / Migration).
|
||
Umgebungsvariablen haben immer Vorrang.
|
||
"""
|
||
import os
|
||
import sys
|
||
import pathlib
|
||
import configparser
|
||
|
||
_BASE = pathlib.Path(sys.executable).parent if getattr(sys, "frozen", False) else pathlib.Path(__file__).parent
|
||
|
||
CONFIG_SECTION_CONNECTION = "connection"
|
||
CONFIG_SECTION_PRINT = "print"
|
||
CONFIG_SECTION_BRIDGE = "bridge"
|
||
|
||
|
||
def _find_config_file() -> pathlib.Path | None:
|
||
for base in (_BASE, _BASE.parent):
|
||
p = base / "config" / "config.ini"
|
||
if p.is_file():
|
||
return p
|
||
return None
|
||
|
||
|
||
def _find_env_file() -> pathlib.Path | None:
|
||
for base in (_BASE, _BASE.parent):
|
||
p = base / ".env"
|
||
if p.is_file():
|
||
return p
|
||
return None
|
||
|
||
|
||
def _load_env_file(path: pathlib.Path):
|
||
"""Lädt .env-Datei als Fallback – setzt nur Keys die noch nicht in os.environ sind."""
|
||
with open(path, encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, val = line.partition("=")
|
||
key = key.strip()
|
||
val = val.strip()
|
||
if key and key not in os.environ:
|
||
os.environ[key] = val
|
||
|
||
|
||
def _load_config_file(path: pathlib.Path):
|
||
"""Lädt config.ini und setzt Keys in os.environ (nur wenn nicht bereits gesetzt)."""
|
||
cfg = configparser.ConfigParser()
|
||
cfg.read(path, encoding="utf-8")
|
||
|
||
mapping = {
|
||
"PRINTER_IP": (CONFIG_SECTION_CONNECTION, "printer_ip"),
|
||
"MQTT_PORT": (CONFIG_SECTION_CONNECTION, "mqtt_port"),
|
||
"MQTT_USERNAME": (CONFIG_SECTION_CONNECTION, "username"),
|
||
"MQTT_PASSWORD": (CONFIG_SECTION_CONNECTION, "password"),
|
||
"MODE_ID": (CONFIG_SECTION_CONNECTION, "mode_id"),
|
||
"DEVICE_ID": (CONFIG_SECTION_CONNECTION, "device_id"),
|
||
"DEFAULT_AMS_SLOT": (CONFIG_SECTION_PRINT, "default_ams_slot"),
|
||
"AUTO_LEVELING": (CONFIG_SECTION_PRINT, "auto_leveling"),
|
||
"VIBRATION_COMPENSATION": (CONFIG_SECTION_PRINT, "vibration_compensation"),
|
||
"FLOW_CALIBRATION": (CONFIG_SECTION_PRINT, "flow_calibration"),
|
||
"CAMERA_ON_PRINT": (CONFIG_SECTION_PRINT, "camera_on_print"),
|
||
"WEB_UPLOAD_WARNING": (CONFIG_SECTION_PRINT, "web_upload_warning"),
|
||
"TIMELAPSE_LOCAL": (CONFIG_SECTION_PRINT, "timelapse_local"),
|
||
"TIMELAPSE_INTERVAL_SEC": (CONFIG_SECTION_PRINT, "timelapse_interval_sec"),
|
||
"TIMELAPSE_PRINTER": (CONFIG_SECTION_PRINT, "timelapse_printer"),
|
||
"BRIDGE_PRINTER_NAME": (CONFIG_SECTION_BRIDGE, "printer_name"),
|
||
}
|
||
for env_key, (section, option) in mapping.items():
|
||
if env_key not in os.environ:
|
||
try:
|
||
val = cfg.get(section, option)
|
||
if val:
|
||
os.environ[env_key] = val
|
||
except (configparser.NoSectionError, configparser.NoOptionError):
|
||
pass
|
||
|
||
|
||
def migrate_env_to_config(env_path: pathlib.Path, config_path: pathlib.Path):
|
||
"""Einmalige Migration: .env → config.ini anlegen."""
|
||
env_vals: dict[str, str] = {}
|
||
with open(env_path, encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
k, _, v = line.partition("=")
|
||
env_vals[k.strip()] = v.strip()
|
||
|
||
config_path.parent.mkdir(parents=True, exist_ok=True)
|
||
cfg = configparser.ConfigParser()
|
||
cfg[CONFIG_SECTION_CONNECTION] = {
|
||
"printer_ip": env_vals.get("PRINTER_IP", ""),
|
||
"mqtt_port": env_vals.get("MQTT_PORT", "9883"),
|
||
"username": env_vals.get("MQTT_USERNAME", ""),
|
||
"password": env_vals.get("MQTT_PASSWORD", ""),
|
||
"mode_id": env_vals.get("MODE_ID", ""),
|
||
"device_id": env_vals.get("DEVICE_ID", ""),
|
||
}
|
||
cfg[CONFIG_SECTION_PRINT] = {
|
||
"default_ams_slot": env_vals.get("DEFAULT_AMS_SLOT", "auto"),
|
||
"auto_leveling": env_vals.get("AUTO_LEVELING", "1"),
|
||
"vibration_compensation": env_vals.get("VIBRATION_COMPENSATION", "0"),
|
||
"flow_calibration": env_vals.get("FLOW_CALIBRATION", "0"),
|
||
"camera_on_print": env_vals.get("CAMERA_ON_PRINT", "0"),
|
||
"web_upload_warning": env_vals.get("WEB_UPLOAD_WARNING", "1"),
|
||
"timelapse_local": env_vals.get("TIMELAPSE_LOCAL", "0"),
|
||
"timelapse_interval_sec": env_vals.get("TIMELAPSE_INTERVAL_SEC", "4"),
|
||
"timelapse_printer": env_vals.get("TIMELAPSE_PRINTER", "0"),
|
||
}
|
||
cfg[CONFIG_SECTION_BRIDGE] = {
|
||
"poll_interval": "3",
|
||
}
|
||
with open(config_path, "w", encoding="utf-8") as f:
|
||
f.write("# KX-Bridge Konfigurationsdatei\n")
|
||
f.write("# Automatisch migriert aus .env\n\n")
|
||
cfg.write(f)
|
||
|
||
|
||
def find_config_path() -> pathlib.Path:
|
||
"""Gibt den Pfad zur config.ini zurück (auch wenn sie noch nicht existiert)."""
|
||
for base in (_BASE, _BASE.parent):
|
||
config_dir = base / "config"
|
||
if config_dir.is_dir():
|
||
return config_dir / "config.ini"
|
||
return _BASE / "config" / "config.ini"
|
||
|
||
|
||
# ─── Laden ───────────────────────────────────────────────────────────────────
|
||
|
||
_config_path = _find_config_file()
|
||
_env_path = _find_env_file()
|
||
|
||
if _config_path:
|
||
_load_config_file(_config_path)
|
||
elif _env_path:
|
||
# Kein config.ini vorhanden → aus .env migrieren
|
||
_target = find_config_path()
|
||
migrate_env_to_config(_env_path, _target)
|
||
_load_config_file(_target)
|
||
_config_path = _target
|
||
|
||
|
||
def list_printers() -> list[dict]:
|
||
"""Liest alle [printer_N]-Sektionen aus config.ini.
|
||
|
||
Jede Sektion kann folgende Keys haben:
|
||
name, printer_ip, mqtt_port, username, password, mode_id, device_id,
|
||
bridge_url, default_ams_slot, auto_leveling
|
||
|
||
Gibt eine leere Liste zurück wenn keine [printer_N]-Sektionen vorhanden sind
|
||
(Single-Printer-Betrieb via [connection]).
|
||
"""
|
||
path = _find_config_file()
|
||
if not path:
|
||
return []
|
||
cfg = configparser.ConfigParser()
|
||
cfg.read(path, encoding="utf-8")
|
||
printers: list[dict] = []
|
||
idx = 1
|
||
while True:
|
||
section = f"printer_{idx}"
|
||
if not cfg.has_section(section):
|
||
break
|
||
p = dict(cfg[section])
|
||
p.setdefault("id", str(idx))
|
||
if "mqtt_port" in p:
|
||
try:
|
||
p["mqtt_port"] = int(p["mqtt_port"])
|
||
except ValueError:
|
||
p["mqtt_port"] = 9883
|
||
printers.append(p)
|
||
idx += 1
|
||
return printers
|
||
|
||
|
||
def list_filament_profiles() -> dict[int, dict]:
|
||
"""Liest die [filament_profiles]-Sektion aus config.ini.
|
||
|
||
Format pro AMS-Slot — primärer Selector ist (vendor, name), die `id` wird
|
||
aus der orca_filaments.json beim Speichern nachgeschlagen und mitgeführt
|
||
(als Hint für OrcaSlicer; das Orca-Datenmodell hat ~136 Profile mit
|
||
derselben filament_id wie 'OGFL99', d.h. die ID ist nicht eindeutig):
|
||
|
||
[filament_profiles]
|
||
slot_0_vendor = Polymaker
|
||
slot_0_name = PolyTerra PLA
|
||
slot_0_id = OGFL01
|
||
|
||
Gibt einen Dict {slot_index: {"id": ..., "vendor": ..., "name": ...}}
|
||
zurück. Leere/fehlende Slots werden NICHT aufgenommen — das Default-Mapping
|
||
(per filament_type) in der Bridge bleibt dann aktiv.
|
||
|
||
Backwards-Kompat: alte Configs mit nur (vendor, id) bleiben lesbar; `name`
|
||
fehlt dann und der Aufrufer kann optional aus der orca_filaments.json
|
||
rekonstruieren.
|
||
"""
|
||
path = _find_config_file()
|
||
if not path:
|
||
return {}
|
||
cfg = configparser.ConfigParser()
|
||
cfg.read(path, encoding="utf-8")
|
||
if not cfg.has_section("filament_profiles"):
|
||
return {}
|
||
result: dict[int, dict] = {}
|
||
for key, value in cfg.items("filament_profiles"):
|
||
# Erwartet: slot_<idx>_id oder slot_<idx>_vendor oder slot_<idx>_name
|
||
if not key.startswith("slot_"):
|
||
continue
|
||
parts = key.split("_", 2)
|
||
if len(parts) < 3:
|
||
continue
|
||
try:
|
||
slot_idx = int(parts[1])
|
||
except ValueError:
|
||
continue
|
||
field = parts[2]
|
||
if field not in ("id", "vendor", "name"):
|
||
continue
|
||
if not value.strip():
|
||
continue
|
||
result.setdefault(slot_idx, {})[field] = value.strip()
|
||
return result
|
||
|
||
|
||
def save_filament_profiles(profiles: dict[int, dict]) -> bool:
|
||
"""Schreibt die übergebenen Slot-Profile in die [filament_profiles]-
|
||
Sektion der config.ini. Existierende Einträge werden komplett ersetzt.
|
||
|
||
profiles: {slot_index: {"id": "OGFL01", "vendor": "Polymaker", "name": "PolyTerra PLA"}}
|
||
Mindestens vendor+name müssen gesetzt sein; id ist optional (Hint).
|
||
"""
|
||
path = _find_config_file()
|
||
if not path:
|
||
return False
|
||
cfg = configparser.ConfigParser()
|
||
cfg.read(path, encoding="utf-8")
|
||
if cfg.has_section("filament_profiles"):
|
||
cfg.remove_section("filament_profiles")
|
||
if profiles:
|
||
cfg["filament_profiles"] = {}
|
||
for slot_idx in sorted(profiles.keys()):
|
||
entry = profiles[slot_idx] or {}
|
||
if entry.get("vendor"):
|
||
cfg["filament_profiles"][f"slot_{slot_idx}_vendor"] = entry["vendor"]
|
||
if entry.get("name"):
|
||
cfg["filament_profiles"][f"slot_{slot_idx}_name"] = entry["name"]
|
||
if entry.get("id"):
|
||
cfg["filament_profiles"][f"slot_{slot_idx}_id"] = entry["id"]
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
cfg.write(f)
|
||
return True
|
||
|
||
|
||
def get(key: str, default: str = "") -> str:
|
||
return os.environ.get(key, default)
|
||
|
||
|
||
# Häufig verwendete Shortcuts
|
||
PRINTER_IP = get("PRINTER_IP", "")
|
||
MQTT_PORT = int(get("MQTT_PORT", "9883"))
|
||
USERNAME = get("MQTT_USERNAME", "")
|
||
PASSWORD = get("MQTT_PASSWORD", "")
|
||
MODE_ID = get("MODE_ID", "")
|
||
DEVICE_ID = get("DEVICE_ID", "")
|
||
DEFAULT_AMS_SLOT = get("DEFAULT_AMS_SLOT", "auto")
|
||
AUTO_LEVELING = int(get("AUTO_LEVELING", "1"))
|
||
VIBRATION_COMPENSATION = int(get("VIBRATION_COMPENSATION", "0"))
|
||
FLOW_CALIBRATION = int(get("FLOW_CALIBRATION", "0"))
|
||
CAMERA_ON_PRINT = int(get("CAMERA_ON_PRINT", "0"))
|
||
WEB_UPLOAD_WARNING = int(get("WEB_UPLOAD_WARNING", "1"))
|
||
TIMELAPSE_LOCAL = int(get("TIMELAPSE_LOCAL", "0"))
|
||
TIMELAPSE_INTERVAL_SEC = int(get("TIMELAPSE_INTERVAL_SEC", "4"))
|
||
TIMELAPSE_PRINTER = int(get("TIMELAPSE_PRINTER", "0"))
|