Feat/crowdloans

bittensor_cli/src/__init__.py

@@ -732,6 +732,11 @@ class RootSudoOnly(Enum):
     "LIQUIDITY": {
         "LIQUIDITY_MGMT": "Liquidity Management",
     },
+    "CROWD": {
+        "INITIATOR": "Crowdloan Creation & Management",
+        "PARTICIPANT": "Crowdloan Participation",
+        "INFO": "Crowdloan Information",
+    },
 }
 
 

bittensor_cli/src/bittensor/chain_data.py

@@ -1213,3 +1213,51 @@ def from_dict(cls, d: dict, netuid: int) -> "SimSwapResult":
             tao_fee=Balance.from_rao(d["tao_fee"]).set_unit(0),
             alpha_fee=Balance.from_rao(d["alpha_fee"]).set_unit(netuid),
         )
+
+
+@dataclass
+class CrowdloanData(InfoBase):
+    creator: Optional[str]
+    funds_account: Optional[str]
+    deposit: Balance
+    min_contribution: Balance
+    cap: Balance
+    raised: Balance
+    end: int
+    finalized: bool
+    contributors_count: int
+    target_address: Optional[str]
+    has_call: bool
+    call_details: Optional[dict] = None
+
+    @classmethod
+    def _fix_decoded(cls, decoded: dict[str, Any]) -> "CrowdloanData":
+        creator = (
+            decode_account_id(creator_raw)
+            if (creator_raw := decoded.get("creator"))
+            else None
+        )
+        funds_account = (
+            decode_account_id(funds_raw)
+            if (funds_raw := decoded.get("funds_account"))
+            else None
+        )
+        target_address = (
+            decode_account_id(target_raw)
+            if (target_raw := decoded.get("target_address"))
+            else None
+        )
+        return cls(
+            creator=creator,
+            funds_account=funds_account,
+            deposit=Balance.from_rao(int(decoded["deposit"])),
+            min_contribution=Balance.from_rao(int(decoded["min_contribution"])),
+            cap=Balance.from_rao(int(decoded["cap"])),
+            raised=Balance.from_rao(int(decoded["raised"])),
+            end=int(decoded["end"]),
+            finalized=bool(decoded["finalized"]),
+            contributors_count=int(decoded["contributors_count"]),
+            target_address=target_address,
+            has_call=bool(decoded["call"]),
+            call_details=decoded["call_details"],
+        )

bittensor_cli/src/bittensor/subtensor_interface.py

@@ -14,7 +14,7 @@
 from bittensor_wallet import Wallet
 from bittensor_wallet.bittensor_wallet import Keypair
 from bittensor_wallet.utils import SS58_FORMAT
-from scalecodec import GenericCall
+from scalecodec import GenericCall, ScaleBytes
 import typer
 import websockets
 
@@ -30,6 +30,7 @@
     SubnetState,
     MetagraphInfo,
     SimSwapResult,
+    CrowdloanData,
 )
 from bittensor_cli.src import DelegatesDetails
 from bittensor_cli.src.bittensor.balances import Balance, fixed_to_float
@@ -167,6 +168,44 @@ async def query(
         else:
             return result
 
+    async def _decode_inline_call(
+        self,
+        call_option: Any,
+        block_hash: Optional[str] = None,
+    ) -> Optional[dict[str, Any]]:
+        """
+        Decode an `Option<BoundedCall>` returned from storage into a structured dictionary.
+        """
+        if not call_option or "Inline" not in call_option:
+            return None
+        inline_bytes = bytes(call_option["Inline"][0][0])
+        call_obj = await self.substrate.create_scale_object(
+            "Call",
+            data=ScaleBytes(inline_bytes),
+            block_hash=block_hash,
+        )
+        call_value = call_obj.decode()
+
+        if not isinstance(call_value, dict):
+            return None
+
+        call_args = call_value.get("call_args") or []
+        args_map: dict[str, dict[str, Any]] = {}
+        for arg in call_args:
+            if isinstance(arg, dict) and arg.get("name"):
+                args_map[arg["name"]] = {
+                    "type": arg.get("type"),
+                    "value": arg.get("value"),
+                }
+
+        return {
+            "call_index": call_value.get("call_index"),
+            "pallet": call_value.get("call_module"),
+            "method": call_value.get("call_function"),
+            "args": args_map,
+            "hash": call_value.get("call_hash"),
+        }
+
     async def get_all_subnet_netuids(
         self, block_hash: Optional[str] = None
     ) -> list[int]:
@@ -1693,6 +1732,101 @@ async def get_scheduled_coldkey_swap(
             keys_pending_swap.append(decode_account_id(ss58))
         return keys_pending_swap
 
+    async def get_crowdloans(
+        self, block_hash: Optional[str] = None
+    ) -> list[CrowdloanData]:
+        """Retrieves all crowdloans from the network.
+
+        Args:
+            block_hash (Optional[str]): The blockchain block hash at which to perform the query.
+
+        Returns:
+            dict[int, CrowdloanData]: A dictionary mapping crowdloan IDs to CrowdloanData objects
+                containing details such as creator, deposit, cap, raised amount, and finalization status.
+
+        This function fetches information about all crowdloans
+        """
+        crowdloans_data = await self.substrate.query_map(
+            module="Crowdloan",
+            storage_function="Crowdloans",
+            block_hash=block_hash,
+            fully_exhaust=True,
+        )
+        crowdloans = {}
+        async for fund_id, fund_info in crowdloans_data:
+            decoded_call = await self._decode_inline_call(
+                fund_info["call"],
+                block_hash=block_hash,
+            )
+            info_dict = dict(fund_info.value)
+            info_dict["call_details"] = decoded_call
+            crowdloans[fund_id] = CrowdloanData.from_any(info_dict)
+
+        return crowdloans
+
+    async def get_single_crowdloan(
+        self,
+        crowdloan_id: int,
+        block_hash: Optional[str] = None,
+    ) -> Optional[CrowdloanData]:
+        """Retrieves detailed information about a specific crowdloan.
+
+        Args:
+            crowdloan_id (int): The unique identifier of the crowdloan to retrieve.
+            block_hash (Optional[str]): The blockchain block hash at which to perform the query.
+
+        Returns:
+            Optional[CrowdloanData]: A CrowdloanData object containing the crowdloan's details if found,
+                None if the crowdloan does not exist.
+
+        The returned data includes crowdloan details such as funding targets,
+        contribution minimums, timeline, and current funding status
+        """
+        crowdloan_info = await self.query(
+            module="Crowdloan",
+            storage_function="Crowdloans",
+            params=[crowdloan_id],
+            block_hash=block_hash,
+        )
+        if crowdloan_info:
+            decoded_call = await self._decode_inline_call(
+                crowdloan_info.get("call"),
+                block_hash=block_hash,
+            )
+            crowdloan_info["call_details"] = decoded_call
+            return CrowdloanData.from_any(crowdloan_info)
+        return None
+
+    async def get_crowdloan_contribution(
+        self,
+        crowdloan_id: int,
+        contributor: str,
+        block_hash: Optional[str] = None,
+    ) -> Optional[Balance]:
+        """Retrieves a user's contribution to a specific crowdloan.
+
+        Args:
+            crowdloan_id (int): The ID of the crowdloan.
+            contributor (str): The SS58 address of the contributor.
+            block_hash (Optional[str]): The blockchain block hash at which to perform the query.
+
+        Returns:
+            Optional[Balance]: The contribution amount as a Balance object if found, None otherwise.
+
+        This function queries the Contributions storage to find the amount a specific address
+        has contributed to a given crowdloan.
+        """
+        contribution = await self.query(
+            module="Crowdloan",
+            storage_function="Contributions",
+            params=[crowdloan_id, contributor],
+            block_hash=block_hash,
+        )
+
+        if contribution:
+            return Balance.from_rao(contribution)
+        return None
+
     async def get_coldkey_swap_schedule_duration(
         self,
         block_hash: Optional[str] = None,

bittensor_cli/src/bittensor/utils.py

@@ -1504,11 +1504,11 @@ async def print_extrinsic_id(
         query = await substrate.rpc_request("system_chainType", [])
         if query.get("result") == "Live":
             console.print(
-                f":white_heavy_check_mark:Your extrinsic has been included as {ext_id}: "
+                f":white_heavy_check_mark: Your extrinsic has been included as {ext_id}: "
                 f"[blue]https://tao.app/extrinsic/{ext_id}[/blue]"
             )
             return
     console.print(
-        f":white_heavy_check_mark:Your extrinsic has been included as {ext_id}"
+        f":white_heavy_check_mark: Your extrinsic has been included as {ext_id}"
     )
     return