Merge pull request #1 from officialnico/nico/world_interactions

Created World and Player classes

Author: officialnico

examples/cafecosmos-contracts

@@ -0,0 +1 @@
+Subproject commit f02134dddda929c655c02e3b6c46e7fd1503d52b

examples/env.example

@@ -0,0 +1,2 @@
+PLAYER1=PRIVATE_KEY1
+PLAYER2=PRIVATE_KEY2

examples/mud.config.ts

@@ -1,3 +1,4 @@
+from typing import Any, Dict, List, TypedDict, Type
 import re
 import requests
 
@@ -23,7 +24,7 @@ def parse_mud_config(file_path: str):
 
 
 class BaseTable:
-    RESERVED_SQL_KEYWORDS = {"exists", "from", "values", "limit", "index"}  # Add more keywords if needed
+    RESERVED_SQL_KEYWORDS = {"exists", "from", "values", "limit", "index"}
 
     def __init__(self, sdk, table_name, schema, keys):
         self.sdk = sdk
@@ -46,25 +47,10 @@ def get(self, limit=1000, **filters):
         Returns:
             List[Dict[str, Any]]: A list of records from the table.
         """
-        # Determine the columns to select based on whether the keys are filtered
-        if any(key in filters for key in self.keys):
-            # Exclude keys from SELECT if they are filtered
-            select_columns = ", ".join(
-                self._escape_column_name(col)
-                for col in self.schema.keys()
-                if col not in self.keys
-            )
-        else:
-            # Include all columns if no keys are filtered
-            select_columns = ", ".join(
-                self._escape_column_name(col) for col in self.schema.keys()
-            )
-
-        # Construct the WHERE clause
+        select_columns = ", ".join(self.schema.keys())
         where_clause = " AND ".join(
             f"{self._escape_column_name(key)}={repr(value)}" for key, value in filters.items()
         )
-
         query = f"SELECT {select_columns} FROM {self.table_name}"
         if where_clause:
             query += f" WHERE {where_clause}"
@@ -77,24 +63,40 @@ def get(self, limit=1000, **filters):
     def _parse_response(self, response):
         if "result" not in response or not response["result"]:
             return None  # Return None if there are no results
-
         results = response["result"][0]
         if not results:  # Check if the results array is empty
             return None
-
         headers, *rows = results
         return [dict(zip(headers, row)) for row in rows]
 
 
 class TableRegistry:
     def __init__(self, sdk):
         self.sdk = sdk
+        self.SOLIDITY_TO_PYTHON_TYPE = self._generate_solidity_to_python_type_map()
 
-    def register_table(self, table_name, schema, keys):
+    @staticmethod
+    def _generate_solidity_to_python_type_map():
         """
-        Dynamically create and register a table as an attribute of the registry.
+        Generate a mapping of all Solidity integer types to Python int.
         """
-        table_class = type(table_name, (BaseTable,), {})
+        solidity_types = {}
+        for bits in range(8, 257, 8):
+            solidity_types[f"int{bits}"] = int
+            solidity_types[f"uint{bits}"] = int
+        solidity_types.update({"bool": bool, "address": str, "string": str, "bytes32": bytes, "bytes": bytes})
+        return solidity_types
+
+    def register_table(self, table_name, schema, keys):
+        schema_typed_dict = TypedDict(
+            f"{table_name}Schema",
+            {k: self.SOLIDITY_TO_PYTHON_TYPE.get(v, Any) for k, v in schema.items()}
+        )
+
+        def get(self, limit: int = 1000, **filters: schema_typed_dict) -> List[schema_typed_dict]:
+            return super(type(self), self).get(limit=limit, **filters)
+
+        table_class = type(table_name, (BaseTable,), {"get": get})
         table_instance = table_class(self.sdk, table_name, schema, keys)
         setattr(self, table_name, table_instance)
 
@@ -104,8 +106,6 @@ def __init__(self, indexer_url, world_address, mud_config_path):
         self.indexer_url = indexer_url
         self.world_address = world_address
         self.tables = TableRegistry(self)
-
-        # Parse and register tables from the configuration file
         self._parsed_tables = parse_mud_config(mud_config_path)
         for table_name, table_info in self._parsed_tables.items():
             self.tables.register_table(table_name, table_info["schema"], table_info["key"])
@@ -117,6 +117,4 @@ def post(self, payload):
         return response.json()
 
     def get_table_names(self):
-        """Return a list of all table names from the parsed configuration."""
         return list(self._parsed_tables.keys())
-    

examples/player.ipynb

@@ -0,0 +1,58 @@
+from .World import World
+from web3 import Web3
+from dotenv import load_dotenv
+import os
+
+# Load environment variables from a .env file
+load_dotenv()
+
+class Player:
+    def __init__(self, private_key: str = None, env_key_name: str = None):
+        """
+        Initialize the Player instance.
+        Either `private_key` or `env_key_name` must be provided.
+
+        Args:
+            private_key (str): The private key for the player (optional).
+            env_key_name (str): The name of the environment variable holding the private key (optional).
+
+        Raises:
+            ValueError: If neither `private_key` nor `env_key_name` is provided.
+            ValueError: If `env_key_name` is provided but the environment variable is not set.
+        """
+        if private_key:
+            self.private_key = private_key
+        elif env_key_name:
+            self.private_key = os.getenv(env_key_name)
+            if not self.private_key:
+                raise ValueError(
+                    f"Environment variable '{env_key_name}' is not set or contains an invalid value. "
+                    "Please set it in your .env file or provide a private key directly."
+                )
+        else:
+            raise ValueError(
+                "Initialization failed: You must provide either a `private_key` or an `env_key_name`. "
+                "For example: Player(private_key='0xYourPrivateKey') or Player(env_key_name='PLAYER1')."
+            )
+
+        self.private_key = self.private_key if self.private_key.startswith('0x') else '0x' + self.private_key
+        self.player_address = self._derive_address(self.private_key)
+        self.worlds = {}  # Dictionary to manage multiple worlds
+
+    def add_world(self, world: World, world_name: str):
+        """
+        Add a world to the player and assign it a dynamic name.
+
+        Args:
+            world (World): An instance of the World class.
+            world_name (str): The name to assign to the world for dynamic access.
+        """
+        if not isinstance(world, World):
+            raise TypeError("The `world` parameter must be an instance of the World class.")
+        self.worlds[world_name] = world
+        setattr(self, world_name, world)
+
+    def _derive_address(self, private_key):
+        """Derive the Ethereum address from the private key."""
+        account = Web3().eth.account.from_key(private_key)
+        return account.address

examples/querying_tables.ipynb

@@ -0,0 +1,132 @@
+from web3 import Web3
+import json
+from pathlib import Path
+from .MUDIndexerSDK import MUDIndexerSDK
+
+def find_abi_files(root_dir):
+    """Recursively find all ABI files"""
+    abi_files = []
+    root_path = Path(root_dir)
+    abi_patterns = ["*.abi.json", "*.json"]
+    for pattern in abi_patterns:
+        abi_files.extend(root_path.rglob(pattern))
+    return abi_files
+
+def load_abis(root_dir) -> dict:
+    """Load all ABI files from directory structure"""
+    abis = {}
+    for abi_file in find_abi_files(root_dir):
+        try:
+            with open(abi_file, 'r') as f:
+                abi_data = json.load(f)
+                
+            if isinstance(abi_data, dict):
+                if 'abi' in abi_data:
+                    abi_data = abi_data['abi']
+                elif 'contracts' in abi_data:
+                    for contract_name, contract_data in abi_data['contracts'].items():
+                        if 'abi' in contract_data:
+                            abis[contract_name] = contract_data['abi']
+                    continue
+                
+            contract_name = abi_file.parent.name if abi_file.name == 'abi.json' else abi_file.stem.replace('.abi', '')
+            abis[contract_name] = abi_data
+                
+        except Exception as e:
+            print(f"Error processing {abi_file}: {e}")
+    
+    return abis
+
+class World:
+    def __init__(self, rpc, world_address, abis_dir, indexer_url=None, mud_config_path=None):
+        """
+        Initialize the World instance.
+
+        Args:
+            rpc (str): RPC endpoint URL.
+            world_address (str): The address of the World contract.
+            abis_dir (str): Directory containing ABI files.
+            indexer_url (str, optional): URL for the indexer. If provided, initializes the indexer.
+            mud_config_path (str, optional): Path to the mud.config.ts file. Required if indexer_url is provided.
+        """
+        self.w3 = Web3(Web3.HTTPProvider(rpc))
+        self.chain_id = self.w3.eth.chain_id  # Automatically fetch the chain ID
+        self.abis = load_abis(abis_dir)
+        self.indexer = None
+
+        # Initialize the contract
+        if "IWorld" in self.abis:
+            self.contract = self.w3.eth.contract(address=world_address, abi=self.abis["IWorld"])
+            self.errors = self._extract_all_errors()
+
+            for func_name in dir(self.contract.functions):
+                if not func_name.startswith("_"):
+                    original_function = getattr(self.contract.functions, func_name)
+                    setattr(self, func_name, self._wrap_function(original_function, func_name))
+        else:
+            raise Exception("IWorld ABI not found")
+
+        # Automatically set up the indexer if parameters are provided
+        if indexer_url and mud_config_path:
+            self._initialize_indexer(indexer_url, world_address, mud_config_path)
+
+    def _initialize_indexer(self, indexer_url, world_address, mud_config_path):
+        """
+        Initialize and set the indexer.
+
+        Args:
+            indexer_url (str): URL for the indexer.
+            world_address (str): The address of the World contract.
+            mud_config_path (str): Path to the mud.config.ts file.
+        """
+        from mud import MUDIndexerSDK  # Assuming MUDIndexerSDK is part of your mud package
+
+        # Create the indexer
+        indexer = MUDIndexerSDK(indexer_url, world_address, mud_config_path)
+        self.set_indexer(indexer)
+
+    def set_indexer(self, indexer):
+        """
+        Set the indexer instance and expose its tables.
+
+        Args:
+            indexer (MUDIndexerSDK): The indexer instance.
+        """
+        self.indexer = indexer
+
+        # Expose tables as attributes directly under world.indexer
+        for table_name in indexer.get_table_names():
+            table_instance = getattr(indexer.tables, table_name)
+            setattr(self.indexer, table_name, table_instance)
+
+    def _extract_all_errors(self):
+        errors = {}
+        for contract_name, abi in self.abis.items():
+            if not isinstance(abi, list):
+                continue
+
+            for item in abi:
+                if item.get('type') == 'error':
+                    signature = f"{item['name']}({','.join(inp['type'] for inp in item.get('inputs', []))})"
+                    selector = self.w3.keccak(text=signature)[:4].hex()
+                    errors[selector] = (contract_name, item['name'])
+        return errors
+
+    def _wrap_function(self, contract_function, func_name):
+        def wrapped_function(*args, **kwargs):
+            try:
+                return contract_function(*args, **kwargs).call()
+            except Exception as e:
+                error_str = str(e)
+                if '0x' in error_str:
+                    import re
+                    hex_match = re.search(r'0x[a-fA-F0-9]+', error_str)
+                    if hex_match:
+                        selector = hex_match.group(0)[2:10]
+                        if selector in self.errors:
+                            contract, error = self.errors[selector]
+                            error_msg = f"{error} when calling {func_name}"
+                            new_error = type(e)((error_msg,))
+                            raise new_error from None
+                raise e
+        return wrapped_function

mud/MUDIndexerSDK.py

@@ -1,3 +1,5 @@
+from .World import World
+from .Player import Player
 from .MUDIndexerSDK import MUDIndexerSDK
 
-__all__ = ["MUDIndexerSDK"]  # Optional, restricts what is accessible via `from mud import *`
+__all__ = ["World", "Player", "MUDIndexerSDK"]