""" gcode_store.py - persistent per-bridge SQLite store for uploaded GCode files and print-job history. Extracted from kobrax_moonraker_bridge.py; re-exported from there so existing imports keep working. ──────────────────────────────────────────────────────────────────────────── Copyright (C) 2026 viewit (KX-Bridge contributors) Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md. """ import os import json import time import uuid import sqlite3 import threading class GCodeStore: """Persistenter GCode-Store pro Bridge-Instanz (SQLite).""" def __init__(self, data_dir: str): os.makedirs(data_dir, exist_ok=True) self._gcode_dir = os.path.join(data_dir, "gcodes") os.makedirs(self._gcode_dir, exist_ok=True) db_path = os.path.join(data_dir, "kx-bridge.db") self._conn = sqlite3.connect(db_path, check_same_thread=False) self._conn.row_factory = sqlite3.Row self._lock = threading.Lock() self._init_schema() def _init_schema(self): with self._lock: self._conn.executescript(""" CREATE TABLE IF NOT EXISTS gcode_files ( id TEXT PRIMARY KEY, filename TEXT NOT NULL, path TEXT NOT NULL, size_bytes INTEGER NOT NULL, uploaded_at TEXT NOT NULL, thumbnail_b64 TEXT, est_print_time_sec INTEGER, filament_used_mm REAL, layer_count INTEGER, gcode_filaments TEXT, objects_skip_parts TEXT, svg_image TEXT, web_unverified INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS print_jobs ( id TEXT PRIMARY KEY, gcode_file_id TEXT NOT NULL, printer_id TEXT NOT NULL, started_at TEXT NOT NULL, ended_at TEXT, status TEXT NOT NULL, duration_sec INTEGER, filament_assignments TEXT, abort_reason TEXT ); """) # Migration: add gcode_filaments column for older databases try: self._conn.execute("ALTER TABLE gcode_files ADD COLUMN gcode_filaments TEXT") self._conn.commit() except Exception: pass # Migration: columns objects_skip_parts + svg_image (part-skip feature, v0.9.10) # Plus layer_height / first_layer_height (Obico Z height, v0.9.18) for col, typ in ( ("objects_skip_parts", "TEXT"), ("svg_image", "TEXT"), ("layer_height", "REAL"), ("first_layer_height", "REAL"), ): try: self._conn.execute(f"ALTER TABLE gcode_files ADD COLUMN {col} {typ}") self._conn.commit() except Exception: pass # Migration: flag for web uploads (warning before print) try: self._conn.execute("ALTER TABLE gcode_files ADD COLUMN web_unverified INTEGER NOT NULL DEFAULT 0") self._conn.commit() except Exception: pass def save_file(self, file_id: str, filename: str, data: bytes, est_time_sec: int = 0, thumbnail_b64: str = "", gcode_filaments: list | None = None, web_unverified: bool = False, layer_height: float = 0.0, first_layer_height: float = 0.0) -> str: """Saves a GCode file to disk and DB. Returns the path.""" safe_name = os.path.basename(filename) path = os.path.join(self._gcode_dir, safe_name) with open(path, "wb") as f: f.write(data) now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) with self._lock: filaments_json = json.dumps(gcode_filaments) if gcode_filaments else None self._conn.execute( """INSERT OR REPLACE INTO gcode_files (id, filename, path, size_bytes, uploaded_at, thumbnail_b64, est_print_time_sec, gcode_filaments, web_unverified, layer_height, first_layer_height) VALUES (?,?,?,?,?,?,?,?,?,?,?)""", (file_id, filename, path, len(data), now, thumbnail_b64 or None, est_time_sec or None, filaments_json, 1 if web_unverified else 0, layer_height or None, first_layer_height or None) ) self._conn.commit() return path def list_files(self) -> list: with self._lock: rows = self._conn.execute( "SELECT * FROM gcode_files ORDER BY uploaded_at DESC" ).fetchall() return [dict(r) for r in rows] def get_file(self, file_id: str) -> dict | None: with self._lock: row = self._conn.execute( "SELECT * FROM gcode_files WHERE id=?", (file_id,) ).fetchone() return dict(row) if row else None def get_file_by_name(self, filename: str) -> dict | None: with self._lock: row = self._conn.execute( "SELECT * FROM gcode_files WHERE filename=? ORDER BY uploaded_at DESC LIMIT 1", (filename,) ).fetchone() return dict(row) if row else None def update_file_objects(self, filename: str, objects: list, svg: str = "") -> None: """Saves the object list + optional SVG for a file (matched via filename).""" if not filename: return with self._lock: self._conn.execute( "UPDATE gcode_files SET objects_skip_parts=?, svg_image=? " "WHERE filename=?", (json.dumps(objects), svg or "", filename), ) self._conn.commit() def update_file_filaments(self, file_id: str, gcode_filaments: list | None) -> None: """Updates parsed GCode filaments for an existing DB entry.""" with self._lock: self._conn.execute( "UPDATE gcode_files SET gcode_filaments=? WHERE id=?", (json.dumps(gcode_filaments) if gcode_filaments else None, file_id), ) self._conn.commit() def clear_web_unverified(self, file_id: str) -> bool: with self._lock: cur = self._conn.execute( "UPDATE gcode_files SET web_unverified=0 WHERE id=?", (file_id,), ) self._conn.commit() return cur.rowcount > 0 def delete_file(self, file_id: str) -> bool: row = self.get_file(file_id) if not row: return False try: os.remove(row["path"]) except OSError: pass with self._lock: self._conn.execute("DELETE FROM gcode_files WHERE id=?", (file_id,)) self._conn.commit() return True def start_job(self, gcode_file_id: str, printer_id: str, filament_assignments: list | None = None) -> str: job_id = str(uuid.uuid4()) now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) assignments_json = json.dumps(filament_assignments) if filament_assignments else None with self._lock: self._conn.execute( """INSERT INTO print_jobs (id, gcode_file_id, printer_id, started_at, status, filament_assignments) VALUES (?,?,?,?,'printing',?)""", (job_id, gcode_file_id, printer_id, now, assignments_json) ) self._conn.commit() return job_id def finish_job(self, job_id: str, status: str = "completed", abort_reason: str = "") -> None: now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) with self._lock: row = self._conn.execute( "SELECT started_at FROM print_jobs WHERE id=?", (job_id,) ).fetchone() duration = None if row: try: import calendar start = time.strptime(row["started_at"], "%Y-%m-%dT%H:%M:%SZ") duration = int(time.time() - calendar.timegm(start)) except Exception: pass self._conn.execute( """UPDATE print_jobs SET ended_at=?, status=?, duration_sec=?, abort_reason=? WHERE id=?""", (now, status, duration, abort_reason or None, job_id) ) self._conn.commit() def list_jobs(self, limit: int = 50, offset: int = 0) -> list: with self._lock: rows = self._conn.execute( """SELECT j.*, f.filename, f.thumbnail_b64 FROM print_jobs j LEFT JOIN gcode_files f ON j.gcode_file_id = f.id ORDER BY j.started_at DESC LIMIT ? OFFSET ?""", (limit, offset) ).fetchall() return [dict(r) for r in rows]