refactor(encoding): replace standalone functions with Reader/Writer classes

examples/example_decode_block.py

@@ -11,10 +11,10 @@
 """
 
 import asyncio
-import io
 import random
 
 from pactus.block import Block
+from pactus.encoding import Reader
 from pactus_jsonrpc.client import PactusOpenRPCClient
 
 TESTNET_RPC = "https://testnet1.pactus.org/jsonrpc"
@@ -36,7 +36,7 @@ async def fetch_and_decode_block():
     print(f"\nFetching block at height {height}...")
 
     res = await client.pactus.blockchain.get_block(height=height, verbosity=0)
-    block = Block.decode(io.BytesIO(bytes.fromhex(res["data"])))
+    block = Block.decode(Reader(bytes.fromhex(res["data"])))
 
     print(f"  - Hash:       {block.id}")
     print(f"  - Version:    {block.header.version}")

pactus/block/block.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 
 import hashlib
-import io
 import struct
 
 from pactus.crypto.hash import Hash
-from pactus.encoding import encoding
+from pactus.encoding import Reader, Writer
 from pactus.transaction import Transaction
 
 from .certificate import Certificate
@@ -24,22 +23,22 @@ def __init__(
         self.transactions = transactions
 
     @classmethod
-    def decode(cls, buf: io.BufferedReader) -> Block:
+    def decode(cls, reader: Reader) -> Block:
         """Decode a Block from the stream."""
-        header = Header.decode(buf)
+        header = Header.decode(reader)
 
         # Genesis block has no certificate
         is_genesis = header.prev_block_hash.is_undef()
 
         prev_cert = None
         if not is_genesis:
-            prev_cert = Certificate.decode(buf)
+            prev_cert = Certificate.decode(reader)
 
-        num_txs = encoding.read_var_int(buf)
+        num_txs = reader.read_var_int()
 
         transactions = []
         for _ in range(num_txs):
-            tx = Transaction.decode(buf)
+            tx = Transaction.decode(reader)
             transactions.append(tx)
 
         return cls(header, prev_cert, transactions)
@@ -55,9 +54,9 @@ def id(self) -> Hash:
         return Hash(hashlib.blake2b(buf, digest_size=32).digest())
 
     def _header_bytes(self) -> bytes:
-        buf = io.BytesIO()
-        self.header.encode(buf)
-        return buf.getvalue()
+        writer = Writer()
+        self.header.encode(writer)
+        return writer.bytes()
 
     def _txs_root(self) -> bytes:
         return Block._merkle_root([tx.id().data for tx in self.transactions])

pactus/block/certificate.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
 
 import hashlib
-import io
 
 from pactus.crypto.bls.signature import Signature
-from pactus.encoding import encoding
+from pactus.encoding import Reader, Writer
 from pactus.types.height import Height
 from pactus.types.round import Round
 
@@ -25,40 +24,40 @@ def __init__(
         self.signature = signature
 
     @classmethod
-    def decode(cls, buf: io.BufferedReader) -> Certificate:
+    def decode(cls, reader: Reader) -> Certificate:
         """Decode a Certificate from the stream."""
-        height = Height.decode(buf)
-        round_ = Round.decode(buf)
+        height = Height.decode(reader)
+        round_ = Round.decode(reader)
 
-        num_committers = encoding.read_var_int(buf)
+        num_committers = reader.read_var_int()
         committers = []
         for _ in range(num_committers):
-            n = encoding.read_var_int(buf)
+            n = reader.read_var_int()
             committers.append(n)
 
-        num_absentees = encoding.read_var_int(buf)
+        num_absentees = reader.read_var_int()
         absentees = []
         for _ in range(num_absentees):
-            n = encoding.read_var_int(buf)
+            n = reader.read_var_int()
             absentees.append(n)
 
-        signature = Signature.decode(buf)
+        signature = Signature.decode(reader)
 
         return cls(height, round_, committers, absentees, signature)
 
-    def encode(self, buf: io.Writer) -> None:
-        self.height.encode(buf)
-        self.round.encode(buf)
-        encoding.append_var_int(buf, len(self.committers))
+    def encode(self, writer: Writer) -> None:
+        self.height.encode(writer)
+        self.round.encode(writer)
+        writer.write_var_int(len(self.committers))
         for n in self.committers:
-            encoding.append_var_int(buf, n)
-        encoding.append_var_int(buf, len(self.absentees))
+            writer.write_var_int(n)
+        writer.write_var_int(len(self.absentees))
         for n in self.absentees:
-            encoding.append_var_int(buf, n)
-        self.signature.encode(buf)
+            writer.write_var_int(n)
+        self.signature.encode(writer)
 
     def hash(self) -> bytes:
         """Return the certificate hash (blake2b-256 of encoded bytes)."""
-        buf = io.BytesIO()
-        self.encode(buf)
-        return hashlib.blake2b(buf.getvalue(), digest_size=32).digest()
+        writer = Writer()
+        self.encode(writer)
+        return hashlib.blake2b(writer.bytes(), digest_size=32).digest()

pactus/block/header.py

@@ -1,13 +1,8 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    import io
-
 from pactus.crypto.address import Address
 from pactus.crypto.hash import Hash
-from pactus.encoding import encoding
+from pactus.encoding import Reader, Writer
 
 
 class Header:
@@ -28,14 +23,14 @@ def __init__(
         self.proposer_address = proposer_address
 
     @classmethod
-    def decode(cls, buf: io.BufferedReader) -> Header:
+    def decode(cls, reader: Reader) -> Header:
         """Decode a Header from the stream."""
-        version = encoding.read_uint8(buf)
-        unix_time = encoding.read_uint32(buf)
-        prev_block_hash = Hash.decode(buf)
-        state_root = Hash.decode(buf)
-        sortition_seed = encoding.read_fixed_bytes(buf, 48)
-        proposer_address = Address.decode(buf)
+        version = reader.read_uint8()
+        unix_time = reader.read_uint32()
+        prev_block_hash = Hash.decode(reader)
+        state_root = Hash.decode(reader)
+        sortition_seed = reader.read_fixed_bytes(48)
+        proposer_address = Address.decode(reader)
 
         return cls(
             version,
@@ -46,10 +41,10 @@ def decode(cls, buf: io.BufferedReader) -> Header:
             proposer_address,
         )
 
-    def encode(self, buf: io.Writer) -> None:
-        encoding.append_uint8(buf, self.version)
-        encoding.append_uint32(buf, self.unix_time)
-        self.prev_block_hash.encode(buf)
-        self.state_root.encode(buf)
-        encoding.append_fixed_bytes(buf, self.sortition_seed)
-        self.proposer_address.encode(buf)
+    def encode(self, writer: Writer) -> None:
+        writer.write_uint8(self.version)
+        writer.write_uint32(self.unix_time)
+        self.prev_block_hash.encode(writer)
+        self.state_root.encode(writer)
+        writer.write_fixed_bytes(self.sortition_seed)
+        self.proposer_address.encode(writer)

pactus/crypto/address.py

@@ -1,15 +1,11 @@
 from __future__ import annotations
 
 from enum import Enum
-from typing import TYPE_CHECKING
 
 from pactus.crypto.hrp import HRP
-from pactus.encoding import encoding
+from pactus.encoding import Reader, Writer
 from pactus.utils import utils
 
-if TYPE_CHECKING:
-    import io
-
 # Address format: hrp + `1` + type + data + checksum
 
 ADDRESS_SIZE = 21
@@ -80,28 +76,23 @@ def is_treasury_address(self) -> bool:
         return self.address_type() == AddressType.TREASURY
 
     def is_account_address(self) -> bool:
-        t = self.address_type()
-        return t in (
-            AddressType.TREASURY,
-            AddressType.BLS_ACCOUNT,
-            AddressType.ED25519_ACCOUNT,
-        )
+        return not self.is_validator_address()
 
     def is_validator_address(self) -> bool:
         return self.address_type() == AddressType.VALIDATOR
 
-    def encode(self, buf: io.Writer) -> None:
+    def encode(self, writer: Writer) -> None:
         if self.is_treasury_address():
-            encoding.append_uint8(buf, AddressType.TREASURY.value)
+            writer.write_uint8(AddressType.TREASURY.value)
         else:
-            encoding.append_fixed_bytes(buf, self.raw_bytes())
+            writer.write_fixed_bytes(self.raw_bytes())
 
     @classmethod
-    def decode(cls, buf: io.BufferedReader) -> Address:
+    def decode(cls, reader: Reader) -> Address:
         """Decode an Address from the stream."""
-        addr_type = encoding.read_uint8(buf)
+        addr_type = reader.read_uint8()
         if addr_type == AddressType.TREASURY.value:
             return cls(AddressType.TREASURY, bytes(20))
 
-        data = encoding.read_fixed_bytes(buf, 20)
+        data = reader.read_fixed_bytes(20)
         return cls(AddressType(addr_type), data)

pactus/crypto/bls/public_key.py

@@ -1,22 +1,18 @@
 from __future__ import annotations
 
 import hashlib
-from typing import TYPE_CHECKING
 
 from ripemd.ripemd160 import ripemd160
 
 from pactus.crypto.address import Address, AddressType
 from pactus.crypto.hrp import HRP
-from pactus.encoding import encoding
+from pactus.encoding import Reader, Writer
 from pactus.utils import utils
 
 from .bls12_381.bls_sig_g1 import aggregate_pubs, verify
 from .bls12_381.serdesZ import deserialize, serialize
 from .signature import DST, SIGNATURE_TYPE_BLS, Signature
 
-if TYPE_CHECKING:
-    import io
-
 PUBLIC_KEY_SIZE = 96
 
 
@@ -65,12 +61,12 @@ def string(self) -> str:
             self.raw_bytes(),
         )
 
-    def encode(self, buf: io.Writer) -> None:
-        encoding.append_fixed_bytes(buf, self.raw_bytes())
+    def encode(self, writer: Writer) -> None:
+        writer.write_fixed_bytes(self.raw_bytes())
 
     @classmethod
-    def decode(cls, buf: io.BufferedReader) -> PublicKey:
-        data = encoding.read_fixed_bytes(buf, PUBLIC_KEY_SIZE)
+    def decode(cls, reader: Reader) -> PublicKey:
+        data = reader.read_fixed_bytes(PUBLIC_KEY_SIZE)
         return cls.from_bytes(data)
 
     def account_address(self) -> Address:

pactus/crypto/bls/signature.py

@@ -1,15 +1,10 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-from pactus.encoding import encoding
+from pactus.encoding import Reader, Writer
 
 from .bls12_381.bls_sig_g1 import aggregate_sigs
 from .bls12_381.serdesZ import deserialize, serialize
 
-if TYPE_CHECKING:
-    import io
-
 SIGNATURE_SIZE = 48
 SIGNATURE_TYPE_BLS = 1
 DST = b"BLS_SIG_BLS12381G1_XMD:SHA-256_SSWU_RO_NUL_"
@@ -44,11 +39,11 @@ def raw_bytes(self) -> bytes:
     def string(self) -> str:
         return self.raw_bytes().hex()
 
-    def encode(self, buf: io.Writer) -> None:
-        encoding.append_fixed_bytes(buf, self.raw_bytes())
+    def encode(self, writer: Writer) -> None:
+        writer.write_fixed_bytes(self.raw_bytes())
 
     @classmethod
-    def decode(cls, buf: io.BufferedReader) -> Signature:
-        data = encoding.read_fixed_bytes(buf, SIGNATURE_SIZE)
+    def decode(cls, reader: Reader) -> Signature:
+        data = reader.read_fixed_bytes(SIGNATURE_SIZE)
         point_g1 = deserialize(data, is_ell2=False)
         return cls(point_g1)

pactus/crypto/ed25519/public_key.py

@@ -1,22 +1,18 @@
 from __future__ import annotations
 
 import hashlib
-from typing import TYPE_CHECKING
 
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives.asymmetric import ed25519
 from ripemd.ripemd160 import ripemd160
 
 from pactus.crypto.address import Address, AddressType
 from pactus.crypto.hrp import HRP
-from pactus.encoding import encoding
+from pactus.encoding import Reader, Writer
 from pactus.utils import utils
 
 from .signature import SIGNATURE_TYPE_ED25519, Signature
 
-if TYPE_CHECKING:
-    import io
-
 PUBLIC_KEY_SIZE = 32
 
 
@@ -54,12 +50,12 @@ def string(self) -> str:
             self.raw_bytes(),
         )
 
-    def encode(self, buf: io.Writer) -> None:
-        encoding.append_fixed_bytes(buf, self.raw_bytes())
+    def encode(self, writer: Writer) -> None:
+        writer.write_fixed_bytes(self.raw_bytes())
 
     @classmethod
-    def decode(cls, buf: io.BufferedReader) -> PublicKey:
-        data = encoding.read_fixed_bytes(buf, PUBLIC_KEY_SIZE)
+    def decode(cls, reader: Reader) -> PublicKey:
+        data = reader.read_fixed_bytes(PUBLIC_KEY_SIZE)
         return cls(ed25519.Ed25519PublicKey.from_public_bytes(data))
 
     def account_address(self) -> Address:

pactus/crypto/ed25519/signature.py

@@ -1,11 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-from pactus.encoding import encoding
-
-if TYPE_CHECKING:
-    import io
+from pactus.encoding import Reader, Writer
 
 SIGNATURE_SIZE = 64
 SIGNATURE_TYPE_ED25519 = 3
@@ -31,10 +26,10 @@ def raw_bytes(self) -> bytes:
     def string(self) -> str:
         return self.sig.hex()
 
-    def encode(self, buf: io.Writer) -> None:
-        encoding.append_fixed_bytes(buf, self.sig)
+    def encode(self, writer: Writer) -> None:
+        writer.write_fixed_bytes(self.sig)
 
     @classmethod
-    def decode(cls, buf: io.BufferedReader) -> Signature:
-        data = encoding.read_fixed_bytes(buf, SIGNATURE_SIZE)
+    def decode(cls, reader: Reader) -> Signature:
+        data = reader.read_fixed_bytes(SIGNATURE_SIZE)
         return cls(data)