Cleanup UncurriedNFT

chia/wallet/nft_wallet/nft_puzzles.py

@@ -131,7 +131,7 @@ def get_nft_info_from_puzzle(puzzle: Program, nft_coin: Coin) -> NFTInfo:
     :return: NFTInfo
     """
     # TODO Update this method after the NFT code finalized
-    uncurried_nft: UncurriedNFT = UncurriedNFT.uncurry(puzzle, True)
+    uncurried_nft: UncurriedNFT = UncurriedNFT.uncurry(puzzle)
     data_uris = []
     for uri in uncurried_nft.data_uris.as_python():
         data_uris.append(str(uri, "utf-8"))

chia/wallet/nft_wallet/nft_wallet.py

@@ -185,66 +185,11 @@ async def puzzle_solution_received(self, coin_spend: CoinSpend, in_transaction:
         coin_name = coin_spend.coin.name()
         puzzle: Program = Program.from_bytes(bytes(coin_spend.puzzle_reveal))
         solution: Program = Program.from_bytes(bytes(coin_spend.solution)).rest().rest().first()
-        uncurried_nft: UncurriedNFT = UncurriedNFT.uncurry(puzzle)
         nft_transfer_program = None
-        if uncurried_nft.matched:
-            # check if we already know this hash, if not then try to find reveal in solution
-            for hash, reveal in self.nft_wallet_info.known_transfer_programs:
-                if hash == bytes32(uncurried_nft.transfer_program_hash.as_atom()):
-                    nft_transfer_program = reveal
-            if nft_transfer_program is None:
-                attempt = nft_puzzles.get_transfer_program_from_inner_solution(solution)
-                if attempt is not None:
-                    nft_transfer_program = attempt
-                    await self.add_transfer_program(nft_transfer_program, in_transaction=in_transaction)
-
-            assert nft_transfer_program is not None
-            self.log.info(f"found the info for coin {coin_name}")
-            parent_coin = None
-            coin_record = await self.wallet_state_manager.coin_store.get_coin_record(coin_name)
-            if coin_record is None:
-                coin_states: Optional[List[CoinState]] = await self.wallet_state_manager.wallet_node.get_coin_state(
-                    [coin_name]
-                )
-                if coin_states is not None:
-                    parent_coin = coin_states[0].coin
-            if coin_record is not None:
-                parent_coin = coin_record.coin
-            if parent_coin is None:
-                raise ValueError("Error in finding parent")
-            inner_puzzle: Program = nft_puzzles.create_nft_layer_puzzle_with_curry_params(
-                uncurried_nft.singleton_launcher_id.as_atom(),
-                uncurried_nft.owner_did.as_atom(),
-                uncurried_nft.transfer_program_hash.as_atom(),
-                uncurried_nft.metadata,
-                uncurried_nft.transfer_program_curry_params,
-            )
-            child_coin: Optional[Coin] = None
-            for new_coin in coin_spend.additions():
-                if new_coin.amount % 2 == 1:
-                    child_coin = new_coin
-                    break
-            assert child_coin is not None
-
-            metadata = nft_puzzles.update_metadata(uncurried_nft.metadata, solution)
-            # TODO: add smarter check for -22 to see if curry_params changed and use this for metadata too
-            child_puzzle: Program = nft_puzzles.create_full_puzzle_with_curry_params(
-                uncurried_nft.singleton_launcher_id.as_atom(),
-                self.nft_wallet_info.my_did,
-                uncurried_nft.transfer_program_hash.as_atom(),
-                metadata,
-                uncurried_nft.transfer_program_curry_params,
-            )
 
-            assert child_puzzle.get_tree_hash() == child_coin.puzzle_hash
-            await self.add_coin(
-                child_coin,
-                LineageProof(parent_coin.parent_coin_info, inner_puzzle.get_tree_hash(), parent_coin.amount),
-                nft_transfer_program,
-                child_puzzle,
-                in_transaction=in_transaction,
-            )
-        else:
+        try:
+            uncurried_nft = UncurriedNFT.uncurry(puzzle)
+        except Exception:
             # The parent is not an NFT which means we need to scrub all of its children from our DB
             child_coin_records = await self.wallet_state_manager.coin_store.get_coin_records_by_parent_id(coin_name)
             if len(child_coin_records) > 0:
@@ -254,6 +199,64 @@ async def puzzle_solution_received(self, coin_spend: CoinSpend, in_transaction:
                         # await self.remove_lineage(record.coin.name())
                         # We also need to make sure there's no record of the transaction
                         await self.wallet_state_manager.tx_store.delete_transaction_record(record.coin.name())
+            return
+
+        # check if we already know this hash, if not then try to find reveal in solution
+        for hash, reveal in self.nft_wallet_info.known_transfer_programs:
+            if hash == bytes32(uncurried_nft.transfer_program_hash.as_atom()):
+                nft_transfer_program = reveal
+        if nft_transfer_program is None:
+            attempt = nft_puzzles.get_transfer_program_from_inner_solution(solution)
+            if attempt is not None:
+                nft_transfer_program = attempt
+                await self.add_transfer_program(nft_transfer_program, in_transaction=in_transaction)
+
+        assert nft_transfer_program is not None
+        self.log.info(f"found the info for coin {coin_name}")
+        parent_coin = None
+        coin_record = await self.wallet_state_manager.coin_store.get_coin_record(coin_name)
+        if coin_record is None:
+            coin_states: Optional[List[CoinState]] = await self.wallet_state_manager.wallet_node.get_coin_state(
+                [coin_name]
+            )
+            if coin_states is not None:
+                parent_coin = coin_states[0].coin
+        if coin_record is not None:
+            parent_coin = coin_record.coin
+        if parent_coin is None:
+            raise ValueError("Error in finding parent")
+        inner_puzzle: Program = nft_puzzles.create_nft_layer_puzzle_with_curry_params(
+            uncurried_nft.singleton_launcher_id.as_atom(),
+            uncurried_nft.owner_did.as_atom(),
+            uncurried_nft.transfer_program_hash.as_atom(),
+            uncurried_nft.metadata,
+            uncurried_nft.transfer_program_curry_params,
+        )
+        child_coin: Optional[Coin] = None
+        for new_coin in coin_spend.additions():
+            if new_coin.amount % 2 == 1:
+                child_coin = new_coin
+                break
+        assert child_coin is not None
+
+        metadata = nft_puzzles.update_metadata(uncurried_nft.metadata, solution)
+        # TODO: add smarter check for -22 to see if curry_params changed and use this for metadata too
+        child_puzzle: Program = nft_puzzles.create_full_puzzle_with_curry_params(
+            uncurried_nft.singleton_launcher_id.as_atom(),
+            self.nft_wallet_info.my_did,
+            uncurried_nft.transfer_program_hash.as_atom(),
+            metadata,
+            uncurried_nft.transfer_program_curry_params,
+        )
+
+        assert child_puzzle.get_tree_hash() == child_coin.puzzle_hash
+        await self.add_coin(
+            child_coin,
+            LineageProof(parent_coin.parent_coin_info, inner_puzzle.get_tree_hash(), parent_coin.amount),
+            nft_transfer_program,
+            child_puzzle,
+            in_transaction=in_transaction,
+        )
 
     async def add_coin(
         self, coin: Coin, lineage_proof: LineageProof, transfer_program: Program, puzzle: Program, in_transaction: bool
@@ -515,13 +518,16 @@ async def receive_nft(self, sending_sb: SpendBundle, fee: uint64 = uint64(0)) ->
         trade_price_list_discovered = None
         nft_id = None
         for coin_spend in sending_sb.coin_spends:
-            uncurried_nft: UncurriedNFT = UncurriedNFT.uncurry(Program.from_bytes(bytes(coin_spend.puzzle_reveal)))
-            if uncurried_nft.matched:
-                inner_sol = Program.from_bytes(bytes(coin_spend.solution)).rest().rest().first()
-                trade_price_list_discovered = nft_puzzles.get_trade_prices_list_from_inner_solution(inner_sol)
-                nft_id = uncurried_nft.singleton_launcher_id.as_atom()
-                royalty_address: bytes32 = uncurried_nft.royalty_address.as_atom()
-                royalty_percentage: uint64 = uint64(uncurried_nft.trade_price_percentage.as_int())
+            try:
+                uncurried_nft = UncurriedNFT.uncurry(Program.from_bytes(bytes(coin_spend.puzzle_reveal)))
+            except Exception:
+                continue
+
+            inner_sol = Program.from_bytes(bytes(coin_spend.solution)).rest().rest().first()
+            trade_price_list_discovered = nft_puzzles.get_trade_prices_list_from_inner_solution(inner_sol)
+            nft_id = uncurried_nft.singleton_launcher_id.as_atom()
+            royalty_address: bytes32 = uncurried_nft.royalty_address.as_atom()
+            royalty_percentage: uint64 = uint64(uncurried_nft.trade_price_percentage.as_int())
 
         assert trade_price_list_discovered is not None
         assert nft_id is not None

chia/wallet/nft_wallet/uncurry_nft.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass
+from typing import Type, TypeVar
 
 from chia.types.blockchain_format.program import Program
 from chia.wallet.puzzles.load_clvm import load_clvm
@@ -9,6 +10,9 @@
 NFT_MOD = load_clvm("nft_innerpuz.clvm")
 
 
+_T_UncurriedNFT = TypeVar("_T_UncurriedNFT", bound="UncurriedNFT")
+
+
 @dataclass
 class UncurriedNFT:
     """
@@ -17,92 +21,120 @@ class UncurriedNFT:
     This is the only place you need to change after modified the Chialisp curried parameters.
     """
 
-    matched: bool = field(init=False)
-    """If the puzzle is a NFT puzzle"""
-
-    nft_mod_hash: Program = field(init=False)
+    nft_mod_hash: Program
     """NFT module hash"""
 
-    singleton_struct: Program = field(init=False)
+    singleton_struct: Program
     """
     Singleton struct
     [singleton_mod_hash, singleton_launcher_id, launcher_puzhash]
     """
-    singleton_mod_hash: Program = field(init=False)
-    singleton_launcher_id: Program = field(init=False)
-    launcher_puzhash: Program = field(init=False)
+    singleton_mod_hash: Program
+    singleton_launcher_id: Program
+    launcher_puzhash: Program
 
-    owner_did: Program = field(init=False)
+    owner_did: Program
     """Owner's DID"""
 
-    transfer_program_hash: Program = field(init=False)
+    transfer_program_hash: Program
     """Puzzle hash of the transfer program"""
 
-    transfer_program_curry_params: Program = field(init=False)
+    transfer_program_curry_params: Program
     """
     Curried parameters of the transfer program
     [royalty_address, trade_price_percentage, settlement_mod_hash, cat_mod_hash]
     """
-    royalty_address: Program = field(init=False)
-    trade_price_percentage: Program = field(init=False)
-    settlement_mod_hash: Program = field(init=False)
-    cat_mod_hash: Program = field(init=False)
+    royalty_address: Program
+    trade_price_percentage: Program
+    settlement_mod_hash: Program
+    cat_mod_hash: Program
 
-    metadata: Program = field(init=False)
+    metadata: Program
     """
     NFT metadata
     [("u", data_uris), ("h", data_hash)]
     """
-    data_uris: Program = field(init=False)
-    data_hash: Program = field(init=False)
+    data_uris: Program
+    data_hash: Program
+
+    @classmethod
+    def uncurry(cls: Type[_T_UncurriedNFT], puzzle: Program) -> _T_UncurriedNFT:
+        """Try to uncurry a NFT puzzle
 
-    @staticmethod
-    def uncurry(puzzle: Program, raise_exception: bool = False) -> UncurriedNFT:
-        """
-        Try to uncurry a NFT puzzle
         :param puzzle: Puzzle
         :param raise_exception: If want to raise an exception when the puzzle is invalid
         :return Uncurried NFT
         """
-        nft = UncurriedNFT()
+        mod, curried_args = puzzle.uncurry()
+        if mod != SINGLETON_TOP_LAYER_MOD:
+            # TODO: proper message
+            raise ValueError(f"Cannot uncurry puzzle, mod not a single top layer: {puzzle}")
+
+        mod, curried_args = curried_args.rest().first().uncurry()
+        if mod != NFT_MOD:
+            # TODO: proper message
+            raise ValueError(f"Cannot uncurry puzzle, mod not an NFT mode: {puzzle}")
+
+        # nft parameters
+        # TODO: Centralize the definition of this order with a class and construct
+        #       an instance of it instead of using free variables.
         try:
-            mod, curried_args = puzzle.uncurry()
-            if mod == SINGLETON_TOP_LAYER_MOD:
-                mod, curried_args = curried_args.rest().first().uncurry()
-                if mod == NFT_MOD:
-                    nft.matched = True
-                    # Set nft parameters
-                    (
-                        nft.nft_mod_hash,
-                        nft.singleton_struct,
-                        nft.owner_did,
-                        nft.transfer_program_hash,
-                        nft.transfer_program_curry_params,
-                        nft.metadata,
-                    ) = curried_args.as_iter()
-                    # Set singleton
-                    nft.singleton_mod_hash = nft.singleton_struct.first()
-                    nft.singleton_launcher_id = nft.singleton_struct.rest().first()
-                    nft.launcher_puzhash = nft.singleton_struct.rest().rest()
-                    # Set transfer program parameters
-                    (
-                        nft.royalty_address,
-                        nft.trade_price_percentage,
-                        nft.settlement_mod_hash,
-                        nft.cat_mod_hash,
-                    ) = nft.transfer_program_curry_params.as_iter()
-                    # Set metadata
-                    for kv_pair in nft.metadata.as_iter():
-                        if kv_pair.first().as_atom() == b"u":
-                            nft.data_uris = kv_pair.rest()
-                        if kv_pair.first().as_atom() == b"h":
-                            nft.data_hash = kv_pair.rest()
-                    return nft
-            nft.matched = False
-            return nft
-        except Exception:
-            if raise_exception:
-                raise ValueError(f"Cannot uncurry puzzle {puzzle}, it's not a NFT puzzle.")
-            else:
-                nft.matched = False
-                return nft
+            (
+                nft_mod_hash,
+                singleton_struct,
+                owner_did,
+                transfer_program_hash,
+                transfer_program_curry_params,
+                metadata,
+            ) = curried_args.as_iter()
+        except ValueError as e:
+            # TODO: proper message
+            raise ValueError(f"Cannot uncurry puzzle, incorrect number of arguemnts: {puzzle}") from e
+
+        # singleton
+        singleton_mod_hash = singleton_struct.first()
+        singleton_launcher_id = singleton_struct.rest().first()
+        launcher_puzhash = singleton_struct.rest().rest()
+
+        # transfer program parameters
+        try:
+            (
+                royalty_address,
+                trade_price_percentage,
+                settlement_mod_hash,
+                cat_mod_hash,
+            ) = transfer_program_curry_params.as_iter()
+        except ValueError as e:
+            # TODO: proper message
+            raise ValueError(
+                f"Cannot uncurry puzzle, incorrect number of transfer program curry parameters: {puzzle}",
+            ) from e
+
+        # metadata
+        for kv_pair in metadata.as_iter():
+            if kv_pair.first().as_atom() == b"u":
+                data_uris = kv_pair.rest()
+            if kv_pair.first().as_atom() == b"h":
+                data_hash = kv_pair.rest()
+
+        # TODO: How should we handle the case that no data_uris or no data_hash
+        #       is found?  Is it ok if we find them multiple times and just
+        #       ignore the first ones?
+
+        return cls(
+            nft_mod_hash=nft_mod_hash,
+            singleton_struct=singleton_struct,
+            singleton_mod_hash=singleton_mod_hash,
+            singleton_launcher_id=singleton_launcher_id,
+            launcher_puzhash=launcher_puzhash,
+            owner_did=owner_did,
+            transfer_program_hash=transfer_program_hash,
+            transfer_program_curry_params=transfer_program_curry_params,
+            royalty_address=royalty_address,
+            trade_price_percentage=trade_price_percentage,
+            settlement_mod_hash=settlement_mod_hash,
+            cat_mod_hash=cat_mod_hash,
+            metadata=metadata,
+            data_uris=data_uris,
+            data_hash=data_hash,
+        )

chia/wallet/wallet_state_manager.py

@@ -560,8 +560,11 @@ async def determine_coin_type(
         # Check if the coin is a NFT
         #                                                        hint
         # First spend where 1 mojo coin -> Singleton launcher -> NFT -> NFT
-        uncurried_nft: UncurriedNFT = UncurriedNFT.uncurry(Program.from_bytes(bytes(coin_spend.puzzle_reveal)))
-        if uncurried_nft.matched:
+        try:
+            UncurriedNFT.uncurry(Program.from_bytes(bytes(coin_spend.puzzle_reveal)))
+        except Exception:
+            pass
+        else:
             return await self.handle_nft(coin_spend)
 
         # Check if the coin is a DID