Basic heartbeat transaction parsing support.

algosdk/constants.py

@@ -30,6 +30,8 @@
 """str: indicates an app call transaction, allows creating, deleting, and interacting with an application"""
 STATEPROOF_TXN = "stpf"
 """str: indicates an state proof transaction"""
+HEARTBEAT_TXN = "hb"
+"""str: indicates a heartbeat transaction"""
 
 # note field types
 NOTE_FIELD_TYPE_DEPOSIT = "d"
@@ -139,3 +141,4 @@
 logic_sig_max_size = LOGIC_SIG_MAX_SIZE
 app_page_max_size = APP_PAGE_MAX_SIZE
 stateproof_txn = STATEPROOF_TXN
+heartbeat_txn = HEARTBEAT_TXN

algosdk/transaction.py

@@ -263,6 +263,9 @@ def undictify(d):
             args.pop("note"), args.pop("rekey_to"), args.pop("lease")
             args.update(StateProofTxn._undictify(d))
             txn = StateProofTxn(**args)
+        elif txn_type == constants.heartbeat_txn:
+            args.update(HeartbeatTxn._undictify(d))
+            txn = HeartbeatTxn(**args)
         if "grp" in d:
             txn.group = d["grp"]
         return txn
@@ -3010,6 +3013,92 @@ def __eq__(self, other):
         return False
 
 
+class HeartbeatTxn(Transaction):
+    """
+    Represents a heartbeat transaction
+
+    Arguments:
+        sender (str): address of the sender (note that sender may be different from heartbeat_address)
+        sp (SuggestedParams): suggested params from algod
+        heartbeat_address (str, optional): account this txn is proving onlineness for
+        heartbeat_proof (dict(), optional): signature using heartbeat_address's partkey
+        heartbeat_seed (str, optional): the block seed for the block before this transaction's firstValid
+        note (bytes, optional): arbitrary optional bytes
+        lease (byte[32], optional): specifies a lease, and no other transaction
+            with the same sender and lease can be confirmed in this
+            transaction's valid rounds
+        rekey_to (str, optional): additionally rekey the sender to this address
+
+
+    Attributes:
+        sender (str)
+        hb_address (str)
+        hb_proof (dict())
+        hb_seed (str)
+        first_valid_round (int)
+        last_valid_round (int)
+        genesis_id (str)
+        genesis_hash (str)
+        type (str)
+    """
+
+    def __init__(
+        self,
+        sender,
+        sp,
+        heartbeat_address=None,
+        heartbeat_proof=None,
+        heartbeat_seed=None,
+        note=None,
+        lease=None,
+        rekey_to=None,
+    ):
+        Transaction.__init__(
+            self, sender, sp, note, lease, constants.heartbeat_txn, rekey_to
+        )
+
+        self.hb_address = heartbeat_address
+        self.hb_proof = heartbeat_proof
+        self.hb_seed = heartbeat_seed
+
+    def dictify(self):
+        d = dict()
+        if self.hb_address:
+            d["hbaddr"] = self.hb_address
+        if self.hb_proof:
+            d["hbprf"] = self.hb_proof
+        if self.hb_seed:
+            d["hbseed"] = self.hb_seed
+        d.update(super(HeartbeatTxn, self).dictify())
+        od = OrderedDict(sorted(d.items()))
+
+        return od
+
+    @staticmethod
+    def _undictify(d):
+        args = {}
+        if "hbaddr" in d:
+            args["heartbeat_address"] = d["hbaddr"]
+        if "hbprf" in d:
+            args["heartbeat_proof"] = d["hbprf"]
+        if "hbseed" in d:
+            args["heartbeat_seed"] = d["hbseed"]
+
+        return args
+
+    def __eq__(self, other):
+        if not isinstance(other, HeartbeatTxn):
+            return False
+        return (
+            super(HeartbeatTxn, self).__eq__(other)
+            and self.hb_address == other.hb_address
+            and self.hb_proof == other.hb_proof
+            and self.hb_seed == other.hb_seed
+        )
+
+        return False
+
+
 GenericSignedTransaction = Union[
     SignedTransaction,
     LogicSigTransaction,

tests/unit_tests/test_transaction.py

@@ -1414,6 +1414,41 @@ def test_application_delete(self):
         self.assertEqual(i, call)
 
 
+class TestHeartbeatTransactions(unittest.TestCase):
+    sender = "7ZUECA7HFLZTXENRV24SHLU4AVPUTMTTDUFUBNBD64C73F3UHRTHAIOF6Q"
+    genesis = "JgsgCaCTqIaLeVhyL6XlRu3n7Rfk2FxMeK+wRSaQ7dI="
+    hb_address = "MTTDUFUBNBD64C73F3UHRTHAIOF6Q7ZUECA7HFLZTXENRV24SHLU4AVPUT"
+    hb_proof = {
+        "Sig": "AQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==",
+        "PK": "AQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==",
+        "PK2": "AQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==",
+        "PK1Sig": "AQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==",
+        "Pk2Sig": "AQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA==",
+    }
+    hb_seed = "XNOEe4VHUzo+8kWmV8ZE1T/GViVIkC9zbz0IhwhKTjY="
+    note = b"\x00"
+    lease = b"\0" * 32
+    params = transaction.SuggestedParams(0, 1, 100, genesis)
+
+    def test_heartbeat(self):
+        # Test serializing a basic heartbeat with all standard fields filled in (save for rekey).
+        hb = transaction.HeartbeatTxn(
+            self.sender,
+            self.params,
+            self.hb_address,
+            self.hb_proof,
+            self.hb_seed,
+            self.note,
+            self.lease,
+        )
+
+        enc = encoding.msgpack_encode(hb)
+        re_enc = encoding.msgpack_encode(encoding.msgpack_decode(enc))
+        self.assertEqual(enc, re_enc)
+
+        self.assertEqual(transaction.HeartbeatTxn.undictify(hb.dictify()), hb)
+
+
 class TestBoxReference(unittest.TestCase):
     def test_translate_box_references(self):
         # Test case: reference input, foreign app array, caller app id, expected output