392acef28e
- filamentdb.spool_counts(): builds (material, color_name) → roll count from cached stock report; counts rolls_in_stock + rolls_in_use (non-graveyard) - Jobs list: coloured dot before material — green (2+), yellow (1), red (0) - FILAMENTDB_URL added to .env Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
import logging
|
|
import os
|
|
import time
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger("pops")
|
|
|
|
_cache: dict[str, tuple[float, list]] = {}
|
|
_TTL = 300 # seconds
|
|
|
|
|
|
def _base() -> str:
|
|
return os.environ.get("FILAMENTDB_URL", "https://web.filamentdb.orb.local").rstrip("/")
|
|
|
|
|
|
def _get(path: str, params: dict | None = None) -> list | dict:
|
|
r = httpx.get(f"{_base()}{path}", params=params, timeout=5, verify=False)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def _cached(key: str, fetch):
|
|
now = time.monotonic()
|
|
if key in _cache and now - _cache[key][0] < _TTL:
|
|
return _cache[key][1]
|
|
try:
|
|
result = fetch()
|
|
_cache[key] = (now, result)
|
|
return result
|
|
except Exception:
|
|
logger.warning("FilamentDB unreachable — using cached/fallback data")
|
|
return _cache.get(key, (0, []))[1]
|
|
|
|
|
|
def get_stock() -> list[dict]:
|
|
"""All filament types with stock counts from /reports/stock."""
|
|
return _cached("stock", lambda: _get("/reports/stock"))
|
|
|
|
|
|
def available_materials() -> list[str]:
|
|
"""Unique materials that have at least one roll in stock."""
|
|
rows = get_stock()
|
|
seen, out = set(), []
|
|
for r in rows:
|
|
m = r.get("material", "")
|
|
if m and r.get("rolls_in_stock", 0) > 0 and m not in seen:
|
|
seen.add(m)
|
|
out.append(m)
|
|
return sorted(out)
|
|
|
|
|
|
def spool_counts() -> dict[tuple[str, str], int]:
|
|
"""Returns {(material, color_name): available_rolls} from cached stock.
|
|
|
|
'Available' = rolls_in_stock + rolls_in_use (anything not in the graveyard).
|
|
Multiple filament types with the same material+color are summed.
|
|
"""
|
|
result: dict[tuple[str, str], int] = {}
|
|
for r in get_stock():
|
|
key = (r.get("material") or "", r.get("color_name") or "")
|
|
count = (r.get("rolls_in_stock") or 0) + (r.get("rolls_in_use") or 0)
|
|
result[key] = result.get(key, 0) + count
|
|
return result
|
|
|
|
|
|
def available_colors(material: str | None = None) -> list[str]:
|
|
"""Unique color names in stock, optionally filtered by material."""
|
|
rows = get_stock()
|
|
seen, out = set(), []
|
|
for r in rows:
|
|
if r.get("rolls_in_stock", 0) <= 0:
|
|
continue
|
|
if material and r.get("material") != material:
|
|
continue
|
|
c = r.get("color_name") or ""
|
|
if c and c not in seen:
|
|
seen.add(c)
|
|
out.append(c)
|
|
return sorted(out)
|