add custom colored

electrumx/lib/atomicals_blueprint_builder.py

@@ -1,4 +1,5 @@
 from electrumx.lib.util_atomicals import (
+    is_custom_colored_operation,
     is_splat_operation,
     is_split_operation,
     is_mint_operation,
@@ -325,8 +326,37 @@ def calculate_output_blueprint_fts(cls, tx, ft_atomicals, operations_found_at_in
       should_split_ft_atomicals = is_split_operation(operations_found_at_inputs)
       if should_split_ft_atomicals:
           return AtomicalsTransferBlueprintBuilder.color_ft_atomicals_split(ft_atomicals, operations_found_at_inputs, tx, is_split_activated)
+
+      should_custom_colored_ft_atomicals = is_custom_colored_operation(operations_found_at_inputs)
+      if should_custom_colored_ft_atomicals:
+          return AtomicalsTransferBlueprintBuilder.custom_color_ft_atomicals(ft_atomicals, operations_found_at_inputs, tx)
       # Normal assignment in all cases including fall through of failure to provide a target exponent in the above resubstantiation
       return AtomicalsTransferBlueprintBuilder.color_ft_atomicals_regular(ft_atomicals, tx, sort_fifo, is_split_activated)
+
+  @classmethod
+  def custom_color_ft_atomicals(cls, ft_atomicals, operations_found_at_inputs, tx):
+    output_colored_map = {}
+    fts_burned = {}
+    cleanly_assigned = True
+    for atomical_id, atomical_info in sorted(ft_atomicals.items()):
+      remaining_value = atomical_info.total_atomical_value
+      for out_idx, txout in enumerate(tx.outputs):
+        expected_output_index = str(out_idx)
+        compact_atomical_id = location_id_bytes_to_compact(atomical_id)
+        expected_value = operations_found_at_inputs["payload"][compact_atomical_id].get(expected_output_index, 0)
+        if expected_value <= 0 or remaining_value <= 0:
+            continue
+        if expected_value > txout.value:
+            expected_value = txout.value
+        if expected_value < txout.value:
+            cleanly_assigned = False
+        output_colored_map[expected_output_index] = output_colored_map.get(expected_output_index) or {'atomicals': {}}
+        output_colored_map[expected_output_index]['atomicals'][atomical_id] = AtomicalColoredOutputFt(txout.value, expected_value, atomical_info)
+        remaining_value -= expected_value
+      if remaining_value > 0:
+        cleanly_assigned = False
+        fts_burned[atomical_id] = remaining_value
+    return AtomicalFtOutputBlueprintAssignmentSummary(output_colored_map, fts_burned, cleanly_assigned, None)
 
   @classmethod
   def color_ft_atomicals_split(cls, ft_atomicals, operations_found_at_inputs, tx, is_split_activated):

electrumx/lib/util_atomicals.py

@@ -1069,6 +1069,8 @@ def parse_operation_from_script(script, n):
             atom_op_decoded = 'x'  # extract - move atomical to 0'th output
         elif atom_op == "0179":
             atom_op_decoded = 'y'  # split - 
+        elif atom_op == "0180":
+            atom_op_decoded = 'z'  # custom colored
 
         if atom_op_decoded:
             return atom_op_decoded, parse_atomicals_data_definition_operation(script, n + one_letter_op_len)
@@ -1507,6 +1509,9 @@ def is_splat_operation(operations_found_at_inputs):
 def is_split_operation(operations_found_at_inputs):
     return operations_found_at_inputs and operations_found_at_inputs.get('op') == 'y' and operations_found_at_inputs.get('input_index') == 0
 
+def is_custom_colored_operation(operations_found_at_inputs):
+    return operations_found_at_inputs and operations_found_at_inputs.get('op') == 'z' and operations_found_at_inputs.get('input_index') == 0
+
 def is_seal_operation(operations_found_at_inputs):
     return operations_found_at_inputs and operations_found_at_inputs.get('op') == 'sl' and operations_found_at_inputs.get('input_index') == 0
 

tests/lib/test_atomicals_blueprint_builder.py

@@ -898,4 +898,63 @@ def mock_mint_fetcher(self, atomical_id):
     assert(len(ft_output_blueprint.outputs) == 2)
     assert(ft_output_blueprint.outputs[2]["atomicals"][ft_atomical_id].atomical_value == 1000)
     assert(ft_output_blueprint.fts_burned != {})
-    assert(blueprint_builder.get_are_fts_burned() == True)
+    assert(blueprint_builder.get_are_fts_burned() == True)
+
+
+def test_custom_colored_ft_normal():
+    raw_tx_str = '0100000000010213ac24b68388e0e32f3b19e95764c67d03b151d1f524eb07bc6e4f2790a3b7f00000000000ffffffff2423c79220c41bd904699aada54868e5c5aecb15168971964c6f5950a7b1d6860000000000ffffffff03e80300000000000022512011b6ce99eab0d8873d787e99e68a351358228893cdf1049ac48aae51391598abe80300000000000022512011b6ce99eab0d8873d787e99e68a351358228893cdf1049ac48aae51391598abe80300000000000022512011b6ce99eab0d8873d787e99e68a351358228893cdf1049ac48aae51391598ab03401aaa5ca0d475dcec02867f28f687494a639b3b43aff0a776c68d94f8cd3e987bb08a3463d8ab937f18f5dadfc916337b2df98cdd700b8514c6fdaff7f5ddffc975201764381bc0b54064cc55a0dda055c5e9875e5cdd7a7c1452d9b93d6015546170ac00630461746f6d017948a178423935323765666134333236323633366438663539313766633736336662646430393333336534623338376166643664346564376139303561313237623237623469301903e86821c01764381bc0b54064cc55a0dda055c5e9875e5cdd7a7c1452d9b93d60155461700140101db7c999f69c7f551d6800341a75ae659e8c100d1bb116b0935afc9ac3aec69bb97eed3ea72fa75912401400aa53f85f8a862f0f672620f31c5e704d8b4d5c00000000'
+    raw_tx = bytes.fromhex(raw_tx_str)
+    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"
+    operation_found_at_inputs["payload"] = {
+        "9527efa43262636d8f5917fc763fbdd09333e4b387afd6d4ed7a905a127b27b4i0": {
+            "0": 200,
+            "1": 300,
+        },
+        "8888e25790cc118773c8590522e1cab2ccc073a9375b238aaf9aadb13a764a13i0": {
+            "2": 8000,
+            "4": 8000,
+        }
+    }
+    atomicals_spent_at_inputs = {
+        1: [
+            {'atomical_id': subject_atomical_id1, 'location_id': b'not_used', 'data': b'not_used', 'data_value': {'sat_value': 1000, 'atomical_value': 1000}},
+            {'atomical_id': subject_atomical_id, 'location_id': b'not_used', 'data': b'not_used', 'data_value': {'sat_value': 1000, 'atomical_value': 1000}}
+        ]
+    }
+    def mock_mint_fetcher(self, atomical_id):
+        return {
+            'atomical_id': atomical_id,
+            'type': 'FT'
+        }
+    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) == 0)
+    ft_output_blueprint = blueprint_builder.get_ft_output_blueprint()
+    assert(ft_output_blueprint.cleanly_assigned == False)
+    assert(len(ft_output_blueprint.outputs) == 3)
+    assert(ft_output_blueprint.fts_burned != {})
+    assert(blueprint_builder.get_are_fts_burned() == True)
+
+    operation_found_at_inputs["payload"] = {
+        "9527efa43262636d8f5917fc763fbdd09333e4b387afd6d4ed7a905a127b27b4i0": {
+            "1": 1000,
+        },
+        "8888e25790cc118773c8590522e1cab2ccc073a9375b238aaf9aadb13a764a13i0": {
+            "2": 1000,
+        }
+    }
+    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) == 0)
+    ft_output_blueprint = blueprint_builder.get_ft_output_blueprint()
+    assert(ft_output_blueprint.cleanly_assigned == True)
+    assert(len(ft_output_blueprint.outputs) == 2)
+    assert(ft_output_blueprint.fts_burned == {})
+    assert(blueprint_builder.get_are_fts_burned() == False)