feat: implement new vote (#30)

solar_crypto/constants.py

@@ -16,7 +16,7 @@
     TRANSACTION_LEGACY_TRANSFER: "legacy_transfer",
     TRANSACTION_SECOND_SIGNATURE_REGISTRATION: "second_signature_registration",
     TRANSACTION_DELEGATE_REGISTRATION: "delegate_registration",
-    TRANSACTION_VOTE: "vote",
+    TRANSACTION_VOTE: "legacy_vote",
     TRANSACTION_MULTI_SIGNATURE_REGISTRATION: "multi_signature_registration",
     TRANSACTION_IPFS: "ipfs",
     TRANSACTION_TRANSFER: "transfer",
@@ -41,13 +41,16 @@
 }
 
 SOLAR_TRANSACTION_BURN = 0
+SOLAR_TRANSACTION_VOTE = 2
 
 SOLAR_TRANSACTION_TYPES = {
     SOLAR_TRANSACTION_BURN: "burn",
+    SOLAR_TRANSACTION_VOTE: "vote",
 }
 
 SOLAR_TRANSACTION_FEES = {
     SOLAR_TRANSACTION_BURN: 0,
+    SOLAR_TRANSACTION_VOTE: 9000000,
 }
 
 

solar_crypto/transactions/builder/legacy_vote.py

@@ -0,0 +1,38 @@
+import typing
+
+from solar_crypto.constants import TRANSACTION_VOTE
+from solar_crypto.identity.address import address_from_passphrase
+from solar_crypto.transactions.builder.base import BaseTransactionBuilder
+
+
+class LegacyVote(BaseTransactionBuilder):
+
+    transaction_type = TRANSACTION_VOTE
+
+    def __init__(self, vote=None, fee=None):
+        """Legacy vote transaction
+
+        Args:
+            vote (str): address of a delegate you want to vote
+            fee (int, optional): fee used for the transaction (default is already set)
+        """
+        super().__init__()
+
+        self.transaction.asset["votes"] = []
+        if vote:
+            self.transaction.asset["votes"].append(vote)
+
+        if fee:
+            self.transaction.fee = fee
+
+    def set_votes(self, votes: typing.List[str]):
+        """Set votes/unvotes
+
+        Args:
+            votes (List[str]): list of votes
+        """
+        self.transaction.asset["votes"] = votes
+
+    def sign(self, passphrase):
+        self.transaction.recipientId = address_from_passphrase(passphrase)
+        super().sign(passphrase)

solar_crypto/transactions/builder/vote.py

@@ -1,38 +1,129 @@
+import re
 import typing
+from decimal import Decimal
+from functools import cmp_to_key
+from math import trunc
 
-from solar_crypto.constants import TRANSACTION_VOTE
-from solar_crypto.identity.address import address_from_passphrase
+from solar_crypto.constants import SOLAR_TRANSACTION_VOTE, TRANSACTION_TYPE_GROUP
+from solar_crypto.exceptions import SolarInvalidTransaction
 from solar_crypto.transactions.builder.base import BaseTransactionBuilder
 
 
 class Vote(BaseTransactionBuilder):
 
-    transaction_type = TRANSACTION_VOTE
+    transaction_type = SOLAR_TRANSACTION_VOTE
+    typeGroup = TRANSACTION_TYPE_GROUP.SOLAR.value
 
-    def __init__(self, vote=None, fee=None):
-        """Create a second signature registration transaction
-
-        Args:
-            vote (str): address of a delegate you want to vote
-            fee (int, optional): fee used for the transaction (default is already set)
-        """
+    def __init__(self):
         super().__init__()
 
         self.transaction.asset["votes"] = []
-        if vote:
-            self.transaction.asset["votes"].append(vote)
-
-        if fee:
-            self.transaction.fee = fee
 
-    def set_votes(self, votes: typing.List[str]):
-        """Set votes/unvotes
+    def set_votes(
+        self,
+        votes: typing.Union[
+            typing.List[str], typing.Dict[str, typing.Union[int, float, Decimal]]
+        ] = dict,
+    ):
+        """Set votes
 
         Args:
-            votes (List[str]): list of votes
+            votes
         """
+        vote_object: typing.Dict[str, typing.Union[float, int]] = {}
+
+        if isinstance(votes, list):
+            vote_list = filter(lambda vote: not vote.startswith("-"), votes)
+            vote_list = list(
+                map(lambda vote: vote[1:] if vote.startswith("+") else vote, vote_list)
+            )
+
+            if len(vote_list) > 53:
+                raise SolarInvalidTransaction("Unable to vote for more than 53 delegates")
+
+            if len(vote_list) == 0:
+                self.transaction.asset["votes"] = {}
+                return
+
+            weight = trunc(((((100 / len(vote_list))) * 100) / 100) * 100)
+            remainder = 10000
+
+            for vote in vote_list:
+                vote_object[vote] = weight / 100
+                remainder -= weight
+
+            for index in range(int(remainder)):
+                key = list(vote_object.keys())[index]
+                vote_object[key] = round((vote_object[key] + 0.01) * 100) / 100
+
+            votes = vote_object
+        else:
+            for key, val in votes.items():
+                votes[key] = val
+
+        validate(votes)
+
+        if votes:
+            nr_of_votes = len(votes.keys())
+            if nr_of_votes > 0:
+                votes = sort_votes(votes)
+
         self.transaction.asset["votes"] = votes
 
-    def sign(self, passphrase):
-        self.transaction.recipientId = address_from_passphrase(passphrase)
-        super().sign(passphrase)
+
+def validate(votes):
+    for value in votes.values():
+        if not valid_precision(value):
+            raise SolarInvalidTransaction("Only two decimal places are allowed.")
+
+    if Decimal(sum(votes.values())) != Decimal("100"):
+        raise SolarInvalidTransaction("Total vote weight must equal 100.")
+
+
+def valid_precision(value, max_precision=2):
+    if isinstance(value, Decimal):
+        if abs(value.as_tuple().exponent) <= max_precision:
+            return True
+    elif isinstance(value, float) or isinstance(value, str) or isinstance(value, int):
+        if str(value)[::-1].find(".") <= max_precision:
+            return True
+    return False
+
+
+def cmp(a: typing.List[typing.Union[int, str]], b: typing.List[typing.Union[int, str]]):
+    """
+    Compare two alphanum keys
+    """
+    return (a > b) - (a < b)
+
+
+def nat_cmp(a: str, b: str):
+    """
+    Natural comparison
+    """
+    convert = lambda text: int(text) if text.isdigit() else text.lower()  # noqa: E731
+    alphanum_key = lambda key: [convert(c) for c in re.split("([0-9]+)", key)]  # noqa: E731
+    return cmp(alphanum_key(a), alphanum_key(b))
+
+
+def sorter(
+    a: typing.Tuple[str, typing.Union[float, int]], b: typing.Tuple[str, typing.Union[float, int]]
+):
+    """
+    Sort using desc weight and asc by name
+    """
+    if b[1] > a[1]:
+        return 1
+    elif b[1] < a[1]:
+        return -1
+    else:
+        return nat_cmp(a[0], b[0])
+
+
+def sort_votes(votes: typing.Dict[str, typing.Union[float, int]]):
+    """
+    Sort votes using custom sorter function
+    """
+    sorter_fn = cmp_to_key(sorter)
+    sorted_votes = sorted(votes.items(), key=sorter_fn)
+    return dict(sorted_votes)

solar_crypto/transactions/deserializers/legacy_vote.py

@@ -0,0 +1,57 @@
+from binascii import hexlify
+
+from binary.unsigned_integer.reader import read_bit8
+
+from solar_crypto.exceptions import SolarDeserializerException
+from solar_crypto.transactions.deserializers.base import BaseDeserializer
+
+
+class LegacyVoteDeserializer(BaseDeserializer):
+    def deserialize(self):
+        starting_position = int(self.asset_offset / 2)
+        offset = 0
+
+        vote_length = read_bit8(self.serialized, offset=starting_position)
+        offset += 1
+
+        self.transaction.asset["votes"] = []
+
+        for _ in range(vote_length):
+            if (
+                self.transaction.version == 2
+                and self.serialized[starting_position + offset : starting_position + offset + 1]
+                != b"\xff"
+            ):
+                vote_buffer = self.serialized[
+                    starting_position + offset : starting_position + offset + 34
+                ]
+                offset += 34
+                prefix = "+" if vote_buffer[0] == 1 else "-"
+                vote = f"{prefix}{vote_buffer[1::].hex()}"
+            else:
+                if self.transaction.version == 2:
+                    offset += 1  # +1 due to NOT moving forwards when checking for `b"\xff"`
+
+                length = read_bit8(
+                    self.serialized[starting_position + offset : starting_position + offset + 1]
+                )
+                offset += 1
+
+                vote_buffer = self.serialized[
+                    starting_position + offset : starting_position + offset + length
+                ]
+                offset += length
+
+                prefix = "+" if vote_buffer[0] == 1 else "-"
+                vote = f"{prefix}{vote_buffer[1::].decode()}"
+
+            if len(vote) <= 1:
+                raise SolarDeserializerException("Invalid transaction data")
+
+            self.transaction.asset["votes"].append(vote)
+
+        self.transaction.parse_signatures(
+            hexlify(self.serialized).decode(), self.asset_offset + 2 + ((offset - 1) * 2)
+        )
+
+        return self.transaction

solar_crypto/transactions/deserializers/vote.py

@@ -1,8 +1,7 @@
 from binascii import hexlify
 
-from binary.unsigned_integer.reader import read_bit8
+from binary.unsigned_integer.reader import read_bit8, read_bit16
 
-from solar_crypto.exceptions import SolarDeserializerException
 from solar_crypto.transactions.deserializers.base import BaseDeserializer
 
 
@@ -11,44 +10,24 @@ def deserialize(self):
         starting_position = int(self.asset_offset / 2)
         offset = 0
 
-        vote_length = read_bit8(self.serialized, offset=starting_position)
+        number_of_votes = read_bit8(self.serialized, offset=starting_position)
         offset += 1
 
-        self.transaction.asset["votes"] = []
-
-        for _ in range(vote_length):
-            if (
-                self.transaction.version == 2
-                and self.serialized[starting_position + offset : starting_position + offset + 1]
-                != b"\xff"
-            ):
-                vote_buffer = self.serialized[
-                    starting_position + offset : starting_position + offset + 34
-                ]
-                offset += 34
-                prefix = "+" if vote_buffer[0] == 1 else "-"
-                vote = f"{prefix}{vote_buffer[1::].hex()}"
-            else:
-                if self.transaction.version == 2:
-                    offset += 1  # +1 due to NOT moving forwards when checking for `b"\xff"`
-
-                length = read_bit8(
-                    self.serialized[starting_position + offset : starting_position + offset + 1]
-                )
-                offset += 1
-
-                vote_buffer = self.serialized[
-                    starting_position + offset : starting_position + offset + length
-                ]
-                offset += length
-
-                prefix = "+" if vote_buffer[0] == 1 else "-"
-                vote = f"{prefix}{vote_buffer[1::].decode()}"
-
-            if len(vote) <= 1:
-                raise SolarDeserializerException("Invalid transaction data")
-
-            self.transaction.asset["votes"].append(vote)
+        self.transaction.asset["votes"] = dict()
+
+        for _ in range(number_of_votes):
+            vote_len: int = read_bit8(self.serialized, offset=starting_position + offset)
+            offset += 1
+
+            vote: str = self.serialized[
+                starting_position + offset : starting_position + offset + vote_len
+            ].decode()
+            offset += vote_len
+
+            percent = read_bit16(self.serialized, offset=starting_position + offset) / 100
+            offset += 2
+
+            self.transaction.asset["votes"].update({vote: percent})
 
         self.transaction.parse_signatures(
             hexlify(self.serialized).decode(), self.asset_offset + 2 + ((offset - 1) * 2)

solar_crypto/transactions/serializers/legacy_vote.py

@@ -0,0 +1,33 @@
+from binascii import hexlify, unhexlify
+
+from binary.unsigned_integer.writer import write_bit8
+
+from solar_crypto.transactions.serializers.base import BaseSerializer
+
+
+class LegacyVoteSerializer(BaseSerializer):
+    """Serializer handling legacy vote"""
+
+    def serialize(self):
+        vote_bytes = []
+
+        for vote in self.transaction["asset"]["votes"]:
+            prefix = "01" if vote.startswith("+") else "00"
+            sliced = vote[1::]
+
+            if len(sliced) == 66:
+                vote_bytes.append(f"{prefix}{sliced}")
+                continue
+
+            # vote.length.toString(16).padStart(2, "0") + prefix + Buffer.from(sliced).toString("hex");
+            start = format(len(vote), "x").zfill(2)
+            vote_hex = f"{start}{prefix}{hexlify(sliced.encode()).decode()}"
+            if self.transaction["version"] == 2:
+                vote_hex = f"ff{vote_hex}"
+
+            vote_bytes.append(vote_hex)
+
+        self.bytes_data += write_bit8(len(self.transaction["asset"]["votes"]))
+        self.bytes_data += unhexlify("".join(vote_bytes))
+
+        return self.bytes_data