"""Adaptor pentru pachetul vendorizat `scale_bridge`. Aplicatia importa direct biblioteca SyncPulse Scale Bridge (nu trece prin HTTP): bridge-ul tine portul serial deschis si ne da citirea deja parsata, cu unitate, stare de stabilitate si valoarea bruta. """ from __future__ import annotations import atexit import sys import threading from pathlib import Path from typing import Any, Optional from .config import BASE_DIR VENDOR_DIR = BASE_DIR / "vendor" / "syncpulse-scale-bridge" class ScaleError(RuntimeError): """Cantarul nu poate fi citit (port ocupat, profile lipsa, etc.).""" def _vendor_on_path() -> None: path = str(VENDOR_DIR) if path not in sys.path: sys.path.insert(0, path) class ScaleService: """O singura instanta per proces; porturile rămân deschise intre cereri.""" def __init__(self, bridge_config: str | Path, simulate: str = ""): self.bridge_config = str(bridge_config) self.simulate = simulate self._lock = threading.Lock() self._loaded = False self._config: Any = None self._registry: Any = None self._simulator: Any = None self._opener: Any = None self._error: Optional[str] = None self._closed = False # -- initializare ------------------------------------------------------ def _load(self) -> None: if self._loaded: return with self._lock: if self._loaded: return try: _vendor_on_path() from scale_bridge.config import Config as BridgeConfig from scale_bridge.serialio import Registry, open_real from scale_bridge.simulator import SimulatorBank Path(self.bridge_config).parent.mkdir(parents=True, exist_ok=True) self._config = BridgeConfig(self.bridge_config) if self.simulate: self._simulator = SimulatorBank(self.simulate) self._opener = self._simulator.opener(open_real) self._registry = Registry(self._opener) else: self._opener = open_real self._registry = Registry(self._opener) except Exception as exc: # noqa: BLE001 - afisam motivul in UI self._error = f"Scale bridge indisponibil: {exc}" finally: self._loaded = True @property def available(self) -> bool: self._load() return self._registry is not None and self._config is not None @property def error(self) -> Optional[str]: self._load() return self._error # -- profile ----------------------------------------------------------- def profiles(self) -> list[dict[str, Any]]: self._load() if self._config is None: return [] return [ { "name": p["name"], "label": p.get("label") or p["name"], "port": p.get("port", ""), "protocol": p.get("protocol", ""), } for p in self._config.profiles() ] def resolve_profile(self, name: str = "") -> Optional[dict[str, Any]]: self._load() if self._config is None: return None if name: return self._config.profile(name) profiles = self._config.profiles() return profiles[0] if profiles else None # -- citire ------------------------------------------------------------ def read( self, profile_name: str = "", *, max_age_ms: int = 1500, wait_ms: int = 2500, ) -> dict[str, Any]: """Citire normalizata, gata de trimis catre UI/API.""" self._load() if self._config is None or self._registry is None: return self._failure(self._error or "Scale bridge indisponibil.") try: from scale_bridge import profiles as bridge_profiles from scale_bridge.serialio import PortBusy, now_ms except Exception as exc: # noqa: BLE001 return self._failure(f"Scale bridge indisponibil: {exc}") profile = self.resolve_profile(profile_name) if profile is None: if profile_name: return self._failure(f"Profilul de cantar {profile_name!r} nu exista.") return self._failure("Nu exista niciun profil de cantar configurat.") try: worker = self._registry.worker_for(bridge_profiles.line_settings(profile)) result = bridge_profiles.wait_for_reading( profile, worker, int(max_age_ms), int(wait_ms), now_ms ) worker.touch() except PortBusy as exc: return self._failure(str(exc), profile=profile["name"]) except Exception as exc: # noqa: BLE001 return self._failure(f"Eroare la citirea cantarului: {exc}", profile=profile["name"]) return { "ok": bool(result.get("ok")), "error": result.get("error"), "profile": result.get("profile") or profile["name"], "port": result.get("port") or profile.get("port", ""), "protocol": result.get("protocol") or profile.get("protocol", ""), "open": result.get("open"), "value": result.get("value"), "unit": result.get("unit") or profile.get("unit") or "", "stable": bool(result.get("stable")), "stale": bool(result.get("stale")), "overload": bool(result.get("overload")), "age_ms": result.get("age_ms"), "measured_at": result.get("measured_at"), "raw_hex": result.get("raw_hex") or "", "net": result.get("net"), "fields": result.get("fields") or {}, } @staticmethod def _failure(message: str, profile: str = "") -> dict[str, Any]: return { "ok": False, "error": message, "profile": profile, "value": None, "unit": "", "stable": False, "stale": False, "overload": False, "age_ms": None, "measured_at": None, "raw_hex": "", "fields": {}, } # -- diagnostic -------------------------------------------------------- def probe( self, port: str, baud: int = 9600, framing: str = "8N1", seconds: float = 5.0, ) -> dict[str, Any]: """Asculta un port (fara sa scrie nimic) si raporteaza ce sosește. Citirea normala a cantarului tine portul deschis; pentru monitor il eliberam intai, apoi registrul il redeschide la urmatoarea citire. """ self._load() if self._registry is None: return {"ok": False, "port": port, "error": self._error or "Scale bridge indisponibil.", "frames": [], "guesses": [], "bytes": 0} try: from .probe import probe_port self._registry.release(port) return probe_port(port, baud, framing, seconds, opener=self._opener) except Exception as exc: # noqa: BLE001 - eroarea ajunge in pagina de monitor return {"ok": False, "port": port, "error": str(exc), "frames": [], "guesses": [], "bytes": 0} # -- oprire ------------------------------------------------------------ def close(self) -> None: if self._closed: return self._closed = True registry = self._registry if registry is not None: try: registry.stop_all() except Exception: # noqa: BLE001 pass def get_scale(app) -> ScaleService: service = app.extensions.get("scale") if service is None: # pragma: no cover - create_app il creeaza mereu service = ScaleService(app.config["BRIDGE_CONFIG"], app.config["SIMULATE"]) app.extensions["scale"] = service return service def create_scale(app) -> ScaleService: service = ScaleService(app.config["BRIDGE_CONFIG"], app.config["SIMULATE"]) app.extensions["scale"] = service atexit.register(service.close) return service