feat(iso15): add inventory() with 1/16-slot parsing + AFI/DSFID filtering

inventory() parses both response formats: slots=1 -> single tag dict (or None); slots=16 -> list of tags from the iso15_inventory_response_t anti-collision struct, each CRC15-validated. Optional request-level AFI filter and client-side DSFID filter. One 16-slot round drops colliding slots (full tree-walk resolution is a follow-up); scan() stays the single-tag convenience.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

pm3py/core/hf_iso15.py

@@ -1,6 +1,6 @@
 """ISO 15693 commands: hf.iso15.*"""
 import struct
-from .protocol import Cmd, crc15_bytes
+from .protocol import Cmd, crc15, crc15_bytes
 from .transport import PM3Transport, PM3Error
 
 # Re-export trace symbols for backward compat with existing test imports
@@ -40,6 +40,24 @@ def _iso15_payload(iso_cmd: bytes, flags: int = _READ_FLAGS) -> bytes:
     return struct.pack("<BH", flags, len(raw)) + raw
 
 
+# ISO15693 request-flags byte (raw[0]) — the protocol flags, NOT the PM3 transport flags
+REQ_DATARATE_HIGH = 0x02
+REQ_INVENTORY = 0x04
+REQINV_AFI = 0x10        # an AFI byte follows the INVENTORY command
+REQINV_SLOT1 = 0x20      # set = 1 slot, clear = 16-slot anti-collision
+
+
+def _parse_inv_record(rec: bytes) -> dict | None:
+    """Parse one inventory response ``[resp_flags][dsfid][uid(8)][crc(2)]``.
+
+    Returns ``{'uid', 'dsfid', 'flags'}`` (uid as display-order hex), or None if
+    the record is too short or the CRC is bad (i.e. a collision/garbled slot).
+    """
+    if len(rec) < 12 or crc15(rec[:-2]) != (rec[-2] | (rec[-1] << 8)):
+        return None
+    return {"uid": rec[2:10][::-1].hex(), "dsfid": rec[1], "flags": rec[0]}
+
+
 class HF15Commands:
     """ISO 15693 commands."""
 
@@ -82,6 +100,62 @@ class HF15Commands:
         return {"inv_flags": hex(inv_flags), "status": resp.status,
                 "len": len(resp.data), "data": resp.data.hex()}
 
+    async def inventory(self, slots: int = 16, afi: int | None = None,
+                        dsfid: int | None = None):
+        """Perform an ISO15693 inventory (anti-collision).
+
+        slots=16 (default): one 16-slot anti-collision round → **list** of tag
+            dicts (empty list if none). Handles up to 16 non-colliding tags per
+            round; slots that collide are dropped (CRC fail).
+        slots=1: single-slot → **one** tag dict, or None if no/garbled response.
+        afi: request-level Application Family Identifier — only tags with this AFI
+            answer (sent in the inventory command).
+        dsfid: client-side filter — keep only tags whose DSFID equals this.
+
+        Each tag: ``{'uid': str, 'dsfid': int, 'flags': int[, 'slot': int]}``.
+        (`scan()` remains the quick "is one tag present" convenience.)
+        """
+        req = REQ_DATARATE_HIGH | REQ_INVENTORY
+        if slots == 1:
+            req |= REQINV_SLOT1
+        if afi is not None:
+            req |= REQINV_AFI
+        iso_cmd = bytes([req, 0x01])          # request flags, INVENTORY command
+        if afi is not None:
+            iso_cmd += bytes([afi & 0xFF])    # AFI byte precedes the mask
+        iso_cmd += bytes([0x00])              # mask length = 0
+        resp = await self._t.send_ng(
+            Cmd.HF_ISO15693_COMMAND, _iso15_payload(iso_cmd), timeout=5.0)
+
+        def _keep(tag):
+            return tag is not None and (dsfid is None or tag["dsfid"] == dsfid)
+
+        if slots == 1:
+            if resp.status != 0:
+                return None
+            tag = _parse_inv_record(bytes(resp.data))
+            return tag if _keep(tag) else None
+
+        # 16-slot: iso15_inventory_response_t = [slot_count][16x{status,len}][data]
+        tags = []
+        data = bytes(resp.data)
+        if resp.status != 0 or len(data) < 1 + 16 * 2:
+            return tags
+        slot_count = data[0]
+        off = 1 + slot_count * 2              # concatenated slot data starts here
+        for s in range(slot_count):
+            status = data[1 + s * 2]
+            rlen = data[1 + s * 2 + 1]
+            rec = data[off:off + rlen]
+            off += rlen
+            if status != 1 or rlen == 0:      # 0=empty, 2=collision
+                continue
+            tag = _parse_inv_record(rec)
+            if _keep(tag):
+                tag["slot"] = s
+                tags.append(tag)
+        return tags
+
     async def rdbl(self, block: int, uid: str | None = None) -> dict:
         """Read a single block."""
         # OPTION (0x40) makes the tag return the block-security (lock) byte this parser

tests/test_hf_15.py

@@ -286,3 +286,43 @@ class TestSniffLineAnnotationOverflow:
                  "timestamp": 0, "duration": 0, "parity": b""}
         line = format_sniff_line(entry, width=120, is_tty=False)
         assert "CC" in line
+
+
+def _inv_record(uid_display_hex, dsfid=0, flags=0):
+    from pm3py.core.protocol import crc15_bytes
+    body = bytes([flags, dsfid]) + bytes.fromhex(uid_display_hex)[::-1]
+    return body + crc15_bytes(body)
+
+
+def test_15_inventory_single_slot():
+    t = AsyncMock()
+    iso = HF15Commands(t)
+    t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0,
+                                         reason=0, ng=True, data=_inv_record("e004a2110b37c433"))
+    tag = asyncio.get_event_loop().run_until_complete(iso.inventory(slots=1))
+    assert tag == {"uid": "e004a2110b37c433", "dsfid": 0, "flags": 0}
+    # SLOT1 flag set in the request (raw[0] bit 0x20)
+    payload = t.send_ng.call_args.args[1]
+    assert payload[3] & 0x20  # raw[0] after 3-byte flags+len header
+
+
+def test_15_inventory_16_slot_list_and_dsfid_filter():
+    t = AsyncMock()
+    iso = HF15Commands(t)
+    r0 = _inv_record("e004a2110b37c433")
+    r3 = _inv_record("e004a2110b3746e0", dsfid=5)
+    meta = bytearray(16 * 2)
+    meta[0:2] = bytes([1, len(r0)])   # slot 0 valid
+    meta[6:8] = bytes([1, len(r3)])   # slot 3 valid
+    meta[10:12] = bytes([2, 0])       # slot 5 collision
+    data = bytes([16]) + bytes(meta) + r0 + r3
+    t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0,
+                                         reason=0, ng=True, data=data)
+    tags = asyncio.get_event_loop().run_until_complete(iso.inventory(slots=16))
+    assert [x["uid"] for x in tags] == ["e004a2110b37c433", "e004a2110b3746e0"]
+    assert tags[0]["slot"] == 0 and tags[1]["slot"] == 3
+    # client-side DSFID filter
+    t.send_ng.return_value = PM3Response(cmd=Cmd.HF_ISO15693_COMMAND, status=0,
+                                         reason=0, ng=True, data=data)
+    only5 = asyncio.get_event_loop().run_until_complete(iso.inventory(slots=16, dsfid=5))
+    assert [x["uid"] for x in only5] == ["e004a2110b3746e0"]