Jonah
429 lines
14 KiB
Python
429 lines
14 KiB
Python
"""Roland cut file and LightBurn file discovery and opening.
|
|
|
|
Searches a configured SMB folder (recursively) for files whose name matches
|
|
a SKU, opens the chosen file with the system default application, and
|
|
persists a list of saved folder paths so the user can swap between locations.
|
|
|
|
The search is on a worker thread because SMB roots can be slow.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
CUT_FILE_EXTENSIONS: tuple[str, ...] = (
|
|
".eps", ".ai", ".svg", ".pdf", ".plt", ".gsd", ".dxf", ".cdr", ".ccx",
|
|
)
|
|
|
|
LIGHTBURN_EXTENSIONS: tuple[str, ...] = (
|
|
".lbrn", ".lbrn2",
|
|
)
|
|
|
|
|
|
def _walk_files(base: Path) -> list[Path]:
|
|
"""Yield every file under `base` (recursive), tolerant of per-subfolder errors.
|
|
|
|
`Path.rglob("*")` raises on the first permission error; `os.walk` with an
|
|
error handler lets us skip individual subdirectories.
|
|
"""
|
|
results: list[Path] = []
|
|
|
|
def _on_error(err: OSError) -> None:
|
|
log.debug("Walk skipped (likely permission/IO): %s", err)
|
|
|
|
for root, _dirs, files in os.walk(base, onerror=_on_error):
|
|
for name in files:
|
|
results.append(Path(root) / name)
|
|
return results
|
|
|
|
|
|
def extract_sku_suffix(sku: str) -> str | None:
|
|
"""Pull the part-code suffix off a SKU.
|
|
|
|
Example: ``'8620843-0423L'`` -> ``'0423L'``. Returns ``None`` if the SKU
|
|
has no dash or has nothing after the first dash.
|
|
"""
|
|
if "-" not in sku:
|
|
return None
|
|
suffix = sku.split("-", 1)[1]
|
|
return suffix if suffix else None
|
|
|
|
|
|
def make_lightburn_sku(sku: str) -> str | None:
|
|
"""Derive the LightBurn filename SKU from a part SKU.
|
|
|
|
LightBurn files all share the internal part number ``'E000'`` as their
|
|
prefix followed by the portion of the part SKU after the first dash.
|
|
Returns ``None`` if no suffix can be extracted.
|
|
"""
|
|
suffix = extract_sku_suffix(sku)
|
|
if suffix is None:
|
|
return None
|
|
return "E000-" + suffix
|
|
|
|
|
|
def _whole_token_pattern(token: str) -> re.Pattern:
|
|
"""Return a compiled regex that matches `token` as a whole token in a filename."""
|
|
escaped = re.escape(token)
|
|
return re.compile(
|
|
r"(?:^|[^a-z0-9])" + escaped + r"(?:[^a-z0-9]|$)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def find_cut_files(base_dir: str, sku: str, *, limit: int = 40) -> list[Path]:
|
|
"""Find cut files anywhere under base_dir that match the SKU.
|
|
|
|
Match priority (highest first):
|
|
1. Exact stem match against the full SKU
|
|
2. Exact stem match against the suffix (portion after the first dash)
|
|
|
|
Returns a list of matching Paths, most specific match first.
|
|
"""
|
|
base = Path(base_dir)
|
|
if not base.exists():
|
|
log.warning("Cut file base dir does not exist: %s", base_dir)
|
|
return []
|
|
|
|
try:
|
|
all_files = _walk_files(base)
|
|
except Exception as exc:
|
|
log.error("Cut file base dir unreachable: %s (%s)", base_dir, exc)
|
|
return []
|
|
|
|
suffix = extract_sku_suffix(sku)
|
|
search_tokens = [sku]
|
|
if suffix and suffix.lower() != sku.lower():
|
|
search_tokens.append(suffix)
|
|
|
|
patterns = [_whole_token_pattern(t) for t in search_tokens]
|
|
|
|
matches: list[Path] = []
|
|
seen: set[str] = set()
|
|
for path in all_files:
|
|
stem = path.stem
|
|
for pat in patterns:
|
|
if pat.search(stem):
|
|
key = str(path)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
matches.append(path)
|
|
break
|
|
if len(matches) >= limit:
|
|
break
|
|
|
|
return matches
|
|
|
|
|
|
def open_file(path: str) -> bool:
|
|
"""Open a file with the system default application. Cross-platform."""
|
|
try:
|
|
if sys.platform == "win32":
|
|
os.startfile(path)
|
|
elif sys.platform == "darwin":
|
|
subprocess.Popen(["open", path])
|
|
else:
|
|
subprocess.Popen(["xdg-open", path])
|
|
return True
|
|
except Exception as exc:
|
|
log.error("Failed to open %s: %s", path, exc)
|
|
return False
|
|
|
|
|
|
def reveal_in_explorer(path: str) -> bool:
|
|
"""Open the file's containing folder with the file selected (Windows)."""
|
|
try:
|
|
if sys.platform == "win32":
|
|
subprocess.Popen(["explorer", f"/select,{path}"])
|
|
elif sys.platform == "darwin":
|
|
subprocess.Popen(["open", "-R", path])
|
|
else:
|
|
subprocess.Popen(["xdg-open", str(Path(path).parent)])
|
|
return True
|
|
except Exception as exc:
|
|
log.error("Failed to reveal %s: %s", path, exc)
|
|
return False
|
|
|
|
|
|
class CutFileIndex:
|
|
"""In-memory snapshot of all cut files under one folder.
|
|
|
|
The first search of a session walks the SMB share once (potentially slow)
|
|
and caches the result. Subsequent lookups are O(1) dict access.
|
|
"""
|
|
|
|
def __init__(self, extensions: tuple[str, ...]) -> None:
|
|
self._extensions = extensions
|
|
self._indexed_path: str | None = None
|
|
self._built_at: float | None = None
|
|
self._index: dict[str, list[Path]] = {}
|
|
self._cut_file_count = 0
|
|
self._total_files_seen = 0
|
|
self._dirs_walked = 0
|
|
|
|
@property
|
|
def file_count(self) -> int:
|
|
return self._cut_file_count
|
|
|
|
@property
|
|
def indexed_path(self) -> str | None:
|
|
return self._indexed_path
|
|
|
|
@property
|
|
def built_at(self) -> float | None:
|
|
return self._built_at
|
|
|
|
@property
|
|
def stats(self) -> tuple[int, int, int]:
|
|
"""(cut_files_indexed, total_files_seen, dirs_walked)."""
|
|
return (self._cut_file_count, self._total_files_seen, self._dirs_walked)
|
|
|
|
def is_valid_for(self, path: str | None) -> bool:
|
|
"""True if the index reflects `path` (even if `path` is None)."""
|
|
return self._indexed_path == path and self._built_at is not None
|
|
|
|
def invalidate(self) -> None:
|
|
self._indexed_path = None
|
|
self._built_at = None
|
|
self._index.clear()
|
|
self._cut_file_count = 0
|
|
self._total_files_seen = 0
|
|
self._dirs_walked = 0
|
|
|
|
def build(self, path: str | None) -> int:
|
|
"""Walk `path` and replace the index. Returns the cut-file count.
|
|
|
|
Also records dirs_walked and files_seen so the UI can show a summary.
|
|
"""
|
|
if not path:
|
|
self.invalidate()
|
|
return 0
|
|
|
|
base = Path(path)
|
|
if not base.exists():
|
|
log.warning("Cut file base dir does not exist: %s", path)
|
|
self.invalidate()
|
|
return 0
|
|
|
|
new_index: dict[str, list[Path]] = {}
|
|
cut_count = 0
|
|
total_count = 0
|
|
dir_count = 0
|
|
|
|
def _on_error(err: OSError) -> None:
|
|
log.debug("Walk skipped (likely permission/IO): %s", err)
|
|
|
|
try:
|
|
for root, dirs, files in os.walk(base, onerror=_on_error):
|
|
dir_count += 1
|
|
for name in files:
|
|
total_count += 1
|
|
p = Path(root) / name
|
|
if p.suffix.lower() not in self._extensions:
|
|
continue
|
|
cut_count += 1
|
|
token = p.stem.lower()
|
|
new_index.setdefault(token, []).append(p)
|
|
except Exception as exc:
|
|
log.error("Cut file index build failed in %s: %s", path, exc)
|
|
self.invalidate()
|
|
return 0
|
|
|
|
self._index = new_index
|
|
self._indexed_path = path
|
|
self._built_at = time.monotonic()
|
|
self._cut_file_count = cut_count
|
|
self._total_files_seen = total_count
|
|
self._dirs_walked = dir_count
|
|
|
|
log.info(
|
|
"Cut file index built: %d cut files (of %d total) across %d folders under %s",
|
|
cut_count, total_count, dir_count, path,
|
|
)
|
|
return cut_count
|
|
|
|
def lookup(self, sku: str) -> list[Path]:
|
|
"""Match a SKU against the cached index using the same ranking as
|
|
`find_cut_files` (exact, suffix-as-token, full-SKU substring).
|
|
"""
|
|
suffix = extract_sku_suffix(sku)
|
|
search_tokens = [sku.lower()]
|
|
if suffix and suffix.lower() != sku.lower():
|
|
search_tokens.append(suffix.lower())
|
|
|
|
seen: set[str] = set()
|
|
results: list[Path] = []
|
|
for token_query in search_tokens:
|
|
pat = _whole_token_pattern(token_query)
|
|
for stem_token, paths in self._index.items():
|
|
if pat.search(stem_token):
|
|
for p in paths:
|
|
key = str(p)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
results.append(p)
|
|
return results
|
|
|
|
|
|
class CutFileManager:
|
|
"""Holds saved cut-file folder paths, the active selection, and the
|
|
in-memory file index.
|
|
|
|
Persists path config to the shared ui_state.json dict via `save_fn` so
|
|
each workstation can have its own set of folders. Pass different
|
|
`state_key_paths` / `state_key_active` values to reuse this class for
|
|
both Roland cut files and LightBurn files.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
state: dict,
|
|
save_fn: Callable[[dict], None],
|
|
extensions: tuple[str, ...],
|
|
state_key_paths: str | None = None,
|
|
state_key_active: str | None = None,
|
|
) -> None:
|
|
self._state = state
|
|
self._save_fn = save_fn
|
|
self._extensions = extensions
|
|
self._key_paths = state_key_paths or "cut_file_paths"
|
|
self._key_active = state_key_active or "active_cut_file_path"
|
|
|
|
raw = state.get(self._key_paths, [])
|
|
self._paths: list[dict[str, str]] = [
|
|
p for p in (raw if isinstance(raw, list) else [])
|
|
if isinstance(p, dict) and "name" in p and "path" in p
|
|
]
|
|
self._active_name: str = state.get(self._key_active, "") or ""
|
|
if self._active_name and not any(p["name"] == self._active_name for p in self._paths):
|
|
self._active_name = self._paths[0]["name"] if self._paths else ""
|
|
|
|
self._index = CutFileIndex(extensions)
|
|
self._lock = threading.Lock()
|
|
|
|
# ── Accessors ──────────────────────────────────────────────────────────────
|
|
|
|
def get_paths(self) -> list[dict[str, str]]:
|
|
return list(self._paths)
|
|
|
|
def get_active_name(self) -> str | None:
|
|
return self._active_name or None
|
|
|
|
def get_active_path(self) -> str | None:
|
|
for p in self._paths:
|
|
if p["name"] == self._active_name:
|
|
return p["path"]
|
|
return None
|
|
|
|
def index_file_count(self) -> int:
|
|
return self._index.file_count
|
|
|
|
def index_built_at(self) -> float | None:
|
|
return self._index.built_at
|
|
|
|
def index_stats(self) -> tuple[int, int, int]:
|
|
"""(cut_files_indexed, total_files_seen, dirs_walked)."""
|
|
return self._index.stats
|
|
|
|
def index_is_fresh_for_active(self) -> bool:
|
|
return self._index.is_valid_for(self.get_active_path())
|
|
|
|
# ── Mutations (each persists immediately) ──────────────────────────────────
|
|
|
|
def add_or_update(self, name: str, path: str) -> None:
|
|
for p in self._paths:
|
|
if p["name"] == name:
|
|
p["path"] = path
|
|
self._index.invalidate()
|
|
self._persist()
|
|
return
|
|
self._paths.append({"name": name, "path": path})
|
|
if not self._active_name:
|
|
self._active_name = name
|
|
self._index.invalidate()
|
|
self._persist()
|
|
|
|
def remove(self, name: str) -> None:
|
|
self._paths = [p for p in self._paths if p["name"] != name]
|
|
if self._active_name == name:
|
|
self._active_name = self._paths[0]["name"] if self._paths else ""
|
|
self._index.invalidate()
|
|
self._persist()
|
|
|
|
def set_active(self, name: str) -> None:
|
|
if any(p["name"] == name for p in self._paths):
|
|
if self._active_name != name:
|
|
self._active_name = name
|
|
self._index.invalidate()
|
|
self._persist()
|
|
|
|
def _persist(self) -> None:
|
|
self._state[self._key_paths] = list(self._paths)
|
|
self._state[self._key_active] = self._active_name
|
|
try:
|
|
self._save_fn(self._state)
|
|
except Exception as exc:
|
|
log.warning("Failed to persist cut file state: %s", exc)
|
|
|
|
# ── Search ─────────────────────────────────────────────────────────────────
|
|
|
|
def search_async(
|
|
self,
|
|
sku: str,
|
|
callback: Callable[[list[Path]], None],
|
|
) -> None:
|
|
"""Look up a SKU. Builds the index on first use or after a path switch."""
|
|
active_path = self.get_active_path()
|
|
|
|
def worker() -> None:
|
|
try:
|
|
with self._lock:
|
|
if not self._index.is_valid_for(active_path):
|
|
self._index.build(active_path)
|
|
results = self._index.lookup(sku)
|
|
callback(results)
|
|
except Exception:
|
|
log.exception("Cut file search blew up")
|
|
callback([])
|
|
|
|
threading.Thread(target=worker, daemon=True).start()
|
|
|
|
def refresh_index_async(
|
|
self,
|
|
callback: Callable[[int], None] | None = None,
|
|
) -> None:
|
|
"""Force-rebuild the index against the current active path."""
|
|
active_path = self.get_active_path()
|
|
|
|
def worker() -> None:
|
|
try:
|
|
with self._lock:
|
|
count = self._index.build(active_path)
|
|
if callback:
|
|
callback(count)
|
|
except Exception:
|
|
log.exception("Cut file index refresh failed")
|
|
if callback:
|
|
callback(0)
|
|
|
|
threading.Thread(target=worker, daemon=True).start()
|
|
|
|
# ── State serialisation (for destroy()) ────────────────────────────────────
|
|
|
|
def to_state(self) -> dict:
|
|
"""Return the manager's portion of ui_state.json."""
|
|
return {
|
|
self._key_paths: list(self._paths),
|
|
self._key_active: self._active_name,
|
|
}
|