diff --git a/spoolman/api/v1/externaldb.py b/spoolman/api/v1/externaldb.py index 74feebf18..ff3b5e084 100644 --- a/spoolman/api/v1/externaldb.py +++ b/spoolman/api/v1/externaldb.py @@ -1,11 +1,13 @@ """External database API.""" +import json import logging from typing import Annotated from fastapi import APIRouter, Query, Response -from fastapi.responses import FileResponse +from fastapi.responses import FileResponse, JSONResponse +from spoolman.env import is_tigertag_enabled from spoolman.externaldb import ( ExternalFilament, ExternalMaterial, @@ -29,9 +31,23 @@ response_model_exclude_none=True, response_model=list[ExternalFilament], ) -async def filaments() -> FileResponse: - """Get all external filaments.""" - return FileResponse(path=get_filaments_file(), media_type="application/json") +async def filaments() -> FileResponse | JSONResponse: + """Get all external filaments. + + When the optional TigerTag add-on is enabled, its synced catalog is merged in + alongside the default SpoolmanDB catalog. Import stays lazy so a deployment that + never enables the add-on never loads its module. + """ + if not is_tigertag_enabled(): + return FileResponse(path=get_filaments_file(), media_type="application/json") + + from spoolman.tigertagdb import get_tigertag_filaments_file # noqa: PLC0415 + + combined = json.loads(get_filaments_file().read_bytes()) + tigertag_path = get_tigertag_filaments_file() + if tigertag_path.exists(): + combined.extend(json.loads(tigertag_path.read_bytes())) + return JSONResponse(content=combined) @router.get( diff --git a/spoolman/api/v1/tag.py b/spoolman/api/v1/tag.py index e51ccf4fb..722497ffd 100644 --- a/spoolman/api/v1/tag.py +++ b/spoolman/api/v1/tag.py @@ -173,7 +173,7 @@ async def scan( db_spool = target if isinstance(target, models.Spool) else None db_filament = target if isinstance(target, models.Filament) else None - decoded = _decode_payload(body.format, body.payload_b64, uid) + decoded = await _decode_payload(body.format, body.payload_b64, uid) created = False # A tag can identify a filament as well as a spool; either counts as already linked. Creating @@ -232,12 +232,17 @@ async def scan( return JSONResponse(content=content) -def _decode_payload(tag_format: str | None, payload_b64: str | None, uid: str) -> tag_decode.DecodedTag | None: +async def _decode_payload(tag_format: str | None, payload_b64: str | None, uid: str) -> tag_decode.DecodedTag | None: """Decode a scan's payload, if both a format and a payload were given. Soft-fails on everything: a missing field, invalid base64, an unknown format or an unparseable payload for a known one all return None. Decoding is enrichment of a scan, never a reason to fail one -- see the class docstring on TagDecodedInfo. + + Async (rather than the plain `tag_decode.decode()` most callers want) so a TigerTag + scan can await a live per-tag catalog lookup when the external-DB add-on is enabled -- + see `tag_decode.decode_async`. Every other format, and TigerTag with the add-on off, + take the same path `decode()` would; the `await` costs nothing extra there. """ if tag_format is None or payload_b64 is None: return None @@ -253,7 +258,7 @@ def _decode_payload(tag_format: str | None, payload_b64: str | None, uid: str) - uid_bytes: bytes | None = bytes.fromhex(uid) except ValueError: uid_bytes = None - return tag_decode.decode(tag_format, raw, uid_bytes=uid_bytes) + return await tag_decode.decode_async(tag_format, raw, uid_bytes=uid_bytes, uid=uid) def _decoded_response(decoded: tag_decode.DecodedTag | None) -> TagDecodedInfo | None: diff --git a/spoolman/env.py b/spoolman/env.py index 2220d1237..0de0c92ef 100644 --- a/spoolman/env.py +++ b/spoolman/env.py @@ -573,3 +573,46 @@ def is_legacy_client_enabled() -> bool: raise ValueError( f"Failed to parse SPOOLMAN_LEGACY_CLIENT variable: Unknown value '{legacy_client}'.", ) + + +def is_tigertag_enabled() -> bool: + """Get whether the TigerTag external database add-on is enabled. + + Off by default: this talks to a third-party service (api.tigertag.io) on a + schedule, which nothing in core should do without an explicit opt-in. + + Returns: + bool: Whether the TigerTag database add-on is enabled. + + """ + enabled = os.getenv("SPOOLMAN_TIGERTAG_ENABLED", "FALSE").upper() + if enabled in {"FALSE", "0"}: + return False + if enabled in {"TRUE", "1"}: + return True + raise ValueError( + f"Failed to parse SPOOLMAN_TIGERTAG_ENABLED variable: Unknown value '{enabled}'.", + ) + + +def get_tigertag_api_url() -> str: + """Get the TigerTag API base URL from environment variables. + + Returns: + str: The TigerTag API base URL. + + """ + return os.getenv("SPOOLMAN_TIGERTAG_API_URL", "https://api.tigertag.io/api:tigertag/") + + +def get_tigertag_sync_interval() -> int: + """Get the TigerTag catalog sync interval, in seconds, from environment variables. + + A value of 0 disables periodic re-sync; the catalog is still fetched once at + startup. + + Returns: + int: The sync interval in seconds. + + """ + return int(os.getenv("SPOOLMAN_TIGERTAG_SYNC_INTERVAL", "3600")) diff --git a/spoolman/externaldb.py b/spoolman/externaldb.py index 61579862e..f83ccd86a 100644 --- a/spoolman/externaldb.py +++ b/spoolman/externaldb.py @@ -89,6 +89,13 @@ class ExternalFilament(BaseModel): ) pattern: Pattern | None = Field(default=None, description="Pattern of the filament.", examples=[Pattern.MARBLE]) translucent: bool = Field(default=False, description="Whether the filament is translucent.") + source: str | None = Field( + default=None, + description="Which catalog this entry came from, if not the default SpoolmanDB. " + "Unset for SpoolmanDB entries; set by optional add-ons that merge in their own catalog, " + "e.g. 'tigertag'.", + examples=["tigertag"], + ) glow: bool = Field(default=False, description="Whether the filament is glow-in-the-dark.") diff --git a/spoolman/main.py b/spoolman/main.py index 64182c517..5fffde95a 100644 --- a/spoolman/main.py +++ b/spoolman/main.py @@ -239,6 +239,13 @@ async def startup() -> None: externaldb.schedule_tasks(schedule) update_check.schedule_tasks(schedule) + # Optional TigerTag external database add-on: only imported when enabled, so a + # deployment that never sets SPOOLMAN_TIGERTAG_ENABLED never loads it. + if env.is_tigertag_enabled(): + from spoolman import tigertagdb # noqa: PLC0415 + + tigertagdb.schedule_tasks(schedule) + logger.info("Startup complete.") if env.is_docker() and not env.is_data_dir_mounted(): diff --git a/spoolman/tags/decode.py b/spoolman/tags/decode.py index 579b14818..bc7b6cc34 100644 --- a/spoolman/tags/decode.py +++ b/spoolman/tags/decode.py @@ -9,12 +9,24 @@ Decoding is best-effort and pure: no database, no FastAPI. An unrecognized format or a payload that fails to parse returns None rather than raising -- a scan's core contract (resolve a UID) must never fail because enrichment of it didn't work. + +One exception to "pure": TigerTag's on-tag `id_material`/`id_brand` are catalog IDs, not +names, and resolving them needs the optional external-DB add-on (spoolman/tigertagdb.py, off +by default -- see env.is_tigertag_enabled). `decode()` stays fully pure and reads only the +add-on's background-synced cache when it's enabled; `decode_async()` additionally awaits a +live per-tag API call first, for an exact catalog match, before falling back to that same +cache. Either way the add-on module itself is only ever imported when enabled, so a +deployment that never opts in never loads it. """ import io import logging import math from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from spoolman.tags.tigertag import TigerTagData from spoolman.tags import normalize_format @@ -284,11 +296,50 @@ def decode(tag_format: str | None, payload: bytes, uid_bytes: bytes | None = Non try: if fmt == "openprinttag": return _decode_openprinttag(payload, uid_bytes) + if fmt == "tigertag": + return _decode_tigertag(payload) except Exception: # noqa: BLE001 -- best effort by contract, see the module docstring logger.warning("Could not decode %s payload", fmt, exc_info=True) return None +async def decode_async( + tag_format: str | None, + payload: bytes, + uid_bytes: bytes | None = None, + uid: str | None = None, +) -> DecodedTag | None: + """Decode a tag's raw payload, resolving TigerTag names via a live API call if enabled. + + Identical to `decode()` for every format, and for TigerTag with the add-on off -- the + only difference is TigerTag with it on, where this awaits + `tigertagdb.lookup_product_by_tag(uid, id_product)` for an exact catalog match before + falling back to the same background-synced cache `decode()` uses on its own. Prefer this + from request-handling code, which is already async and can afford the await; `decode()` + stays synchronous for pure/offline callers like the test suite. + + Args: + tag_format: See `decode()`. + payload: See `decode()`. + uid_bytes: See `decode()`. + uid: The tag's UID as the same normalized hex string `/tag/scan` resolves it to (see + spoolman.tags.normalize_uid). TigerTag's live lookup needs it as a string, not + bytes -- it's the key the tag was registered against on TigerTag's own API. + + Returns: + DecodedTag | None: See `decode()`. + + """ + fmt = normalize_format(tag_format) + try: + if fmt == "tigertag": + return await _decode_tigertag_async(payload, uid) + except Exception: # noqa: BLE001 -- best effort by contract, see the module docstring + logger.warning("Could not decode %s payload", fmt, exc_info=True) + return None + return decode(tag_format, payload, uid_bytes) + + def _decode_openprinttag(payload: bytes, uid_bytes: bytes | None) -> DecodedTag | None: # Imported here, not at module level: cbor2 is only needed when a payload actually arrives, # so a missing or broken cbor2 must not stop the server from starting. @@ -322,3 +373,115 @@ def _decode_openprinttag(payload: bytes, uid_bytes: bytes | None) -> DecodedTag bed_temp=bed_temp, instance_uuid=data.effective_instance_uuid, ) + + +def _decode_tigertag_raw(payload: bytes) -> "TigerTagData | None": + # Imported here for the same reason as openprinttag above: the dependency should only be + # paid for when a tigertag payload actually arrives. + from spoolman.tags import tigertag as tigertag_codec # noqa: PLC0415 + + try: + data = tigertag_codec.decode_ntag213(payload) + except ValueError: + logger.debug("Could not decode payload as TigerTag", exc_info=True) + return None + + if not tigertag_codec.is_tigertag(data.id_tigertag): + # Wrong magic number -- either not a TigerTag at all, or a blank/Init tag with + # nothing written to it yet. Either way there's nothing usable to surface. + return None + + return data + + +def _tigertag_decoded( + raw: "TigerTagData", + *, + material: str | None = None, + name: str | None = None, + vendor_name: str | None = None, + density: float | None = None, +) -> DecodedTag: + # color_hex/diameter/weight always come straight off the physical tag, never from the + # external-DB add-on -- the tag is ground truth for this specific roll's physical + # properties; the add-on only ever fills in what the tag can't carry: names for + # id_material/id_brand's catalog IDs, and a material-specific density. + # id_product is a catalog SKU, not a per-roll identifier, so it's never used as + # instance_uuid -- same instance-uniqueness caution as OpenPrintTag's UID-derived id. + # TigerTag (NTAG213 "Maker"/"+") only ever tags filament spools, so material_class is + # always "FFF" -- unlike OpenPrintTag, there is no resin variant of this format. + return DecodedTag( + material_class="FFF", + material=material, + name=name, + vendor_name=vendor_name, + color_hex=raw.color_hex, + diameter=raw.diameter_mm or None, + density=density, + # weight_grams is None both when the tag has no weight and when it reports one in a + # unit other than grams (e.g. kilograms) -- either way, not a usable weight in grams. + weight=float(raw.weight_grams) if raw.weight_grams else None, + ) + + +def _tigertag_decoded_from_cache(raw: "TigerTagData") -> DecodedTag: + """Resolve names for a raw TigerTag decode from the add-on's background-synced cache. + + Only called once the caller has already confirmed the add-on is enabled -- this is where + spoolman/tigertagdb.py actually gets imported, kept out of every other code path so a + deployment that never opts in never loads it. + """ + from spoolman import tigertagdb # noqa: PLC0415 -- see docstring + + return _tigertag_decoded( + raw, + material=tigertagdb.lookup_material_name(raw.id_material), + name=tigertagdb.lookup_material_name(raw.id_material), + vendor_name=tigertagdb.lookup_brand_name(raw.id_brand), + density=tigertagdb.lookup_material_density(raw.id_material), + ) + + +def _decode_tigertag(payload: bytes) -> DecodedTag | None: + from spoolman import env # noqa: PLC0415 + + raw = _decode_tigertag_raw(payload) + if raw is None: + return None + if not env.is_tigertag_enabled(): + return _tigertag_decoded(raw) + return _tigertag_decoded_from_cache(raw) + + +async def _decode_tigertag_async(payload: bytes, uid: str | None) -> DecodedTag | None: + from spoolman import env # noqa: PLC0415 + + raw = _decode_tigertag_raw(payload) + if raw is None: + return None + if not env.is_tigertag_enabled(): + return _tigertag_decoded(raw) + + from spoolman import tigertagdb # noqa: PLC0415 -- see _tigertag_decoded_from_cache + + if raw.id_product and uid is not None: + try: + ext = await tigertagdb.lookup_product_by_tag(uid, raw.id_product) + except Exception: # noqa: BLE001 -- enrichment must never fail the scan it's decoding + logger.debug("TigerTag live product lookup failed for uid=%s", uid, exc_info=True) + ext = None + if ext is not None: + return _tigertag_decoded( + raw, + material=ext.material if ext.material != "Unknown" else None, + name=ext.name, + vendor_name=ext.manufacturer if ext.manufacturer != "Unknown" else None, + # ext.density is a fixed 1.24 placeholder for every TigerTag product (see + # tigertagdb._to_external_filament) -- the materials cache's own per-material + # figure is the real one, so prefer it here too. + density=tigertagdb.lookup_material_density(raw.id_material), + ) + + # No product id on the tag, or the live lookup was unavailable/failed/found nothing -- + # fall back to the coarser but still real background-synced cache. + return _tigertag_decoded_from_cache(raw) diff --git a/spoolman/tags/tigertag.py b/spoolman/tags/tigertag.py new file mode 100644 index 000000000..d1f2f0284 --- /dev/null +++ b/spoolman/tags/tigertag.py @@ -0,0 +1,231 @@ +"""TigerTag NTAG213 binary decoder. + +Decodes the TigerTag Maker/+ format from NTAG213 NFC chips (144 bytes user memory, +pages 4-39). Byte offsets below are verified against the normative table in +TigerTag-Project/TigerTag-RFID-Guide's README and the offsets its own +`Sample code/parse_tigertag.py` reference parser reads, which agree exactly. Only the +fields something downstream actually reads are decoded; the rest of the spec +(Color 2/3, tag index/count, HueForge TD, Measure Available, the ECDSA signature) is +left for a later PR, same scope discipline as the OpenPrintTag codec. +""" + +import logging +import struct +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +# NTAG213 has 144 bytes of user memory (pages 4-39, 36 pages x 4 bytes) +NTAG213_USER_BYTES = 144 + +# TigerTag magic numbers (from id_version.json API database — the README hex values are incorrect) +TIGERTAG_MAKER_V1 = 0x5BF59264 # TigerTag Maker v1.0 — offline only (decimal 1542820452) +TIGERTAG_PRO_V1 = 0xBC0FCB97 # TigerTag+ v1.0 — offline + cloud sync (decimal 3155151767) +TIGERTAG_INIT = 0x6C41A2E1 # TigerTag Init — blank/uninitialized (decimal 1816240865) + +# All valid TigerTag magic numbers (for detection) +TIGERTAG_MAGIC_NUMBERS = {TIGERTAG_MAKER_V1, TIGERTAG_PRO_V1} + +# Hex color string lengths: RGB (no alpha) vs RGBA. +_RGB_HEX_LEN = 6 +_RGBA_HEX_LEN = 8 + +# TigerTag diameter IDs, from the spec repo's database/id_diameter.json (and confirmed by +# every worked example in its README): 56 (0x38) = 1.75mm, 221 (0xDD) = 2.85mm. No other +# values are defined -- a tag using anything else has an id this codec doesn't recognize. +_DIAMETER_ID_175 = 56 +_DIAMETER_ID_285 = 221 + +# Measure unit IDs, from the spec repo's database/id_measure_unit.json. The "Measure" field +# (see decode_ntag213) is only grams when this is 21 -- the README's own Polymaker example +# uses 35 (kilograms) for the same field, so the raw value can't be trusted as grams on its +# own. Other defined units (mg, ml, cl, L, m^3, mm, cm, m, m^2) aren't grams either. +_UNIT_ID_GRAMS = 21 + + +def is_tigertag(magic: int) -> bool: + """Check if a magic number identifies a valid TigerTag (Maker or Pro/+).""" + return magic in TIGERTAG_MAGIC_NUMBERS + + +@dataclass +class TigerTagData: + """Represents all data fields stored on a TigerTag NFC chip.""" + + # Core identifiers + id_tigertag: int = 0 # 4 bytes - magic number / format identifier + id_product: int = 0 # 4 bytes - product/filament ID (0xFFFFFFFF = Factory) + id_material: int = 0 # 2 bytes - material type ID + id_diameter: int = 0 # 1 byte - diameter ID (TigerTag numbering) + id_aspect: int = 0 # 1 byte - aspect/finish ID + id_type: int = 0 # 1 byte - type ID (142=filament) + id_brand: int = 0 # 2 bytes - brand/manufacturer ID + + # Color as RGBA + color_r: int = 0 # 1 byte + color_g: int = 0 # 1 byte + color_b: int = 0 # 1 byte + color_a: int = 255 # 1 byte + + # Spool properties. `weight` is the raw "Measure" value -- only trustable as grams when + # `id_unit` says so, see `weight_grams`. Defaults to grams, the overwhelmingly common + # real-world case (and the spec repo's own first worked example), so constructing a + # TigerTagData without specifying a unit behaves the way every caller already expects. + weight: int = 0 + id_unit: int = _UNIT_ID_GRAMS # 1 byte - measure unit ID (database/id_measure_unit.json) + + # Temperature settings + nozzle_temp: int = 0 # nozzle temp min in °C + nozzle_temp_max: int = 0 # nozzle temp max in °C + bed_temp: int = 0 # bed temp min in °C + bed_temp_max: int = 0 # bed temp max in °C + drying_temp: int = 0 # drying temp in °C + drying_duration: int = 0 # drying time in hours + + # Metadata + timestamp: int = 0 # 4 bytes - seconds since 2000-01-01 GMT + + # User message - 28 bytes UTF-8 + user_message: str = "" + + @property + def color_hex(self) -> str: + """Get color as hex string (without alpha).""" + return f"{self.color_r:02x}{self.color_g:02x}{self.color_b:02x}" + + @color_hex.setter + def color_hex(self, value: str) -> None: + """Set color from hex string.""" + value = value.lstrip("#") + if len(value) == _RGB_HEX_LEN: + self.color_r = int(value[0:2], 16) + self.color_g = int(value[2:4], 16) + self.color_b = int(value[4:6], 16) + elif len(value) == _RGBA_HEX_LEN: + self.color_r = int(value[0:2], 16) + self.color_g = int(value[2:4], 16) + self.color_b = int(value[4:6], 16) + self.color_a = int(value[6:8], 16) + + @property + def diameter_mm(self) -> float: + """Get diameter in mm from the diameter ID.""" + if self.id_diameter == _DIAMETER_ID_175: + return 1.75 + if self.id_diameter == _DIAMETER_ID_285: + return 2.85 + return 0.0 + + @property + def weight_grams(self) -> int | None: + """The measured weight in grams, or None if the tag reports it in some other unit. + + `weight` is the raw "Measure" field -- the spec also carries a measure *unit* ID + alongside it (database/id_measure_unit.json: 21=g, but also mg/kg/ml/cl/L/m^3/mm/ + cm/m/m^2), and a real tag can use any of them -- the README's own worked example + for a Polymaker TigerTag+ tag reports its weight in kilograms, not grams. Rather + than guess a conversion, a non-gram tag is treated as not carrying a usable weight. + """ + return self.weight if self.id_unit == _UNIT_ID_GRAMS else None + + +# TigerTag binary format (big-endian, 36-byte header), per the spec repo's byte table: +# +# Offset Size Format Field +# 0 4 >I id_tigertag +# 4 4 >I id_product +# 8 2 >H id_material +# 10 1 B id_aspect_1 +# 11 1 B id_aspect_2 +# 12 1 B id_type +# 13 1 B id_diameter +# 14 2 >H id_brand +# 16 4 >I color_rgba (R<<24 | G<<16 | B<<8 | A) +# 20 4 >I measure_unit (measure<<8 | id_unit) +# 24 2 >H nozzle_temp_min +# 26 2 >H nozzle_temp_max +# 28 1 B drying_temp +# 29 1 B drying_time (hours) +# 30 1 B bed_temp_min +# 31 1 B bed_temp_max +# 32 4 >I timestamp +# +# After the header (optional; not written by every tool, so length-guarded on read): +# 36-47 12 - Color 2/3, tag index, HueForge TD, reserved -- not decoded here +# 48 28 UTF-8 user_message +# 76-143 68 - Measure Available, reserved, ECDSA signature -- not decoded here + +_HEADER_FMT = ">II HBB BBH I I HH BBBB I" +_HEADER_SIZE = struct.calcsize(_HEADER_FMT) # 36 bytes +_USER_MESSAGE_SIZE = 28 +_USER_MESSAGE_OFFSET = 48 + + +def decode_ntag213(raw_bytes: bytes) -> TigerTagData: + """Decode raw NTAG213 user memory bytes into TigerTagData. + + Args: + raw_bytes: The raw bytes from NTAG213 pages 4-39 (up to 144 bytes). + + Returns: + TigerTagData: The decoded tag data. + + Raises: + ValueError: If the data is too short to decode. + + """ + if len(raw_bytes) < _HEADER_SIZE: + raise ValueError(f"Data too short: expected at least {_HEADER_SIZE} bytes, got {len(raw_bytes)}") + + # Debug: dump raw bytes for diagnosing format issues + hex_dump = " ".join(f"{b:02x}" for b in raw_bytes[:68]) + logger.info("TigerTag raw bytes (first 68): %s", hex_dump) + + values = struct.unpack_from(_HEADER_FMT, raw_bytes, 0) + + # Unpack color from RGBA uint32: value = R<<24 | G<<16 | B<<8 | A + color_val = values[8] + color_r = (color_val >> 24) & 0xFF + color_g = (color_val >> 16) & 0xFF + color_b = (color_val >> 8) & 0xFF + color_a = color_val & 0xFF + + # Unpack measure + unit: measure in upper 24 bits, id_unit in lower 8. Whether `measure` + # means grams depends on id_unit -- see TigerTagData.weight_grams. + measure_unit = values[9] + weight = (measure_unit >> 8) & 0xFFFFFF + id_unit = measure_unit & 0xFF + + data = TigerTagData( + id_tigertag=values[0], + id_product=values[1], + id_material=values[2], + id_aspect=values[3], # aspect_1 + # index 4 is aspect_2, ignored + id_type=values[5], + id_diameter=values[6], + id_brand=values[7], + color_r=color_r, + color_g=color_g, + color_b=color_b, + color_a=color_a, + weight=weight, + id_unit=id_unit, + nozzle_temp=values[10], + nozzle_temp_max=values[11], + drying_temp=values[12], + drying_duration=values[13], + bed_temp=values[14], + bed_temp_max=values[15], + timestamp=values[16], + ) + + # Read user message at offset 48 (28 bytes, optional region) + if len(raw_bytes) >= _USER_MESSAGE_OFFSET + _USER_MESSAGE_SIZE: + msg_bytes = raw_bytes[_USER_MESSAGE_OFFSET : _USER_MESSAGE_OFFSET + _USER_MESSAGE_SIZE] + null_idx = msg_bytes.find(b"\x00") + if null_idx >= 0: + msg_bytes = msg_bytes[:null_idx] + data.user_message = msg_bytes.decode("utf-8", errors="replace") + + return data diff --git a/spoolman/tigertagdb.py b/spoolman/tigertagdb.py new file mode 100644 index 000000000..b32bcf559 --- /dev/null +++ b/spoolman/tigertagdb.py @@ -0,0 +1,346 @@ +"""Functions for syncing data from the TigerTag external filament database.""" + +import datetime +import json +import logging +from pathlib import Path +from urllib.parse import urljoin + +import httpx +from pydantic import BaseModel +from scheduler.asyncio.scheduler import Scheduler + +from spoolman import filecache +from spoolman.env import get_tigertag_api_url, get_tigertag_sync_interval, is_tigertag_enabled +from spoolman.externaldb import ExternalFilament + +logger = logging.getLogger(__name__) + +TIGERTAG_CACHE_FILE = "tigertag_filaments.json" +TIGERTAG_BRANDS_CACHE_FILE = "tigertag_brands.json" + +# TigerTag returns an 8-char RGBA hex string; Spoolman only stores RGB. +_RGBA_HEX_LEN = 8 +TIGERTAG_MATERIALS_CACHE_FILE = "tigertag_materials.json" + + +class TigerTagBrand(BaseModel): + """A brand/manufacturer from the TigerTag API.""" + + id_brand: int + name: str + + +class TigerTagMaterial(BaseModel): + """A material/type from the TigerTag API.""" + + id_type: int + name: str + density: float | None = None + + +class _TigerTagProductFilament(BaseModel): + """The nested detail block the single-product endpoint (`product/get`) returns. + + The bulk endpoint (`product/get/all`, used for the synced cache) puts material and + weight at the top level instead -- `TigerTagProduct.material`/`.measure` cover that + shape. Confirmed against the live API: the two endpoints disagree on this. + """ + + material: str | None = None + grams: float | None = None + + +class TigerTagProduct(BaseModel): + """A filament product from the TigerTag API. + + `product/get/all` (bulk) and `product/get` (single, live tag lookup) disagree on + where material and weight live -- see `_TigerTagProductFilament`. `resolved_material`/ + `resolved_weight_g` read whichever shape actually populated the response. + """ + + id: int + product_type: str | None = None + brand: str | None = None + title: str | None = None + material: str | None = None + color: str | None = None + color_info: dict | None = None + measure: str | None = None + sku: str | None = None + filament: _TigerTagProductFilament | None = None + + @property + def resolved_material(self) -> str | None: + """The material name, from whichever endpoint shape this response came from.""" + if self.material is not None: + return self.material + return self.filament.material if self.filament is not None else None + + @property + def resolved_weight_g(self) -> float | None: + """The weight in grams, if the single-product shape's `filament.grams` has it. + + None when it doesn't (including the bulk shape, which has no such field) -- + callers fall back to parsing the top-level `measure` string in that case. + """ + if self.filament is not None and self.filament.grams is not None: + return float(self.filament.grams) + return None + + +def _parse_weight_from_measure(measure: str | None) -> float: + """Parse weight in grams from measure string like '1 kg' or '500 g'.""" + if not measure: + return 1000.0 + measure = measure.strip().lower() + try: + if "kg" in measure: + return float(measure.replace("kg", "").strip()) * 1000 + if "g" in measure: + return float(measure.replace("g", "").strip()) + except ValueError: + pass + return 1000.0 + + +def _to_external_filament(product: TigerTagProduct) -> ExternalFilament: + """Convert a TigerTag product into an ExternalFilament.""" + manufacturer = product.brand or "Unknown" + material = product.resolved_material or "Unknown" + name = product.title or f"{manufacturer} {material}" + weight = product.resolved_weight_g + if weight is None: + weight = _parse_weight_from_measure(product.measure) + + # Clean up color hex - remove leading # and alpha channel if present + color_hex = None + if product.color: + hex_str = product.color.lstrip("#") + if len(hex_str) == _RGBA_HEX_LEN: + hex_str = hex_str[:6] + color_hex = hex_str + + return ExternalFilament( + id=f"tigertag_{product.id}", + manufacturer=manufacturer, + name=name, + material=material, + density=1.24, + weight=weight, + diameter=1.75, + color_hex=color_hex, + source="tigertag", + ) + + +TIGERTAG_FILAMENT_TYPE_ID = 142 +TIGERTAG_PAGE_SIZE = 50 + + +async def _fetch_all_products(base_url: str) -> list[dict]: + """Fetch all filament products from TigerTag API using pagination.""" + products_url = urljoin(base_url, "product/get/all") + all_items: list[dict] = [] + page = 1 + + async with httpx.AsyncClient() as client: + while True: + response = await client.post( + products_url, + json={"page": page, "per_page": TIGERTAG_PAGE_SIZE, "product_type_id": TIGERTAG_FILAMENT_TYPE_ID}, + ) + response.raise_for_status() + data = response.json() + items = data.get("items", []) + all_items.extend(items) + + if data.get("nextPage") is None: + break + page = data["nextPage"] + + return all_items + + +async def _fetch_brands(base_url: str) -> list[dict]: + """Fetch all brands from TigerTag API (GET, returns list with id/name).""" + url = urljoin(base_url, "brand/get/all") + async with httpx.AsyncClient() as client: + response = await client.get(url) + response.raise_for_status() + data = response.json() + return data.get("items", data) if isinstance(data, dict) else data + + +async def _fetch_materials(base_url: str) -> list[dict]: + """Fetch all filament materials from TigerTag API (GET, returns list with id/label/density).""" + url = urljoin(base_url, "material/get/all") + async with httpx.AsyncClient() as client: + response = await client.get(url) + response.raise_for_status() + data = response.json() + return data.get("items", data) if isinstance(data, dict) else data + + +async def _sync_tigertag() -> None: + logger.info("Syncing TigerTag DB.") + + base_url = get_tigertag_api_url() + + try: + # Fetch all filament products via paginated API + products_list = await _fetch_all_products(base_url) + + # Parse products + products = [TigerTagProduct(**p) for p in products_list] + + # Convert to ExternalFilament format + filaments = [_to_external_filament(p) for p in products] + + # Cache to local file + filaments_json = json.dumps( + [f.model_dump(exclude_none=True) for f in filaments], + ensure_ascii=False, + ) + filecache.update_file(TIGERTAG_CACHE_FILE, filaments_json.encode("utf-8")) + + logger.info("TigerTag DB synced. Filaments: %d", len(filaments)) + + except Exception: + logger.exception("Failed to sync TigerTag DB") + + # Fetch and cache brands (id_brand -> name mapping) + try: + brands_list = await _fetch_brands(base_url) + filecache.update_file(TIGERTAG_BRANDS_CACHE_FILE, json.dumps(brands_list, ensure_ascii=False).encode("utf-8")) + logger.info("TigerTag brands synced: %d", len(brands_list)) + except Exception: + logger.exception("Failed to sync TigerTag brands") + + # Fetch and cache materials (id_type -> name mapping) + try: + materials_list = await _fetch_materials(base_url) + filecache.update_file( + TIGERTAG_MATERIALS_CACHE_FILE, + json.dumps(materials_list, ensure_ascii=False).encode("utf-8"), + ) + logger.info("TigerTag materials synced: %d", len(materials_list)) + except Exception: + logger.exception("Failed to sync TigerTag materials") + + +def get_tigertag_filaments_file() -> Path: + """Get the path to the cached TigerTag filaments file.""" + return filecache.get_file(TIGERTAG_CACHE_FILE) + + +# A cache miss/corrupt cache file must fall through to the tag's raw data rather +# than raise, so these three lookups deliberately swallow anything that can go +# wrong reading and parsing the file (missing file, truncated JSON, wrong shape). +_CACHE_LOOKUP_ERRORS = (OSError, ValueError, KeyError, TypeError) + + +def lookup_brand_name(id_brand: int) -> str | None: + """Look up a brand name by its TigerTag numeric ID from the cached brand list. + + Brand API returns: [{"id": 19961, "name": "Rosa3D", "type_ids": [142]}, ...] + """ + try: + data = json.loads(filecache.get_file_contents(TIGERTAG_BRANDS_CACHE_FILE)) + for entry in data: + if entry.get("id") == id_brand: + return entry.get("name") + except _CACHE_LOOKUP_ERRORS: + logger.debug("Could not look up TigerTag brand %d from cache", id_brand) + return None + + +def lookup_material_name(id_material: int) -> str | None: + """Look up a material name by its TigerTag numeric ID from the cached material list. + + Material API returns: [{"id": 38219, "label": "PLA", "density": 1.24, ...}, ...] + """ + try: + data = json.loads(filecache.get_file_contents(TIGERTAG_MATERIALS_CACHE_FILE)) + for entry in data: + if entry.get("id") == id_material: + return entry.get("label") + except _CACHE_LOOKUP_ERRORS: + logger.debug("Could not look up TigerTag material %d from cache", id_material) + return None + + +def lookup_material_density(id_material: int) -> float | None: + """Look up material density by its TigerTag numeric ID from the cached material list.""" + try: + data = json.loads(filecache.get_file_contents(TIGERTAG_MATERIALS_CACHE_FILE)) + for entry in data: + if entry.get("id") == id_material: + density = entry.get("density") + if density is not None: + return float(density) + except _CACHE_LOOKUP_ERRORS: + logger.debug("Could not look up TigerTag material density %d from cache", id_material) + return None + + +async def lookup_product_by_tag(nfc_tag_uid: str, id_product: int) -> ExternalFilament | None: + """Look up a TigerTag product via the real-time API using tag UID + product_id. + + The tag stores a small product_id (e.g. 28) which is different from the API's + internal database ID (billions). The single-product endpoint requires both the + tag's hardware UID and the product_id to resolve to a full product record. + + Returns an ExternalFilament if found, None otherwise. + """ + if not is_tigertag_enabled(): + return None + + base_url = get_tigertag_api_url() + url = urljoin(base_url, "product/get") + + try: + async with httpx.AsyncClient() as client: + response = await client.get(url, params={"uid": nfc_tag_uid, "product_id": id_product}, timeout=5.0) + response.raise_for_status() + data = response.json() + + product = TigerTagProduct(**data) + except httpx.HTTPStatusError as e: + if e.response.status_code == httpx.codes.NOT_FOUND: + logger.debug("TigerTag product_id=%d not found via API (uid=%s)", id_product, nfc_tag_uid) + else: + logger.warning("TigerTag API error for product_id=%d: %s", id_product, e) + return None + except (httpx.HTTPError, ValueError) as e: + # ValueError covers a non-JSON body or a JSON body that doesn't match + # TigerTagProduct's schema (pydantic raises ValidationError, a ValueError subclass). + logger.debug("TigerTag API lookup failed for product_id=%d: %s", id_product, e) + return None + else: + ext = _to_external_filament(product) + logger.info("TigerTag API resolved product_id=%d → %s (%s)", id_product, ext.name, ext.manufacturer) + return ext + + +def schedule_tasks(scheduler: Scheduler) -> None: + """Schedule TigerTag sync tasks. + + Args: + scheduler: The scheduler to use for scheduling tasks. + + """ + if not is_tigertag_enabled(): + logger.info("TigerTag integration is disabled. Skipping sync.") + return + + logger.info("Scheduling TigerTag DB sync.") + + # Run once on startup + scheduler.once(datetime.timedelta(seconds=0), _sync_tigertag) # type: ignore[arg-type] + + sync_interval = get_tigertag_sync_interval() + if sync_interval > 0: + scheduler.cyclic(datetime.timedelta(seconds=sync_interval), _sync_tigertag) # type: ignore[arg-type] + else: + logger.info("TigerTag sync interval is 0, skipping periodic sync.") diff --git a/tests/test_tag_decode.py b/tests/test_tag_decode.py index 88c02b0a3..866f32459 100644 --- a/tests/test_tag_decode.py +++ b/tests/test_tag_decode.py @@ -1,4 +1,4 @@ -"""Tests for the format-dispatching decode wiring in spoolman/tag_decode.py. +"""Tests for the format-dispatching decode wiring in spoolman/tags/decode.py. Builds the same synthetic OpenPrintTag NFC-V memory dump as test_openprinttag_codec.py, but only enough of it to exercise dispatch and field mapping -- the codec's own parsing edge @@ -8,14 +8,18 @@ import uuid import cbor2 +import httpx import pytest +from spoolman import env, tigertagdb +from spoolman.externaldb import ExternalFilament from spoolman.tags import decode as tag_decode from spoolman.tags import openprinttag as openprinttag_codec from spoolman.tags.decode import ( DecodedTag, approximate_density, decode, + decode_async, density_or_fallback, ) from spoolman.tags.openprinttag import ( @@ -27,6 +31,8 @@ MF_PRIMARY_COLOR, uuid5_bytes, ) +from spoolman.tags.tigertag import TIGERTAG_INIT, TIGERTAG_MAKER_V1, TigerTagData +from tests.test_tigertag_codec import encode_ntag213 def _cbor_payload(main: dict, aux: dict | None = None) -> bytes: @@ -124,6 +130,244 @@ def test_decode_derives_external_id_from_uid_when_tag_has_no_instance_uuid() -> assert with_uid.instance_uuid == str(uuid5_bytes(uuid.UUID("31062f81-b5bd-4f86-a5f8-46367e841508"), uid_bytes)) +def test_decode_dispatches_tigertag() -> None: + raw = encode_ntag213( + TigerTagData( + id_tigertag=TIGERTAG_MAKER_V1, + id_diameter=56, # 0x38, the spec's real 1.75mm id + weight=1000, + color_r=0x11, + color_g=0x22, + color_b=0x33, + ), + ) + result = decode("tigertag", raw) + + assert result == DecodedTag( + material_class="FFF", + material=None, + name=None, + vendor_name=None, + color_hex="112233", + diameter=1.75, + density=None, + weight=1000.0, + spool_weight=None, + used_weight=None, + instance_uuid=None, + ) + + +def test_decode_tigertag_rejects_blank_init_tag() -> None: + """TIGERTAG_INIT marks an uninitialized tag -- nothing usable to surface.""" + raw = encode_ntag213(TigerTagData(id_tigertag=TIGERTAG_INIT)) + assert decode("tigertag", raw) is None + + +def test_decode_tigertag_rejects_junk_magic() -> None: + raw = encode_ntag213(TigerTagData(id_tigertag=0xDEADBEEF)) + assert decode("tigertag", raw) is None + + +def test_decode_tigertag_unparseable_payload_returns_none() -> None: + assert decode("tigertag", b"too short") is None + + +def test_decode_tigertag_unknown_diameter_is_none_not_zero() -> None: + raw = encode_ntag213(TigerTagData(id_tigertag=TIGERTAG_MAKER_V1, id_diameter=99)) + result = decode("tigertag", raw) + assert result is not None + assert result.diameter is None + + +def test_decode_tigertag_zero_weight_is_none() -> None: + raw = encode_ntag213(TigerTagData(id_tigertag=TIGERTAG_MAKER_V1, weight=0)) + result = decode("tigertag", raw) + assert result is not None + assert result.weight is None + + +def test_decode_tigertag_non_gram_unit_is_none_not_a_scaled_value() -> None: + """A tag reporting its weight in kilograms must not be read as that many grams.""" + raw = encode_ntag213(TigerTagData(id_tigertag=TIGERTAG_MAKER_V1, weight=1000, id_unit=35)) # 35 = kilograms + result = decode("tigertag", raw) + assert result is not None + assert result.weight is None + + +def _tigertag_raw(**kwargs: object) -> bytes: + kwargs.setdefault("id_tigertag", TIGERTAG_MAKER_V1) + kwargs.setdefault("color_r", 0x11) + kwargs.setdefault("color_g", 0x22) + kwargs.setdefault("color_b", 0x33) + return encode_ntag213(TigerTagData(**kwargs)) # type: ignore[arg-type] + + +# --- decode() / decode_async() with the external-DB add-on enabled ----------------- +# +# tag_decode never imports spoolman.tigertagdb unless env.is_tigertag_enabled() says so +# (see _decode_tigertag_raw et al.'s docstrings) -- these tests monkeypatch that flag plus +# tigertagdb's lookup functions directly, rather than exercising the real cache/API, since +# tigertagdb's own behavior is covered by tests/test_tigertagdb.py. + + +def test_decode_tigertag_addon_disabled_leaves_names_unresolved(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: False) + monkeypatch.setattr(tigertagdb, "lookup_material_name", lambda _id: pytest.fail("should not be called")) + + result = decode("tigertag", _tigertag_raw(id_material=38219)) + assert result is not None + assert result.material is None + assert result.vendor_name is None + + +def test_decode_tigertag_addon_enabled_resolves_names_from_cache(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: True) + monkeypatch.setattr( + tigertagdb, + "lookup_material_name", + lambda id_material: "PLA" if id_material == 38219 else None, + ) + monkeypatch.setattr(tigertagdb, "lookup_brand_name", lambda id_brand: "Rosa3D" if id_brand == 19961 else None) + monkeypatch.setattr( + tigertagdb, + "lookup_material_density", + lambda id_material: 1.24 if id_material == 38219 else None, + ) + + result = decode("tigertag", _tigertag_raw(id_material=38219, id_brand=19961)) + assert result is not None + assert result.material == "PLA" + assert result.name == "PLA" + assert result.vendor_name == "Rosa3D" + assert result.density == 1.24 + # Physical properties still come straight off the tag, never the catalog. + assert result.color_hex == "112233" + + +@pytest.mark.asyncio +async def test_decode_async_addon_disabled_matches_sync_decode(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: False) + raw = _tigertag_raw(id_material=38219) + + assert await decode_async("tigertag", raw, uid="04A2B3C4") == decode("tigertag", raw) + + +@pytest.mark.asyncio +async def test_decode_async_prefers_live_product_lookup(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: True) + monkeypatch.setattr(tigertagdb, "lookup_material_density", lambda _id: 1.24) + + async def _fake_lookup(uid: str, id_product: int) -> ExternalFilament: + assert uid == "04A2B3C4" + assert id_product == 28 + return ExternalFilament( + id="tigertag_28", + manufacturer="Rosa3D", + name="Rosa3D PLA Starter", + material="PLA", + density=1.24, + weight=1000, + diameter=1.75, + source="tigertag", + ) + + monkeypatch.setattr(tigertagdb, "lookup_product_by_tag", _fake_lookup) + monkeypatch.setattr( + tigertagdb, + "lookup_brand_name", + lambda _id: pytest.fail("cached fallback should not run when the live lookup succeeds"), + ) + + result = await decode_async("tigertag", _tigertag_raw(id_product=28), uid="04A2B3C4") + assert result is not None + assert result.material == "PLA" + assert result.name == "Rosa3D PLA Starter" + assert result.vendor_name == "Rosa3D" + assert result.density == 1.24 + # Physical properties still come straight off the tag, never the live API response. + assert result.color_hex == "112233" + + +@pytest.mark.asyncio +async def test_decode_async_falls_back_to_cache_when_live_lookup_finds_nothing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: True) + + async def _fake_lookup(_uid: str, _id_product: int) -> None: + return None + + monkeypatch.setattr(tigertagdb, "lookup_product_by_tag", _fake_lookup) + monkeypatch.setattr(tigertagdb, "lookup_material_name", lambda _id: "PLA") + monkeypatch.setattr(tigertagdb, "lookup_brand_name", lambda _id: "Rosa3D") + monkeypatch.setattr(tigertagdb, "lookup_material_density", lambda _id: 1.24) + + result = await decode_async("tigertag", _tigertag_raw(id_product=28), uid="04A2B3C4") + assert result is not None + assert result.material == "PLA" + assert result.vendor_name == "Rosa3D" + + +@pytest.mark.asyncio +async def test_decode_async_falls_back_to_cache_when_live_lookup_raises(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: True) + + async def _fake_lookup(_uid: str, _id_product: int) -> None: + raise httpx.ConnectError("no route to host") + + monkeypatch.setattr(tigertagdb, "lookup_product_by_tag", _fake_lookup) + monkeypatch.setattr(tigertagdb, "lookup_material_name", lambda _id: "PLA") + monkeypatch.setattr(tigertagdb, "lookup_brand_name", lambda _id: None) + monkeypatch.setattr(tigertagdb, "lookup_material_density", lambda _id: 1.24) + + result = await decode_async("tigertag", _tigertag_raw(id_product=28), uid="04A2B3C4") + assert result is not None + assert result.material == "PLA" + + +@pytest.mark.asyncio +async def test_decode_async_skips_live_lookup_without_a_product_id(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: True) + monkeypatch.setattr( + tigertagdb, + "lookup_product_by_tag", + lambda *_a, **_k: pytest.fail("should not be called without a product id"), + ) + monkeypatch.setattr(tigertagdb, "lookup_material_name", lambda _id: "PLA") + monkeypatch.setattr(tigertagdb, "lookup_brand_name", lambda _id: None) + monkeypatch.setattr(tigertagdb, "lookup_material_density", lambda _id: None) + + result = await decode_async("tigertag", _tigertag_raw(id_product=0), uid="04A2B3C4") + assert result is not None + assert result.material == "PLA" + + +@pytest.mark.asyncio +async def test_decode_async_skips_live_lookup_without_a_uid(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: True) + monkeypatch.setattr( + tigertagdb, + "lookup_product_by_tag", + lambda *_a, **_k: pytest.fail("should not be called without a uid"), + ) + monkeypatch.setattr(tigertagdb, "lookup_material_name", lambda _id: "PLA") + monkeypatch.setattr(tigertagdb, "lookup_brand_name", lambda _id: None) + monkeypatch.setattr(tigertagdb, "lookup_material_density", lambda _id: None) + + result = await decode_async("tigertag", _tigertag_raw(id_product=28), uid=None) + assert result is not None + assert result.material == "PLA" + + +@pytest.mark.asyncio +async def test_decode_async_non_tigertag_format_is_unaffected(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(env, "is_tigertag_enabled", lambda: True) + raw = _build_openprinttag({MF_MATERIAL_TYPE: 0}) + + assert await decode_async("openprinttag", raw) == decode("openprinttag", raw) + + # --- density fallback -------------------------------------------------------------- @@ -165,7 +409,8 @@ def boom(*_args: object, **_kwargs: object) -> None: assert tag_decode.decode("openprinttag", b"\x00", uid_bytes=b"\x04\xa2") is None -def test_odd_length_uid_still_decodes_without_an_instance_uid(monkeypatch: pytest.MonkeyPatch) -> None: +@pytest.mark.asyncio +async def test_odd_length_uid_still_decodes_without_an_instance_uid(monkeypatch: pytest.MonkeyPatch) -> None: """A UID that fromhex rejects must not turn the decode into a 500. normalize_uid stops odd lengths earlier.""" import base64 # noqa: PLC0415 @@ -174,12 +419,18 @@ def test_odd_length_uid_still_decodes_without_an_instance_uid(monkeypatch: pytes seen = {} - def fake_decode(_tag_format: str, _raw: bytes, uid_bytes: bytes | None = None) -> None: + async def fake_decode_async( + _tag_format: str, + _raw: bytes, + uid_bytes: bytes | None = None, + uid: str | None = None, + ) -> None: seen["uid_bytes"] = uid_bytes + seen["uid"] = uid - monkeypatch.setattr(tag_decode, "decode", fake_decode) + monkeypatch.setattr(tag_decode, "decode_async", fake_decode_async) payload_b64 = base64.b64encode(b"\x00").decode() - assert tag_api._decode_payload("openprinttag", payload_b64, "ABC") is None # noqa: SLF001 + assert await tag_api._decode_payload("openprinttag", payload_b64, "ABC") is None # noqa: SLF001 assert seen["uid_bytes"] is None diff --git a/tests/test_tigertag_codec.py b/tests/test_tigertag_codec.py new file mode 100644 index 000000000..2d03391fc --- /dev/null +++ b/tests/test_tigertag_codec.py @@ -0,0 +1,344 @@ +"""Tests for the TigerTag NTAG213 binary codec. + +Field offsets are checked against three worked examples from the spec repo's own README +(TigerTag-Project/TigerTag-RFID-Guide, sections 4, 4.1, 4.2): a Rosa3D red PLA Maker tag, a +Polymaker PolyTerra+ tag (whose weight is in kilograms, not grams), and a blank Init tag. +Real reference values, not just synthetic ones -- same role the Prusa spec repo's sample +dumps played for the OpenPrintTag codec's tests. +""" + +import struct + +import pytest + +from spoolman.tags.tigertag import ( + NTAG213_USER_BYTES, + TIGERTAG_INIT, + TIGERTAG_MAKER_V1, + TIGERTAG_PRO_V1, + TigerTagData, + decode_ntag213, + is_tigertag, +) + +# encode_ntag213 has no production caller -- nothing writes a TigerTag today -- so it lives +# here rather than in spoolman/tags/tigertag.py, same precedent set for OpenPrintTag's own +# encoder (spoolman/tags/openprinttag.py dropped encode_nfcv_memory for the same reason). +_HEADER_FMT = ">II HBB BBH I I HH BBBB I" +_USER_MESSAGE_SIZE = 28 +_USER_MESSAGE_OFFSET = 48 + + +def encode_ntag213(data: TigerTagData) -> bytes: + """Encode TigerTagData into raw bytes for NTAG213 user memory, test-only. + + Mirrors decode_ntag213's layout exactly (see that module's byte table) so round-trip + tests exercise real field packing, not a shortcut. + """ + color_val = (data.color_r << 24) | (data.color_g << 16) | (data.color_b << 8) | data.color_a + measure_unit = ((data.weight & 0xFFFFFF) << 8) | (data.id_unit & 0xFF) + + header = struct.pack( + _HEADER_FMT, + data.id_tigertag, + data.id_product, + data.id_material, + data.id_aspect, # aspect_1 + 0, # aspect_2 + data.id_type, + data.id_diameter, + data.id_brand, + color_val, + measure_unit, + data.nozzle_temp, + data.nozzle_temp_max, + data.drying_temp, + data.drying_duration, + data.bed_temp, + data.bed_temp_max, + data.timestamp, + ) + + payload = bytearray(NTAG213_USER_BYTES) + payload[: len(header)] = header + + msg_bytes = data.user_message.encode("utf-8")[:_USER_MESSAGE_SIZE] + payload[_USER_MESSAGE_OFFSET : _USER_MESSAGE_OFFSET + len(msg_bytes)] = msg_bytes + + return bytes(payload) + + +# --- is_tigertag ----------------------------------------------------------------- + + +def test_is_tigertag_true_for_maker_and_pro() -> None: + assert is_tigertag(TIGERTAG_MAKER_V1) + assert is_tigertag(TIGERTAG_PRO_V1) + + +def test_is_tigertag_false_for_init_and_junk() -> None: + # TIGERTAG_INIT marks a blank/uninitialized tag, deliberately excluded from the + # detectable-magic set. + assert not is_tigertag(TIGERTAG_INIT) + assert not is_tigertag(0xDEADBEEF) + + +# --- TigerTagData.color_hex -------------------------------------------------------- + + +def test_color_hex_getter_formats_rgb_only() -> None: + data = TigerTagData(color_r=0x11, color_g=0x22, color_b=0x33, color_a=0xFF) + assert data.color_hex == "112233" + + +def test_color_hex_setter_accepts_rgb() -> None: + data = TigerTagData() + data.color_hex = "aabbcc" + assert (data.color_r, data.color_g, data.color_b) == (0xAA, 0xBB, 0xCC) + + +def test_color_hex_setter_accepts_rgba_and_updates_alpha() -> None: + data = TigerTagData() + data.color_hex = "aabbcc80" + assert (data.color_r, data.color_g, data.color_b, data.color_a) == (0xAA, 0xBB, 0xCC, 0x80) + + +def test_color_hex_setter_strips_leading_hash() -> None: + data = TigerTagData() + data.color_hex = "#112233" + assert (data.color_r, data.color_g, data.color_b) == (0x11, 0x22, 0x33) + + +def test_color_hex_setter_ignores_wrong_length() -> None: + data = TigerTagData(color_r=1, color_g=2, color_b=3) + data.color_hex = "abc" # neither 6 nor 8 hex chars + assert (data.color_r, data.color_g, data.color_b) == (1, 2, 3) + + +# --- TigerTagData.diameter_mm ------------------------------------------------------- +# +# 56 (0x38) and 221 (0xDD) are the only ids the spec repo's database/id_diameter.json +# defines -- confirmed against every worked example in its README too. + + +def test_diameter_mm_175() -> None: + assert TigerTagData(id_diameter=56).diameter_mm == 1.75 + + +def test_diameter_mm_285() -> None: + assert TigerTagData(id_diameter=221).diameter_mm == 2.85 + + +def test_diameter_mm_unknown_id_is_zero() -> None: + assert TigerTagData(id_diameter=99).diameter_mm == 0.0 + + +# --- TigerTagData.weight_grams ------------------------------------------------------- +# +# database/id_measure_unit.json: 21=g is the only unit weight_grams trusts. The README's +# own Polymaker PolyTerra+ example reports its weight in kilograms (unit 35), which is +# exactly the case this guards against -- see test_decode_polymaker_example below. + + +def test_weight_grams_passes_through_for_grams_unit() -> None: + assert TigerTagData(weight=1000, id_unit=21).weight_grams == 1000 + + +def test_weight_grams_is_none_for_a_non_gram_unit() -> None: + assert TigerTagData(weight=1000, id_unit=35).weight_grams is None # 35 = kilograms + + +# --- decode_ntag213 / encode_ntag213 round trip ------------------------------------- + + +def _sample_data() -> TigerTagData: + data = TigerTagData( + id_tigertag=TIGERTAG_MAKER_V1, + id_product=0xFFFFFFFF, + id_material=38219, + id_diameter=56, + id_aspect=2, + id_type=142, + id_brand=19961, + weight=1000, + id_unit=21, + nozzle_temp=200, + nozzle_temp_max=220, + bed_temp=60, + bed_temp_max=70, + drying_temp=45, + drying_duration=6, + timestamp=800000000, + user_message="Hello Spoolman", + ) + data.color_hex = "112233" + data.color_a = 0xFF + return data + + +def test_encode_produces_full_user_memory_size() -> None: + assert len(encode_ntag213(_sample_data())) == NTAG213_USER_BYTES + + +def test_round_trip_preserves_all_fields() -> None: + original = _sample_data() + decoded = decode_ntag213(encode_ntag213(original)) + + assert decoded.id_tigertag == original.id_tigertag + assert decoded.id_product == original.id_product + assert decoded.id_material == original.id_material + assert decoded.id_aspect == original.id_aspect + assert decoded.id_type == original.id_type + assert decoded.id_diameter == original.id_diameter + assert decoded.id_brand == original.id_brand + assert decoded.color_hex == original.color_hex + assert decoded.color_a == original.color_a + assert decoded.weight == original.weight + assert decoded.id_unit == original.id_unit + assert decoded.nozzle_temp == original.nozzle_temp + assert decoded.nozzle_temp_max == original.nozzle_temp_max + assert decoded.bed_temp == original.bed_temp + assert decoded.bed_temp_max == original.bed_temp_max + assert decoded.drying_temp == original.drying_temp + assert decoded.drying_duration == original.drying_duration + assert decoded.timestamp == original.timestamp + assert decoded.user_message == original.user_message + + +def test_round_trip_is_tigertag_true_after_decode() -> None: + decoded = decode_ntag213(encode_ntag213(_sample_data())) + assert is_tigertag(decoded.id_tigertag) + + +def test_encode_truncates_user_message_over_28_bytes() -> None: + data = _sample_data() + data.user_message = "x" * 40 + decoded = decode_ntag213(encode_ntag213(data)) + assert decoded.user_message == "x" * 28 + + +def test_encode_masks_weight_to_24_bits() -> None: + data = _sample_data() + data.weight = 0xFFFFFFFF # larger than the 24-bit field can hold + decoded = decode_ntag213(encode_ntag213(data)) + assert decoded.weight == 0xFFFFFF + + +def test_decode_raises_on_too_short_data() -> None: + with pytest.raises(ValueError, match="Data too short"): + decode_ntag213(b"\x00" * 10) + + +def test_decode_accepts_header_only_data_and_defaults_the_rest() -> None: + # Exactly the 36-byte header, no user-message region. + raw = encode_ntag213(_sample_data())[:36] + decoded = decode_ntag213(raw) + assert decoded.user_message == "" + + +def test_decode_user_message_stops_at_null_terminator() -> None: + data = _sample_data() + data.user_message = "short" + raw = bytearray(encode_ntag213(data)) + # Message is null-padded already by encode; decode must stop at the first \x00 + # rather than reading through the padding as garbage. + decoded = decode_ntag213(bytes(raw)) + assert decoded.user_message == "short" + + +def test_decode_user_message_handles_invalid_utf8_with_replacement() -> None: + data = _sample_data() + data.user_message = "ok" + raw = bytearray(encode_ntag213(data)) + # Corrupt one byte of the (short, non-null-padded) message region to invalid UTF-8. + raw[_USER_MESSAGE_OFFSET] = 0xFF + decoded = decode_ntag213(bytes(raw)) + assert decoded.user_message.startswith("\N{REPLACEMENT CHARACTER}") + + +# --- Real reference payloads, from the spec repo's README ------------------------------- + + +def test_decode_rosa3d_red_pla_example() -> None: + """README section 4: TigerTag-encoded Rosa3D red PLA, a plain Maker tag.""" + data = TigerTagData( + id_tigertag=0x5BF59264, + id_product=0xFFFFFFFF, + id_material=0x954B, # 38219 + id_aspect=0x68, # 104, "Basic" + id_type=0x8E, # 142, Filament + id_diameter=0x38, # 56, 1.75mm + id_brand=0x4E19, # 19961, Rosa3D + weight=0x0003E8, # 1000 + id_unit=0x15, # 21, grams + nozzle_temp=0x00C3, # 195 + nozzle_temp_max=0x00E6, # 230 + drying_temp=0x32, # 50 + drying_duration=0x05, + bed_temp=0x32, # 50 + bed_temp_max=0x3C, # 60 + timestamp=0x2F7A0A5C, # 796527196 + user_message="Starter Red", + ) + data.color_hex = "FF0000" + data.color_a = 0xFF + + decoded = decode_ntag213(encode_ntag213(data)) + + assert is_tigertag(decoded.id_tigertag) + assert decoded.diameter_mm == 1.75 + assert decoded.weight_grams == 1000 + assert decoded.color_hex == "ff0000" + assert decoded.nozzle_temp == 195 + assert decoded.nozzle_temp_max == 230 + assert decoded.bed_temp == 50 + assert decoded.bed_temp_max == 60 + assert decoded.user_message == "Starter Red" + + +def test_decode_polymaker_tigertag_plus_example_reports_weight_in_kilograms() -> None: + """README section 4.1: a TigerTag+ tag whose Measure is in kilograms, not grams. + + This is the real-world case weight_grams exists to guard: trusting the raw measure + as grams here would read 1000 kg as 1000 g, a thousandfold error. + """ + data = TigerTagData( + id_tigertag=0xBC0FCB97, # TigerTag+ V1.0 + id_product=0x0000000A, + id_material=0x954B, # 38219, PLA + id_aspect=0x86, # 134, Matt + id_type=0x8E, # Filament + id_diameter=0x38, # 56, 1.75mm + id_brand=0xC5DC, # 50652, Polymaker + weight=0x0003E8, # 1000 + id_unit=0x23, # 35, kilograms -- not grams + nozzle_temp=0x00BE, # 190 + nozzle_temp_max=0x00F0, # 240 + drying_temp=0x37, # 55 + drying_duration=0x06, + bed_temp=0x23, # 35 + bed_temp_max=0x41, # 65 + timestamp=0x2F7A0E90, # 796528272 + user_message="Private msg", + ) + data.color_hex = "89D9D9" + data.color_a = 0xFF + + decoded = decode_ntag213(encode_ntag213(data)) + + assert is_tigertag(decoded.id_tigertag) + assert decoded.diameter_mm == 1.75 + assert decoded.weight == 1000 # the raw measure round-trips... + assert decoded.weight_grams is None # ...but it is not usable as grams. + assert decoded.color_hex == "89d9d9" + + +def test_decode_init_tag_example() -> None: + """README section 4.2: a blank/uninitialized Init tag, every field at its default.""" + decoded = decode_ntag213(encode_ntag213(TigerTagData(id_tigertag=0x6C41A2E1, id_unit=0x00))) + + assert not is_tigertag(decoded.id_tigertag) + assert decoded.id_tigertag == TIGERTAG_INIT + assert decoded.diameter_mm == 0.0 + # Unit ID 0x00 is "Not defined" on a blank tag -- not grams, so not a usable weight. + assert decoded.weight_grams is None diff --git a/tests/test_tigertagdb.py b/tests/test_tigertagdb.py new file mode 100644 index 000000000..4ae1ae17e --- /dev/null +++ b/tests/test_tigertagdb.py @@ -0,0 +1,333 @@ +"""Tests for the optional TigerTag external database add-on.""" + +import json +from collections.abc import Callable +from pathlib import Path +from unittest.mock import MagicMock + +import httpx +import pytest + +from spoolman import tigertagdb +from spoolman.tigertagdb import TigerTagProduct, _parse_weight_from_measure, _to_external_filament + +# --- _parse_weight_from_measure ----------------------------------------------------- + + +@pytest.mark.parametrize( + ("measure", "expected"), + [ + ("1 kg", 1000.0), + ("1kg", 1000.0), + ("0.75 kg", 750.0), + ("500 g", 500.0), + ("500g", 500.0), + (None, 1000.0), + ("", 1000.0), + ("not a weight", 1000.0), + ], +) +def test_parse_weight_from_measure(measure: str | None, expected: float) -> None: + assert _parse_weight_from_measure(measure) == expected + + +# --- _to_external_filament ----------------------------------------------------------- + + +def test_to_external_filament_full_product() -> None: + product = TigerTagProduct( + id=2995423176, + brand="Rosa3D", + title="Rosa3D PLA Starter", + material="PLA", + color="112233ff", + measure="1 kg", + ) + ext = _to_external_filament(product) + + assert ext.id == "tigertag_2995423176" + assert ext.manufacturer == "Rosa3D" + assert ext.name == "Rosa3D PLA Starter" + assert ext.material == "PLA" + assert ext.weight == 1000.0 + assert ext.diameter == 1.75 + assert ext.color_hex == "112233" # alpha channel stripped + assert ext.source == "tigertag" + + +def test_to_external_filament_strips_leading_hash_and_keeps_rgb_only_color() -> None: + product = TigerTagProduct(id=1, color="#aabbcc") + ext = _to_external_filament(product) + assert ext.color_hex == "aabbcc" + + +def test_to_external_filament_resolves_material_and_weight_from_nested_filament_block() -> None: + """Real shape of GET product/get (the live, single-tag lookup `lookup_product_by_tag` uses). + + Checked 2026-11-09 against api.tigertag.io/api:tigertag/product/get?uid=123456789&product_id=10 + (Polymaker PolyTerra Artic Teal): unlike product/get/all's bulk shape -- still flat, see + test_to_external_filament_full_product -- this one nests material and weight under + "filament" instead of putting them at the top level. Trimmed to the fields that matter here. + """ + product = TigerTagProduct( + id=10, + brand="Polymaker", + title="Polymaker - PolyTerra™ - Artic Teal - 1.75 mm - 1000 g", + color="#89D9D9FF", + filament={ + "material": "PLA", + "grams": 1000, + }, + ) + ext = _to_external_filament(product) + + assert ext.manufacturer == "Polymaker" + assert ext.material == "PLA" + assert ext.weight == 1000.0 + assert ext.color_hex == "89D9D9" + + +def test_to_external_filament_defaults_when_fields_missing() -> None: + product = TigerTagProduct(id=1) + ext = _to_external_filament(product) + + assert ext.manufacturer == "Unknown" + assert ext.material == "Unknown" + assert ext.name == "Unknown Unknown" + assert ext.color_hex is None + assert ext.weight == 1000.0 + + +# --- lookup_brand_name / lookup_material_name / lookup_material_density ------------- + + +@pytest.fixture +def cache_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.setattr(tigertagdb.filecache, "get_cache_dir", lambda: tmp_path) + return tmp_path + + +def test_lookup_brand_name_hit(cache_dir: Path) -> None: + (cache_dir / tigertagdb.TIGERTAG_BRANDS_CACHE_FILE).write_text( + json.dumps([{"id": 19961, "name": "Rosa3D", "type_ids": [142]}]), + ) + assert tigertagdb.lookup_brand_name(19961) == "Rosa3D" + + +def test_lookup_brand_name_miss_returns_none(cache_dir: Path) -> None: + (cache_dir / tigertagdb.TIGERTAG_BRANDS_CACHE_FILE).write_text(json.dumps([{"id": 1, "name": "Other"}])) + assert tigertagdb.lookup_brand_name(99999) is None + + +def test_lookup_brand_name_no_cache_file_returns_none(cache_dir: Path) -> None: + assert not (cache_dir / tigertagdb.TIGERTAG_BRANDS_CACHE_FILE).exists() + assert tigertagdb.lookup_brand_name(19961) is None + + +def test_lookup_brand_name_corrupt_cache_returns_none(cache_dir: Path) -> None: + (cache_dir / tigertagdb.TIGERTAG_BRANDS_CACHE_FILE).write_text("not json") + assert tigertagdb.lookup_brand_name(19961) is None + + +def test_lookup_material_name_hit(cache_dir: Path) -> None: + (cache_dir / tigertagdb.TIGERTAG_MATERIALS_CACHE_FILE).write_text( + json.dumps([{"id": 38219, "label": "PLA", "density": 1.24}]), + ) + assert tigertagdb.lookup_material_name(38219) == "PLA" + + +def test_lookup_material_density_hit(cache_dir: Path) -> None: + (cache_dir / tigertagdb.TIGERTAG_MATERIALS_CACHE_FILE).write_text( + json.dumps([{"id": 38219, "label": "PLA", "density": 1.24}]), + ) + assert tigertagdb.lookup_material_density(38219) == 1.24 + + +def test_lookup_material_density_missing_density_field_returns_none(cache_dir: Path) -> None: + (cache_dir / tigertagdb.TIGERTAG_MATERIALS_CACHE_FILE).write_text(json.dumps([{"id": 1, "label": "PLA"}])) + assert tigertagdb.lookup_material_density(1) is None + + +# --- lookup_product_by_tag (real-time API) ------------------------------------------- + + +def _mock_client(monkeypatch: pytest.MonkeyPatch, handler: Callable[[httpx.Request], httpx.Response]) -> None: + transport = httpx.MockTransport(handler) + # tigertagdb.httpx and this module's httpx are the same module object (Python + # caches imports), so capture the real class before patching it -- otherwise the + # factory below would call the patched attribute and recurse into itself. + real_async_client = httpx.AsyncClient + + def _factory(*args: object, **kwargs: object) -> httpx.AsyncClient: + kwargs["transport"] = transport + return real_async_client(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(tigertagdb.httpx, "AsyncClient", _factory) + + +@pytest.mark.asyncio +async def test_lookup_product_by_tag_disabled_makes_no_request(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tigertagdb, "is_tigertag_enabled", lambda: False) + + def _handler(_request: httpx.Request) -> httpx.Response: + pytest.fail("No HTTP request should be made when the add-on is disabled") + + _mock_client(monkeypatch, _handler) + + assert await tigertagdb.lookup_product_by_tag("04A2B3C4", 28) is None + + +@pytest.mark.asyncio +async def test_lookup_product_by_tag_success(monkeypatch: pytest.MonkeyPatch) -> None: + """Uses the real nested-filament shape this endpoint actually returns. + + See test_to_external_filament_resolves_material_and_weight_from_nested_filament_block + for where/when that was checked against the live API. + """ + monkeypatch.setattr(tigertagdb, "is_tigertag_enabled", lambda: True) + + def _handler(request: httpx.Request) -> httpx.Response: + assert request.url.params["uid"] == "04A2B3C4" + assert request.url.params["product_id"] == "28" + return httpx.Response( + 200, + json={"id": 28, "brand": "Rosa3D", "filament": {"material": "PLA", "grams": 1000}}, + ) + + _mock_client(monkeypatch, _handler) + + ext = await tigertagdb.lookup_product_by_tag("04A2B3C4", 28) + assert ext is not None + assert ext.id == "tigertag_28" + assert ext.manufacturer == "Rosa3D" + assert ext.material == "PLA" + assert ext.weight == 1000.0 + + +@pytest.mark.asyncio +async def test_lookup_product_by_tag_not_found_returns_none(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tigertagdb, "is_tigertag_enabled", lambda: True) + _mock_client(monkeypatch, lambda _r: httpx.Response(404)) + + assert await tigertagdb.lookup_product_by_tag("04A2B3C4", 28) is None + + +@pytest.mark.asyncio +async def test_lookup_product_by_tag_server_error_returns_none(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tigertagdb, "is_tigertag_enabled", lambda: True) + _mock_client(monkeypatch, lambda _r: httpx.Response(500)) + + assert await tigertagdb.lookup_product_by_tag("04A2B3C4", 28) is None + + +@pytest.mark.asyncio +async def test_lookup_product_by_tag_malformed_body_returns_none(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tigertagdb, "is_tigertag_enabled", lambda: True) + _mock_client(monkeypatch, lambda _r: httpx.Response(200, content=b"not json")) + + assert await tigertagdb.lookup_product_by_tag("04A2B3C4", 28) is None + + +# --- _sync_tigertag end-to-end (mocked API + cache dir) ------------------------------- + + +@pytest.mark.asyncio +async def test_sync_tigertag_writes_filaments_brands_and_materials_cache( + cache_dir: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("product/get/all"): + body = json.loads(request.content) + if body["page"] == 1: + return httpx.Response( + 200, + json={ + "items": [{"id": 1, "brand": "Rosa3D", "material": "PLA", "measure": "1 kg"}], + "nextPage": 2, + }, + ) + return httpx.Response(200, json={"items": [{"id": 2, "brand": "Sunlu", "material": "PETG"}]}) + if request.url.path.endswith("brand/get/all"): + return httpx.Response(200, json={"items": [{"id": 19961, "name": "Rosa3D"}]}) + if request.url.path.endswith("material/get/all"): + return httpx.Response(200, json={"items": [{"id": 38219, "label": "PLA", "density": 1.24}]}) + pytest.fail(f"Unexpected request: {request.url}") + + _mock_client(monkeypatch, _handler) + + await tigertagdb._sync_tigertag() # noqa: SLF001 + + filaments = json.loads((cache_dir / tigertagdb.TIGERTAG_CACHE_FILE).read_bytes()) + assert {f["id"] for f in filaments} == {"tigertag_1", "tigertag_2"} + + brands = json.loads((cache_dir / tigertagdb.TIGERTAG_BRANDS_CACHE_FILE).read_bytes()) + assert brands == [{"id": 19961, "name": "Rosa3D"}] + + materials = json.loads((cache_dir / tigertagdb.TIGERTAG_MATERIALS_CACHE_FILE).read_bytes()) + assert materials == [{"id": 38219, "label": "PLA", "density": 1.24}] + + +@pytest.mark.asyncio +async def test_sync_tigertag_one_endpoint_failing_does_not_block_the_others( + cache_dir: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Cover the three-way independence of the sync steps. + + Products, brands and materials are synced independently; a broken products fetch + (say, an API outage mid-pagination) shouldn't prevent the brand/material caches -- + which other lookups depend on -- from still being refreshed. + """ + + def _handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("product/get/all"): + return httpx.Response(500) + if request.url.path.endswith("brand/get/all"): + return httpx.Response(200, json={"items": [{"id": 1, "name": "Rosa3D"}]}) + if request.url.path.endswith("material/get/all"): + return httpx.Response(200, json={"items": [{"id": 1, "label": "PLA"}]}) + pytest.fail(f"Unexpected request: {request.url}") + + _mock_client(monkeypatch, _handler) + + await tigertagdb._sync_tigertag() # noqa: SLF001 + + assert not (cache_dir / tigertagdb.TIGERTAG_CACHE_FILE).exists() + assert (cache_dir / tigertagdb.TIGERTAG_BRANDS_CACHE_FILE).exists() + assert (cache_dir / tigertagdb.TIGERTAG_MATERIALS_CACHE_FILE).exists() + + +# --- schedule_tasks ------------------------------------------------------------------- + + +def test_schedule_tasks_disabled_schedules_nothing(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tigertagdb, "is_tigertag_enabled", lambda: False) + scheduler = MagicMock() + + tigertagdb.schedule_tasks(scheduler) + + scheduler.once.assert_not_called() + scheduler.cyclic.assert_not_called() + + +def test_schedule_tasks_enabled_schedules_startup_and_cyclic_sync(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tigertagdb, "is_tigertag_enabled", lambda: True) + monkeypatch.setattr(tigertagdb, "get_tigertag_sync_interval", lambda: 3600) + scheduler = MagicMock() + + tigertagdb.schedule_tasks(scheduler) + + scheduler.once.assert_called_once() + scheduler.cyclic.assert_called_once() + + +def test_schedule_tasks_zero_interval_skips_cyclic_sync(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tigertagdb, "is_tigertag_enabled", lambda: True) + monkeypatch.setattr(tigertagdb, "get_tigertag_sync_interval", lambda: 0) + scheduler = MagicMock() + + tigertagdb.schedule_tasks(scheduler) + + scheduler.once.assert_called_once() + scheduler.cyclic.assert_not_called() diff --git a/tests_integration/tests/_tigertag_fixtures.py b/tests_integration/tests/_tigertag_fixtures.py new file mode 100644 index 000000000..62561f390 --- /dev/null +++ b/tests_integration/tests/_tigertag_fixtures.py @@ -0,0 +1,62 @@ +"""Build a synthetic TigerTag NTAG213 memory dump for integration tests. + +Same construction as spoolman's own unit tests (tests/test_tigertag_codec.py, +tests/test_tag_decode.py), duplicated here rather than imported: this package runs inside +the separate `spoolman-tester` container/image, which has no access to the `spoolman` +package's own test modules. Unlike OpenPrintTag, this needs no extra dependency -- the +TigerTag format is plain stdlib struct packing. +""" + +import struct + +TIGERTAG_MAKER_V1 = 0x5BF59264 +TIGERTAG_INIT = 0x6C41A2E1 + +NTAG213_USER_BYTES = 144 + +# Field order/offsets per spoolman/tags/tigertag.py's byte table, checked against the real +# spec repo (TigerTag-Project/TigerTag-RFID-Guide). 56 (0x38) and 221 (0xDD) are the only +# diameter ids the spec defines; 21 is the "grams" measure-unit id. +_HEADER_FMT = ">II HBB BBH I I HH BBBB I" +_UNIT_ID_GRAMS = 21 + + +def build_tigertag( + *, + id_tigertag: int = TIGERTAG_MAKER_V1, + id_material: int = 0, + id_diameter: int = 56, + id_brand: int = 0, + weight: int = 0, + id_unit: int = _UNIT_ID_GRAMS, + color_hex: str = "000000", +) -> bytes: + """Build a full 144-byte NTAG213 user memory dump wrapping TigerTag fields.""" + color_r, color_g, color_b = (int(color_hex[i : i + 2], 16) for i in (0, 2, 4)) + color_val = (color_r << 24) | (color_g << 16) | (color_b << 8) | 0xFF + measure_unit = ((weight & 0xFFFFFF) << 8) | (id_unit & 0xFF) + + header = struct.pack( + _HEADER_FMT, + id_tigertag, + 0, # id_product + id_material, + 0, # id_aspect + 0, # aspect_2 + 142, # id_type: filament + id_diameter, + id_brand, + color_val, + measure_unit, + 0, # nozzle_temp + 0, # nozzle_temp_max + 0, # drying_temp + 0, # drying_duration + 0, # bed_temp_min + 0, # bed_temp_max + 0, # timestamp + ) + + payload = bytearray(NTAG213_USER_BYTES) + payload[: len(header)] = header + return bytes(payload) diff --git a/tests_integration/tests/tag/test_scan_create.py b/tests_integration/tests/tag/test_scan_create.py index a5cd07608..9d0d8fbb7 100644 --- a/tests_integration/tests/tag/test_scan_create.py +++ b/tests_integration/tests/tag/test_scan_create.py @@ -19,6 +19,7 @@ MF_PRIMARY_COLOR, build_openprinttag, ) +from .._tigertag_fixtures import build_tigertag from ..conftest import URL, assert_httpx_success @@ -141,3 +142,35 @@ def test_create_true_with_an_undecodable_payload_creates_nothing(): listing = httpx.get(f"{URL}/api/v1/spool", params={"tag": uid}) assert_httpx_success(listing) assert listing.json() == [] + + +def test_create_true_on_an_unmatched_tigertag_creates_and_links_a_spool_without_names(): + """Verify auto-create still works without the external-DB addon. + + TigerTag's material/brand are catalog IDs, not names -- material/vendor still get + created (no addon needed to make a usable spool), just without a name. + """ + uid = _uid() + payload = base64.b64encode(build_tigertag(id_diameter=56, weight=1000, color_hex="112233")).decode("ascii") + + result = httpx.post( + f"{URL}/api/v1/tag/scan", + json={ + "uid": uid, + "reader_id": _reader_id(), + "format": "tigertag", + "payload_b64": payload, + "create": True, + }, + ) + assert_httpx_success(result) + body = result.json() + try: + assert body["created"] is True + assert body["spool"]["filament"].get("material") is None + assert body["spool"]["filament"].get("vendor") is None + assert body["spool"]["filament"]["diameter"] == 1.75 + assert body["spool"]["filament"]["weight"] == 1000.0 + assert body["spool"]["filament"]["color_hex"] == "112233" + finally: + _delete_created_spool(body["spool"]) diff --git a/tests_integration/tests/tag/test_scan_decode.py b/tests_integration/tests/tag/test_scan_decode.py index c4ffa95b2..daa85d523 100644 --- a/tests_integration/tests/tag/test_scan_decode.py +++ b/tests_integration/tests/tag/test_scan_decode.py @@ -22,6 +22,7 @@ REAL_PRUSA_TEST01_TAG_UID_HEX, build_openprinttag, ) +from .._tigertag_fixtures import TIGERTAG_INIT, build_tigertag from ..conftest import URL, assert_httpx_success @@ -166,3 +167,35 @@ def test_scan_decodes_a_real_prusa_test_vector_with_derived_instance_id(): assert decoded["spool_weight"] == 280.0 assert decoded["color_hex"] == "3d3e3d" assert decoded["instance_uuid"] == "bf63e92d-9ca5-53d7-9fab-ffdd0240c585" + + +# --- TigerTag --------------------------------------------------------------------- + + +def test_scan_decodes_a_tigertag(): + payload = base64.b64encode(build_tigertag(id_diameter=56, weight=1000, color_hex="112233")).decode("ascii") + result = httpx.post( + f"{URL}/api/v1/tag/scan", + json={"uid": _uid(), "reader_id": _reader_id(), "format": "tigertag", "payload_b64": payload}, + ) + assert_httpx_success(result) + + decoded = result.json()["decoded"] + assert decoded["color_hex"] == "112233" + assert decoded["diameter"] == 1.75 + assert decoded["weight"] == 1000.0 + assert decoded["material_class"] == "FFF" + # id_material/id_brand need the optional external-DB addon to resolve to names. + assert decoded.get("material") is None + assert decoded.get("vendor_name") is None + + +def test_scan_with_a_blank_tigertag_has_no_decoded_field(): + """TIGERTAG_INIT marks an uninitialized tag -- nothing usable to surface.""" + payload = base64.b64encode(build_tigertag(id_tigertag=TIGERTAG_INIT)).decode("ascii") + result = httpx.post( + f"{URL}/api/v1/tag/scan", + json={"uid": _uid(), "reader_id": _reader_id(), "format": "tigertag", "payload_b64": payload}, + ) + assert_httpx_success(result) + assert "decoded" not in result.json() diff --git a/tests_integration/tests/tag/test_scan_tigertag_hostile.py b/tests_integration/tests/tag/test_scan_tigertag_hostile.py new file mode 100644 index 000000000..9216e7559 --- /dev/null +++ b/tests_integration/tests/tag/test_scan_tigertag_hostile.py @@ -0,0 +1,131 @@ +"""Integration tests for POST /tag/scan with hostile or malformed TigerTag payloads. + +TigerTag is a fixed-width, big-endian `struct`-packed binary format, not a self-describing +one like OpenPrintTag's CBOR -- so the type-confusion attacks test_scan_hostile.py covers +(a number field holding a list, a bignum, NaN, Infinity) cannot arrive here: every field is +a plain unsigned int of a fixed byte width, and there is no way to encode a float, a +container, or an out-of-range Python type into any of them. What TigerTag *can* carry that +still needs guarding against is a value that's technically a valid unsigned int but wrong +in a way that matters (a weight over any real spool's size, a unit that silently rescales +the weight, truncated or garbage bytes) -- this file covers those instead. +""" + +import base64 +import uuid +from typing import Any + +import httpx + +from .._tigertag_fixtures import TIGERTAG_INIT, build_tigertag +from ..conftest import URL, assert_httpx_success +from .test_scan_create import _delete_created_spool +from .test_scan_hostile import _assert_list_endpoints_still_work, _row_counts + + +def _uid() -> str: + return uuid.uuid4().hex[:14].upper() + + +def _reader_id() -> str: + return f"reader-{uuid.uuid4().hex[:8]}" + + +def _scan(raw: bytes, *, create: bool | None = None) -> httpx.Response: + body: dict[str, Any] = { + "uid": _uid(), + "reader_id": _reader_id(), + "format": "tigertag", + "payload_b64": base64.b64encode(raw).decode("ascii"), + } + if create is not None: + body["create"] = create + return httpx.post(f"{URL}/api/v1/tag/scan", json=body) + + +def test_weight_over_the_supported_range_creates_nothing(): + """The 24-bit measure field's max is still accepted by the codec, but far too heavy. + + 16,777,215g is well outside any real spool -- the shared range check in + database/tag.py must refuse it. + """ + before = _row_counts() + raw = build_tigertag(weight=0xFFFFFF, color_hex="112233") # grams unit (the default) + result = _scan(raw, create=True) + assert_httpx_success(result) + body = result.json() + + assert body.get("created") is not True + assert "spool" not in body + assert _row_counts() == before + _assert_list_endpoints_still_work() + + +def test_a_non_gram_unit_creates_a_spool_with_no_weight_instead_of_a_scaled_one(): + """A tag reporting its weight in kilograms must not silently become a wrong gram value. + + It must come through as no weight at all, same as OpenPrintTag's never-defined fields. + """ + raw = build_tigertag(weight=1000, id_unit=35, color_hex="112233") # 35 = kilograms + result = _scan(raw, create=True) + assert_httpx_success(result) + body = result.json() + try: + assert body["created"] is True + assert body["spool"]["filament"].get("weight") is None + finally: + _delete_created_spool(body["spool"]) + + +def test_too_short_payload_creates_nothing(): + before = _row_counts() + result = _scan(b"\x00" * 10, create=True) + assert_httpx_success(result) + body = result.json() + + assert body.get("created") is not True + assert "decoded" not in body + assert _row_counts() == before + _assert_list_endpoints_still_work() + + +def test_unmapped_magic_number_creates_nothing(): + before = _row_counts() + raw = build_tigertag(id_tigertag=0xDEADBEEF, color_hex="112233") + result = _scan(raw, create=True) + assert_httpx_success(result) + body = result.json() + + assert body.get("created") is not True + assert "decoded" not in body + assert _row_counts() == before + _assert_list_endpoints_still_work() + + +def test_blank_init_tag_creates_nothing(): + before = _row_counts() + raw = build_tigertag(id_tigertag=TIGERTAG_INIT) + result = _scan(raw, create=True) + assert_httpx_success(result) + body = result.json() + + assert body.get("created") is not True + assert "decoded" not in body + assert _row_counts() == before + _assert_list_endpoints_still_work() + + +def test_unmapped_diameter_id_still_creates_using_the_default(): + """An id_diameter the codec doesn't recognize decodes as 'not set', same as OpenPrintTag. + + Create still succeeds, falling back to the 1.75mm default rather than refusing the + whole tag. + """ + raw = build_tigertag(id_diameter=99, weight=1000, color_hex="112233") + result = _scan(raw, create=True) + assert_httpx_success(result) + body = result.json() + try: + assert body["created"] is True + assert body["spool"]["filament"]["diameter"] == 1.75 + finally: + _delete_created_spool(body["spool"])