Merge pull request #145 from opentensor/feat/thewhaleking/only-use-v14-decoding-for-events

Only use v14 decoding for events

Author: thewhaleking

async_substrate_interface/async_substrate.py

@@ -281,17 +281,26 @@ async def process_events(self):
                     self.__weight = dispatch_info["weight"]
 
                     if "Module" in dispatch_error:
-                        module_index = dispatch_error["Module"][0]["index"]
-                        error_index = int.from_bytes(
-                            bytes(dispatch_error["Module"][0]["error"]),
-                            byteorder="little",
-                            signed=False,
-                        )
+                        if isinstance(dispatch_error["Module"], tuple):
+                            module_index = dispatch_error["Module"][0]
+                            error_index = dispatch_error["Module"][1]
+                        else:
+                            module_index = dispatch_error["Module"]["index"]
+                            error_index = dispatch_error["Module"]["error"]
 
                         if isinstance(error_index, str):
                             # Actual error index is first u8 in new [u8; 4] format
                             error_index = int(error_index[2:4], 16)
-                        module_error = self.substrate.metadata.get_module_error(
+
+                        if self.block_hash:
+                            runtime = await self.substrate.init_runtime(
+                                block_hash=self.block_hash
+                            )
+                        else:
+                            runtime = await self.substrate.init_runtime(
+                                block_id=self.block_number
+                            )
+                        module_error = runtime.metadata.get_module_error(
                             module_index=module_index, error_index=error_index
                         )
                         self.__error_message = {
@@ -823,6 +832,7 @@ async def initialize(self):
                 if ss58_prefix_constant:
                     self.ss58_format = ss58_prefix_constant.value
                     runtime.ss58_format = ss58_prefix_constant.value
+                    runtime.runtime_config.ss58_format = ss58_prefix_constant.value
         self.initialized = True
         self._initializing = False
 
@@ -999,7 +1009,7 @@ async def decode_scale(
         else:
             if not runtime:
                 runtime = await self.init_runtime(block_hash=block_hash)
-            if runtime.metadata_v15 is not None or force_legacy is True:
+            if runtime.metadata_v15 is not None and force_legacy is False:
                 obj = decode_by_type_string(type_string, runtime.registry, scale_bytes)
                 if self.decode_ss58:
                     try:
@@ -1930,7 +1940,13 @@ def convert_event_data(data):
                     if key == "who":
                         who = ss58_encode(bytes(value[0]), self.ss58_format)
                         attributes["who"] = who
-                    if isinstance(value, dict):
+                    elif key == "from":
+                        who_from = ss58_encode(bytes(value[0]), self.ss58_format)
+                        attributes["from"] = who_from
+                    elif key == "to":
+                        who_to = ss58_encode(bytes(value[0]), self.ss58_format)
+                        attributes["to"] = who_to
+                    elif isinstance(value, dict):
                         # Convert nested single-key dictionaries to their keys as strings
                         for sub_key, sub_value in value.items():
                             if isinstance(sub_value, dict):
@@ -1958,16 +1974,15 @@ def convert_event_data(data):
             block_hash = await self.get_chain_head()
 
         storage_obj = await self.query(
-            module="System", storage_function="Events", block_hash=block_hash
+            module="System",
+            storage_function="Events",
+            block_hash=block_hash,
+            force_legacy_decode=True,
         )
+        # bt-decode Metadata V15 is not ideal for events. Force legacy decoding for this
         if storage_obj:
             for item in list(storage_obj):
-                try:
-                    events.append(convert_event_data(item))
-                except (
-                    AttributeError
-                ):  # indicates this was legacy decoded with scalecodec
-                    events.append(item)
+                events.append(item)
         return events
 
     async def get_metadata(self, block_hash=None) -> MetadataV15:
@@ -2175,6 +2190,7 @@ async def _process_response(
         storage_item: Optional[ScaleType] = None,
         result_handler: Optional[ResultHandler] = None,
         runtime: Optional[Runtime] = None,
+        force_legacy_decode: bool = False,
     ) -> tuple[Any, bool]:
         """
         Processes the RPC call response by decoding it, returning it as is, or setting a handler for subscriptions,
@@ -2187,6 +2203,7 @@ async def _process_response(
             storage_item: The ScaleType object used for decoding ScaleBytes results
             result_handler: the result handler coroutine used for handling longer-running subscriptions
             runtime: Optional Runtime to use for decoding. If not specified, the currently-loaded `self.runtime` is used
+            force_legacy_decode: Whether to force the use of the legacy Metadata V14 decoder
 
         Returns:
              (decoded response, completion)
@@ -2208,7 +2225,9 @@ async def _process_response(
                 q = bytes(query_value)
             else:
                 q = query_value
-            result = await self.decode_scale(value_scale_type, q, runtime=runtime)
+            result = await self.decode_scale(
+                value_scale_type, q, runtime=runtime, force_legacy=force_legacy_decode
+            )
         if asyncio.iscoroutinefunction(result_handler):
             # For multipart responses as a result of subscriptions.
             message, bool_result = await result_handler(result, subscription_id)
@@ -2223,6 +2242,7 @@ async def _make_rpc_request(
         result_handler: Optional[ResultHandler] = None,
         attempt: int = 1,
         runtime: Optional[Runtime] = None,
+        force_legacy_decode: bool = False,
     ) -> RequestManager.RequestResults:
         request_manager = RequestManager(payloads)
 
@@ -2267,6 +2287,7 @@ async def _make_rpc_request(
                                 storage_item,
                                 result_handler,
                                 runtime=runtime,
+                                force_legacy_decode=force_legacy_decode,
                             )
 
                             request_manager.add_response(
@@ -2298,6 +2319,7 @@ async def _make_rpc_request(
                             storage_item,
                             result_handler,
                             attempt + 1,
+                            force_legacy_decode,
                         )
 
         return request_manager.get_results()
@@ -3323,6 +3345,7 @@ async def query(
         subscription_handler=None,
         reuse_block_hash: bool = False,
         runtime: Optional[Runtime] = None,
+        force_legacy_decode: bool = False,
     ) -> Optional[Union["ScaleObj", Any]]:
         """
         Queries substrate. This should only be used when making a single request. For multiple requests,
@@ -3355,6 +3378,7 @@ async def query(
             storage_item,
             result_handler=subscription_handler,
             runtime=runtime,
+            force_legacy_decode=force_legacy_decode,
         )
         result = responses[preprocessed.queryable][0]
         if isinstance(result, (list, tuple, int, float)):

async_substrate_interface/sync_substrate.py

@@ -256,16 +256,17 @@ def process_events(self):
                     self.__weight = dispatch_info["weight"]
 
                     if "Module" in dispatch_error:
-                        module_index = dispatch_error["Module"][0]["index"]
-                        error_index = int.from_bytes(
-                            bytes(dispatch_error["Module"][0]["error"]),
-                            byteorder="little",
-                            signed=False,
-                        )
+                        if isinstance(dispatch_error["Module"], tuple):
+                            module_index = dispatch_error["Module"][0]
+                            error_index = dispatch_error["Module"][1]
+                        else:
+                            module_index = dispatch_error["Module"]["index"]
+                            error_index = dispatch_error["Module"]["error"]
 
                         if isinstance(error_index, str):
                             # Actual error index is first u8 in new [u8; 4] format
                             error_index = int(error_index[2:4], 16)
+
                         module_error = self.substrate.metadata.get_module_error(
                             module_index=module_index, error_index=error_index
                         )
@@ -568,6 +569,7 @@ def initialize(self):
                 if ss58_prefix_constant:
                     self.ss58_format = ss58_prefix_constant.value
                     self.runtime.ss58_format = ss58_prefix_constant.value
+                    self.runtime.runtime_config.ss58_format = ss58_prefix_constant.value
         self.initialized = True
 
     def __exit__(self, exc_type, exc_val, exc_tb):
@@ -679,6 +681,7 @@ def decode_scale(
         type_string: str,
         scale_bytes: bytes,
         return_scale_obj=False,
+        force_legacy: bool = False,
     ) -> Union[ScaleObj, Any]:
         """
         Helper function to decode arbitrary SCALE-bytes (e.g. 0x02000000) according to given RUST type_string
@@ -689,6 +692,7 @@ def decode_scale(
             type_string: the type string of the SCALE object for decoding
             scale_bytes: the bytes representation of the SCALE object to decode
             return_scale_obj: Whether to return the decoded value wrapped in a SCALE-object-like wrapper, or raw.
+            force_legacy: Whether to force the use of the legacy Metadata V14 decoder
 
         Returns:
             Decoded object
@@ -697,7 +701,7 @@ def decode_scale(
             # Decode AccountId bytes to SS58 address
             return ss58_encode(scale_bytes, self.ss58_format)
         else:
-            if self.runtime.metadata_v15 is not None:
+            if self.runtime.metadata_v15 is not None and force_legacy is False:
                 obj = decode_by_type_string(
                     type_string, self.runtime.registry, scale_bytes
                 )
@@ -1631,16 +1635,15 @@ def convert_event_data(data):
             block_hash = self.get_chain_head()
 
         storage_obj = self.query(
-            module="System", storage_function="Events", block_hash=block_hash
+            module="System",
+            storage_function="Events",
+            block_hash=block_hash,
+            force_legacy_decode=True,
         )
+        # bt-decode Metadata V15 is not ideal for events. Force legacy decoding for this
         if storage_obj:
             for item in list(storage_obj):
-                try:
-                    events.append(convert_event_data(item))
-                except (
-                    AttributeError
-                ):  # indicates this was legacy decoded with scalecodec
-                    events.append(item)
+                events.append(item)
         return events
 
     def get_metadata(self, block_hash=None) -> MetadataV15:
@@ -1822,6 +1825,7 @@ def _process_response(
         value_scale_type: Optional[str] = None,
         storage_item: Optional[ScaleType] = None,
         result_handler: Optional[ResultHandler] = None,
+        force_legacy_decode: bool = False,
     ) -> tuple[Any, bool]:
         """
         Processes the RPC call response by decoding it, returning it as is, or setting a handler for subscriptions,
@@ -1833,6 +1837,7 @@ def _process_response(
             value_scale_type: Scale Type string used for decoding ScaleBytes results
             storage_item: The ScaleType object used for decoding ScaleBytes results
             result_handler: the result handler coroutine used for handling longer-running subscriptions
+            force_legacy_decode: Whether to force legacy Metadata V14 decoding of the response
 
         Returns:
              (decoded response, completion)
@@ -1854,7 +1859,9 @@ def _process_response(
                 q = bytes(query_value)
             else:
                 q = query_value
-            result = self.decode_scale(value_scale_type, q)
+            result = self.decode_scale(
+                value_scale_type, q, force_legacy=force_legacy_decode
+            )
         if isinstance(result_handler, Callable):
             # For multipart responses as a result of subscriptions.
             message, bool_result = result_handler(result, subscription_id)
@@ -1868,6 +1875,7 @@ def _make_rpc_request(
         storage_item: Optional[ScaleType] = None,
         result_handler: Optional[ResultHandler] = None,
         attempt: int = 1,
+        force_legacy_decode: bool = False,
     ) -> RequestManager.RequestResults:
         request_manager = RequestManager(payloads)
         _received = {}
@@ -1901,6 +1909,7 @@ def _make_rpc_request(
                         storage_item,
                         result_handler,
                         attempt + 1,
+                        force_legacy_decode,
                     )
             if "id" in response:
                 _received[response["id"]] = response
@@ -1932,6 +1941,7 @@ def _make_rpc_request(
                             value_scale_type,
                             storage_item,
                             result_handler,
+                            force_legacy_decode,
                         )
                         request_manager.add_response(
                             item_id, decoded_response, complete
@@ -2870,6 +2880,7 @@ def query(
         raw_storage_key: Optional[bytes] = None,
         subscription_handler=None,
         reuse_block_hash: bool = False,
+        force_legacy_decode: bool = False,
     ) -> Optional[Union["ScaleObj", Any]]:
         """
         Queries substrate. This should only be used when making a single request. For multiple requests,
@@ -2895,6 +2906,7 @@ def query(
             value_scale_type,
             storage_item,
             result_handler=subscription_handler,
+            force_legacy_decode=force_legacy_decode,
         )
         result = responses[preprocessed.queryable][0]
         if isinstance(result, (list, tuple, int, float)):

tests/integration_tests/test_async_substrate_interface.py

@@ -115,3 +115,20 @@ async def test_fully_exhaust_query_map():
             fully_exhausted_records_count += 1
         assert fully_exhausted_records_count == initial_records_count_fully_exhaust
         assert initial_records_count_fully_exhaust == exhausted_records_count
+
+
+@pytest.mark.asyncio
+async def test_get_events_proper_decoding():
+    # known block/hash pair that has the events we seek to decode
+    block = 5846788
+    block_hash = "0x0a1c45063a59b934bfee827caa25385e60d5ec1fd8566a58b5cc4affc4eec412"
+
+    async with AsyncSubstrateInterface(ARCHIVE_ENTRYPOINT) as substrate:
+        all_events = await substrate.get_events(block_hash=block_hash)
+        event = all_events[1]
+        print(type(event["attributes"]))
+        assert event["attributes"] == (
+            "5G1NjW9YhXLadMWajvTkfcJy6up3yH2q1YzMXDTi6ijanChe",
+            30,
+            "0xa6b4e5c8241d60ece0c25056b19f7d21ae845269fc771ad46bf3e011865129a5",
+        )

tests/integration_tests/test_substrate_interface.py

@@ -68,3 +68,19 @@ def test_ss58_conversion():
             if len(value.value) > 0:
                 for decoded_key in value.value:
                     assert isinstance(decoded_key, str)
+
+
+def test_get_events_proper_decoding():
+    # known block/hash pair that has the events we seek to decode
+    block = 5846788
+    block_hash = "0x0a1c45063a59b934bfee827caa25385e60d5ec1fd8566a58b5cc4affc4eec412"
+
+    with SubstrateInterface(ARCHIVE_ENTRYPOINT) as substrate:
+        all_events = substrate.get_events(block_hash=block_hash)
+        event = all_events[1]
+        print(type(event["attributes"]))
+        assert event["attributes"] == (
+            "5G1NjW9YhXLadMWajvTkfcJy6up3yH2q1YzMXDTi6ijanChe",
+            30,
+            "0xa6b4e5c8241d60ece0c25056b19f7d21ae845269fc771ad46bf3e011865129a5",
+        )