Add nft custom color

electrumx/lib/atomicals_blueprint_builder.py

@@ -400,6 +400,29 @@ def calculate_nft_atomicals_splat(cls, nft_atomicals, tx):
             output_colored_map[expected_output_index]['atomicals'][atomical_id] = atomical_summary_info
             expected_output_index_incrementing += 1
         return AtomicalNftOutputBlueprintAssignmentSummary(output_colored_map)
+
+    @classmethod
+    def custom_color_nft_atomicals(cls, nft_atomicals, operations_found_at_inputs, tx):
+      # define op `z` to custom-color nft
+      output_colored_map = {}
+      for atomical_id, atomical_info in sorted(nft_atomicals.items()):
+        for out_idx, txout in enumerate(tx.outputs):
+          compact_atomical_id = location_id_bytes_to_compact(atomical_id)
+          compact_atomical_id_data = operations_found_at_inputs["payload"].get(compact_atomical_id, {})
+          # if payload try to color two or more outputs, it will try to color output 0.
+          if len(compact_atomical_id_data.keys()) > 1:
+            expected_output_index = 0
+          else:
+            # if out_idx not in payload keys, skip
+            if str(out_idx) not in compact_atomical_id_data.keys():
+              continue
+            # if payload value <= 0, this nft will burn
+            if compact_atomical_id_data.get(str(out_idx), 0) <= 0:
+              continue
+            expected_output_index = out_idx
+          output_colored_map[expected_output_index] = output_colored_map.get(expected_output_index) or {'atomicals': {}}
+          output_colored_map[expected_output_index]['atomicals'][atomical_id] = atomical_info
+      return AtomicalNftOutputBlueprintAssignmentSummary(output_colored_map)
 
     @classmethod
     def calculate_output_blueprint_nfts(
@@ -409,13 +432,17 @@ def calculate_output_blueprint_nfts(
             nft_atomicals,
             atomicals_spent_at_inputs,
             operations_found_at_inputs,
-            sort_fifo
+            sort_fifo,
+            is_custom_coloring_activated
     ):
         if not nft_atomicals or len(nft_atomicals) == 0:
             return AtomicalNftOutputBlueprintAssignmentSummary({})
         should_splat_nft_atomicals = is_splat_operation(operations_found_at_inputs)
         if should_splat_nft_atomicals and len(nft_atomicals.keys()) > 0:
             return AtomicalsTransferBlueprintBuilder.calculate_nft_atomicals_splat(nft_atomicals, tx)
+        should_custom_colored_nft_atomicals = is_custom_coloring_activated and is_custom_colored_operation(operations_found_at_inputs)
+        if should_custom_colored_nft_atomicals and len(nft_atomicals.keys()) > 0:
+            return AtomicalsTransferBlueprintBuilder.custom_color_nft_atomicals(nft_atomicals, operations_found_at_inputs, tx)
         else:
             # To sort by fifo for NFTs, we also need to calculate a mapping of the nfts to inputs first
             nft_map = AtomicalsTransferBlueprintBuilder.build_nft_input_idx_to_atomical_map(
@@ -481,6 +508,7 @@ def custom_color_ft_atomicals(cls, ft_atomicals, operations_found_at_inputs, tx)
                     str(expected_output_index),
                     0
                 )
+                # if expected_value <= 0, ft will burn
                 if expected_value <= 0 or remaining_value <= 0:
                     continue
                 # if expected_value > txout.value
@@ -666,7 +694,8 @@ def calculate_output_blueprint(
             nft_atomicals,
             atomicals_spent_at_inputs,
             operations_found_at_inputs,
-            sort_fifo
+            sort_fifo,
+            is_custom_coloring_activated
         )
         ft_blueprint = AtomicalsTransferBlueprintBuilder.calculate_output_blueprint_fts(
             tx,
@@ -957,4 +986,4 @@ def get_fts_burned(self):
         return self.fts_burned
 
     def get_atomical_ids_spent(self):
-        return self.atomical_ids_spent
+        return self.atomical_ids_spent

electrumx/server/block_processor.py

@@ -1722,6 +1722,8 @@ def put_nft_outputs_by_blueprint(self, nft_blueprint, operations_found_at_inputs
                             self.put_op_data(tx_num, tx_hash, "splat")
                         if operations_found_at_inputs["op"] == "y":
                             self.put_op_data(tx_num, tx_hash, "split")
+                        if operations_found_at_inputs["op"] == "z":
+                            self.put_op_data(tx_num, tx_hash, "custom-color")
                     self.put_or_delete_state_updates(operations_found_at_inputs, atomical_id, tx_num, tx_hash, output_idx_le, height, 0, False)
                     self.put_or_delete_state_updates(operations_found_at_inputs, atomical_id, tx_num, tx_hash, output_idx_le, height, 1, False)
                 # Only allow NFTs to be sealed.
@@ -1748,6 +1750,8 @@ def put_ft_outputs_by_blueprint(self, ft_blueprint, operations_found_at_inputs,
                         self.put_op_data(tx_num, tx_hash, "splat")
                     if operations_found_at_inputs["op"] == "y":
                         self.put_op_data(tx_num, tx_hash, "split")
+                    if operations_found_at_inputs["op"] == "z":
+                        self.put_op_data(tx_num, tx_hash, "custom-color")
                     else:
                         self.put_op_data(tx_num, tx_hash, operations_found_at_inputs["op"])
                 self.put_or_delete_event_updates_if_found(operations_found_at_inputs, ft_blueprint.first_atomical_id, tx_num, tx_hash, tx, height)

tests/lib/test_atomicals_blueprint_builder.py

@@ -907,8 +907,6 @@ def test_custom_colored_ft_normal():
     subject_atomical_id = b'\x13Jv:\xb1\xad\x9a\xaf\x8a#[7\xa9s\xc0\xcc\xb2\xca\xe1"\x05Y\xc8s\x87\x11\xcc\x90W\xe2\x88\x88\x00\x00\x00\x00'
     subject_atomical_id1 = b"\xb4'{\x12Z\x90z\xed\xd4\xd6\xaf\x87\xb3\xe43\x93\xd0\xbd?v\xfc\x17Y\x8fmcb2\xa4\xef'\x95\x00\x00\x00\x00"
     tx, tx_hash = coin.DESERIALIZER(raw_tx, 0).read_tx_and_hash()
-    # print(hash_to_hex_str(subject_atomical_id))
-    # print(hash_to_hex_str(subject_atomical_id1))
     operation_found_at_inputs = parse_protocols_operations_from_witness_array(tx, tx_hash, True)
     # z means costom color
     operation_found_at_inputs["op"] = "z"
@@ -980,4 +978,30 @@ def mock_mint_fetcher(self, atomical_id):
     assert(ft_output_blueprint.cleanly_assigned == False)
     assert(len(ft_output_blueprint.outputs) == 2)
     assert(ft_output_blueprint.fts_burned == {})
+    assert(blueprint_builder.get_are_fts_burned() == False)
+
+def test_custom_colored_nft_normal():
+    raw_tx_str = '0100000000010258f654e38dee561d45847f45d856ad8cb2d7eafd574521d10ad28b30f44a9e020000000000ffffffffbf6b35d1973a17fc67188ff19731341dafad28a2aac9371c5286c955a6e16c450000000000ffffffff022202000000000000225120d9b4878e9915c8c37149942b02102ed86e462e47f6749424852dc4af89551f212202000000000000225120d9b4878e9915c8c37149942b02102ed86e462e47f6749424852dc4af89551f210340a4334065f27cb80fbf39bd28e634ca9b4e4d7c9b90ed6d575edd9856a664b4352d62e4af96ad11e8aabde952994d8fcb5dd2233ca54100f42045fe63bee9819c7d20c145f972a018b8c401ffd9181a1299a319aee1d55bf2d3393bcd659f06830a78ac00630461746f6d017a4c4fa17842363738376633396235643266633032306562306638653638636439323566323937303635633563383263383664313735636365316139626561613431313233396930a2613018c8613119015a6821c0c145f972a018b8c401ffd9181a1299a319aee1d55bf2d3393bcd659f06830a7801407e04393dddd9e6f899b581a64d26be40b9c148bf1696d99a962dd5257af023ad651efdd4d850819f5eb44e5c281ffb458d59382032248eecf39eb86d4d5dfcb300000000'
+    raw_tx = bytes.fromhex(raw_tx_str)
+    subject_atomical_id = b'9\x12A\xaa\xbe\xa9\xe1\xccu\xd1\x86,\xc8\xc5ep)_\x92\xcdh\x8e\x0f\xeb \xc0/]\x9b\xf3\x87g\x00\x00\x00\x00'
+    tx, tx_hash = coin.DESERIALIZER(raw_tx, 0).read_tx_and_hash()
+    operation_found_at_inputs = parse_protocols_operations_from_witness_array(tx, tx_hash, True)
+    # try to custom nft to output 1
+    operation_found_at_inputs["payload"]["6787f39b5d2fc020eb0f8e68cd925f297065c5c82c86d175cce1a9beaa411239i0"] = {'1': 546}
+    atomicals_spent_at_inputs = {
+        0: [{'atomical_id': subject_atomical_id, 'location_id': b'not_used', 'data': b'not_used', 'data_value': {'sat_value': 546, 'atomical_value': 546}}]
+    }
+    def mock_mint_fetcher(self, atomical_id):
+        return {
+            'atomical_id': atomical_id,
+            # set for nft
+            'type': 'NFT'
+        }
+    blueprint_builder = AtomicalsTransferBlueprintBuilder(MockLogger(), atomicals_spent_at_inputs, operation_found_at_inputs, tx_hash, tx, mock_mint_fetcher, True, True)
+    nft_output_blueprint = blueprint_builder.get_nft_output_blueprint()
+    assert(len(nft_output_blueprint.outputs) == 1)
+    ft_output_blueprint = blueprint_builder.get_ft_output_blueprint()
+    assert(ft_output_blueprint.cleanly_assigned == True)
+    assert(len(ft_output_blueprint.outputs) == 0)
+    assert(ft_output_blueprint.fts_burned == {})
     assert(blueprint_builder.get_are_fts_burned() == False)