Add support for CIS-6 Track and Trace

Fixes #1

Author: sderuiter

ccdexplorer_fundamentals/cis.py

@@ -139,6 +139,8 @@ class StandardIdentifiers(Enum):
     CIS_2 = "CIS-2"
     CIS_3 = "CIS-3"
     CIS_4 = "CIS-4"
+    CIS_5 = "CIS-5"
+    CIS_6 = "CIS-6"
 
 
 class LoggedEvents(Enum):
@@ -240,6 +242,20 @@ class tokenMetadataEvent(BaseModel):
     metadata: MetadataUrl
 
 
+class itemCreatedEvent(BaseModel):
+    tag: int
+    item_id: str
+    metadata: MetadataUrl
+    initial_status: str
+
+
+class itemStatusChangedEvent(BaseModel):
+    tag: int
+    item_id: str
+    new_status: str
+    additional_data: bytes
+
+
 class nonceEvent(BaseModel):
     tag: int
     nonce: Optional[int] = None
@@ -705,7 +721,7 @@ def support_result(self, bs: io.BytesIO):
     def supports_response(self, res: bytes):
         bs = io.BytesIO(bytes.fromhex(res.decode()))
         if bs.getbuffer().nbytes > 0:
-            n = int.from_bytes(bs.read(2), byteorder="big")
+            n = int.from_bytes(bs.read(2), byteorder="little")
             responses = []
             for _ in range(n):
                 responses.append(self.support_result(bs))
@@ -970,6 +986,13 @@ def token_id(self, bs: io.BytesIO):
         n = int.from_bytes(bs.read(1), byteorder="little")
         return bytes.hex(bs.read(n))
 
+    def item_id(self, bs: io.BytesIO):
+        n = int.from_bytes(bs.read(1), byteorder="little")
+        return bytes.hex(bs.read(n))
+
+    def status(self, bs: io.BytesIO):
+        return bytes.hex(bs.read(1))
+
     def nonce(self, bs: io.BytesIO):
         return bytes.hex(bs.read(8))
 
@@ -1246,6 +1269,61 @@ def revocationKeyEvent(self, hexParameter: str):
             }
         )
 
+    def ItemCreatedEvent(self, hexParameter: str):
+        bs = io.BytesIO(bytes.fromhex(hexParameter))
+
+        tag_ = int.from_bytes(bs.read(1), byteorder="little")
+
+        item_id_ = self.item_id(bs)
+        metadata_ = self.metadataUrl(bs)
+        initial_status_ = self.status(bs)
+
+        return itemCreatedEvent(
+            **{
+                "tag": tag_,
+                "item_id": item_id_,
+                "metadata": metadata_,
+                "initial_status": initial_status_,
+            }
+        )
+
+    def ItemStatusChangedEvent(self, hexParameter: str):
+        bs = io.BytesIO(bytes.fromhex(hexParameter))
+
+        tag_ = int.from_bytes(bs.read(1), byteorder="little")
+
+        item_id_ = self.item_id(bs)
+        new_status_ = self.status(bs)
+        additional_data_ = self.additionalData(bs)
+
+        return itemStatusChangedEvent(
+            **{
+                "tag": tag_,
+                "item_id": item_id_,
+                "new_status": new_status_,
+                "additional_data": additional_data_,
+            }
+        )
+
+    def process_tnt_log_event(self, hexParameter: str):
+        bs = io.BytesIO(bytes.fromhex(hexParameter))
+
+        tag_ = int.from_bytes(bs.read(1), byteorder="little")
+        if tag_ == 237:
+            try:
+                event = self.ItemCreatedEvent(hexParameter)
+                return tag_, event
+            except:  # noqa: E722
+                return tag_, None
+        elif tag_ == 236:
+            try:
+                event = self.ItemStatusChangedEvent(hexParameter)
+                return tag_, event
+            except:  # noqa: E722
+                return tag_, None
+        else:
+            return tag_, f"Custom even with tag={tag_}."
+
     def process_log_events(self, hexParameter: str):
         bs = io.BytesIO(bytes.fromhex(hexParameter))
 

tests/cis/test_cis.py

@@ -2,24 +2,35 @@
 import sys
 import os
 
-sys.path.append(os.path.dirname("ccdefundamentals"))
+
+sys.path.append(os.path.dirname("ccdexplorer_fundamentals"))
 from ccdexplorer_fundamentals.GRPCClient import GRPCClient
 from rich import print
 from ccdexplorer_fundamentals.enums import NET
+from ccdexplorer_fundamentals.mongodb import MongoDB, Collections
+from ccdexplorer_schema_parser.Schema import Schema
 from ccdexplorer_fundamentals.cis import (
     CIS,
     registerCredentialEvent,
     revokeCredentialEvent,
     transferEvent,
     tokenMetadataEvent,
+    itemCreatedEvent,
+    itemStatusChangedEvent,
 )
+from ccdexplorer_fundamentals.GRPCClient.CCD_Types import CCD_ContractAddress
 
 
 @pytest.fixture
 def grpcclient():
     return GRPCClient()
 
 
+@pytest.fixture
+def mongodb():
+    return MongoDB(tooter=None)
+
+
 def tx_at_index_from(
     tx_index: int,
     block_hash: str,
@@ -151,3 +162,85 @@ def test_checksum_other2(grpcclient: GRPCClient):
     assert tag == 251
     assert result.token_id == "25"
     assert result.metadata.checksum is None
+
+
+def get_schema_from_source(
+    contract_address: CCD_ContractAddress,
+    mongodb: MongoDB,
+    grpcclient: GRPCClient,
+    net: NET,
+):
+    result = mongodb.testnet[Collections.instances].find_one(
+        {"_id": contract_address.to_str()}
+    )
+    module_ref = (
+        result["v1"]["source_module"]
+        if result.get("v1")
+        else result["v0"]["source_module"]
+    )
+    source_module_name = (
+        result["v1"]["name"][5:] if result.get("v1") else result["v0"]["name"][5:]
+    )
+    ms = grpcclient.get_module_source_original_classes(
+        module_ref, "last_final", net=net
+    )
+    schema = Schema(ms.v1.value, 1) if ms.v1 else Schema(ms.v0.value, 0)
+
+    return schema, source_module_name
+
+
+def test_cis_6_create_item(
+    grpcclient: GRPCClient,
+    mongodb: MongoDB,
+):
+    net = NET.TESTNET
+    block_hash = grpcclient.get_finalized_block_at_height(13753259, net).hash
+    tx = tx_at_index_from(0, block_hash, grpcclient, net)
+    # print(tx)
+    contract_address = CCD_ContractAddress(index=8901, subindex=0)
+    schema, source_module_name = get_schema_from_source(
+        contract_address, mongodb, grpcclient, net
+    )
+    ci = CIS(grpcclient, contract_address.index, contract_address.subindex, "", net)
+
+    event = tx.account_transaction.effects.contract_update_issued.effects[0]
+    tag, result = ci.process_tnt_log_event(event.updated.events[0])
+
+    if tag == 237:
+        event_json = schema.event_to_json(
+            source_module_name,
+            bytes.fromhex(event.updated.events[0]),
+        )
+
+        result: itemCreatedEvent
+        result.initial_status = list(
+            event_json["ItemCreated"][0]["initial_status"].keys()
+        )[0]
+    print(result)
+
+
+def test_cis_6_status_changed_item(grpcclient: GRPCClient, mongodb: MongoDB):
+    net = NET.TESTNET
+    block_hash = grpcclient.get_finalized_block_at_height(13757068, net).hash
+    tx = tx_at_index_from(0, block_hash, grpcclient, net)
+    # print(tx)
+    contract_address = CCD_ContractAddress(index=8901, subindex=0)
+    schema, source_module_name = get_schema_from_source(
+        contract_address, mongodb, grpcclient, net
+    )
+    ci = CIS(grpcclient, contract_address.index, contract_address.subindex, "", net)
+
+    event = tx.account_transaction.effects.contract_update_issued.effects[0]
+    tag, result = ci.process_tnt_log_event(event.updated.events[0])
+
+    if tag == 236:
+        event_json = schema.event_to_json(
+            source_module_name,
+            bytes.fromhex(event.updated.events[0]),
+        )
+
+        result: itemStatusChangedEvent
+        result.new_status = list(
+            event_json["ItemStatusChanged"][0]["new_status"].keys()
+        )[0]
+    print(result)