Skip to content

Commit

Permalink
Merge pull request #72 from A91y/master
Browse files Browse the repository at this point in the history
feat: added __repr__ method
  • Loading branch information
GitBolt authored May 26, 2024
2 parents d06af20 + 683ecbe commit 618f0e5
Show file tree
Hide file tree
Showing 4 changed files with 114 additions and 14 deletions.
87 changes: 76 additions & 11 deletions solathon/core/types/block.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from solathon.core.message import Message as CoreMessage, MessageHeader
from typing import Any, List, TypedDict, Union


class HeaderType(TypedDict):
'''
JSON Response type of Header Information received by RPC
Expand All @@ -9,13 +10,18 @@ class HeaderType(TypedDict):
numReadonlyUnsignedAccounts: int
numRequiredSignatures: int


class Header:

def __init__(self, response: HeaderType) -> None:
self.num_readonly_signed_accounts = response['numReadonlySignedAccounts']
self.num_readonly_unsigned_accounts = response['numReadonlyUnsignedAccounts']
self.num_required_signatures = response['numRequiredSignatures']

def __repr__(self) -> str:
return f"Header(num_required_signatures={self.num_required_signatures!r}, num_readonly_signed_accounts={self.num_readonly_signed_accounts!r}, num_readonly_unsigned_accounts={self.num_readonly_unsigned_accounts!r})"


class InstructionType(TypedDict):
'''
JSON Response type of Instruction Information received by RPC
Expand All @@ -24,15 +30,21 @@ class InstructionType(TypedDict):
data: str
programIdIndex: int


class Instruction:
'''
Convert Instruction JSON to Class
'''

def __init__(self, response: InstructionType) -> None:
self.accounts = response['accounts']
self.data = response['data']
self.program_id_index = response['programIdIndex']

def __repr__(self) -> str:
return f"Instruction(num_accounts={len(self.accounts)!r}, program_id_index={self.program_id_index!r})"


class MessageType(TypedDict):
'''
JSON Response type of Message Information received by RPC
Expand All @@ -42,10 +54,12 @@ class MessageType(TypedDict):
instructions: List[InstructionType]
recentBlockhash: str


class Message:
'''
Convert Message JSON to Class
'''

def __init__(self, response: MessageType) -> None:
self.account_keys = response['accountKeys']
header = Header(response['header'])
Expand All @@ -54,9 +68,13 @@ def __init__(self, response: MessageType) -> None:
num_readonly_signed_accounts=header.num_readonly_signed_accounts,
num_readonly_unsigned_accounts=header.num_readonly_unsigned_accounts
)
self.instructions = [Instruction(instruction) for instruction in response['instructions']]
self.instructions = [Instruction(instruction)
for instruction in response['instructions']]
self.recent_blockhash = response['recentBlockhash']

def __repr__(self) -> str:
return f"Message(header={self.header!r}, num_instructions={len(self.instructions)!r}, recent_blockhash={self.recent_blockhash!r})"


class TransactionType(TypedDict):
'''
Expand All @@ -65,10 +83,12 @@ class TransactionType(TypedDict):
message: MessageType
signatures: List[str]


class Transaction:
'''
Convert Transaction JSON to Class
'''

def __init__(self, response: TransactionType) -> None:
message = Message(response['message'])
self.message = CoreMessage(
Expand All @@ -79,6 +99,10 @@ def __init__(self, response: TransactionType) -> None:
)
self.signatures = response['signatures']

def __repr__(self) -> str:
return f"Transaction(message={self.message!r}, signatures={self.signatures!r})"


class MetaType(TypedDict):
'''
JSON Response type of Meta Information received by RPC
Expand All @@ -93,10 +117,12 @@ class MetaType(TypedDict):
preTokenBalances: List[Any]
rewards: Union[Any, None]


class Meta:
'''
Convert Meta JSON to Class
'''

def __init__(self, response: MetaType) -> None:
self.err = response['err']
self.fee = response['fee']
Expand All @@ -108,25 +134,31 @@ def __init__(self, response: MetaType) -> None:
self.pre_token_balances = response['preTokenBalances']
self.rewards = response['rewards']

def __repr__(self) -> str:
return f"Meta(err={self.err!r}, fee={self.fee!r}, num_inner_instructions={len(self.inner_instructions)!r})"


class TransactionElementType(TypedDict):
'''
JSON Response type of Transaction Information received by RPC
'''
meta: MetaType
transaction: TransactionType


class TransactionElement:
'''
Convert Transaction JSON to Class
'''

def __init__(self, response: TransactionElementType) -> None:
self.meta = Meta(response['meta'])
self.transaction = Transaction(response['transaction'])

def __repr__(self) -> str:
return f"TransactionElement(signatures={self.transaction.signatures!r})"


class BlockType(TypedDict):
'''
JSON Response type of Block Information received by RPC
Expand All @@ -138,6 +170,7 @@ class BlockType(TypedDict):
previousBlockhash: str
transactions: List[TransactionElementType]


class Block:
'''
Convert Block JSON to Class
Expand All @@ -149,7 +182,12 @@ def __init__(self, response: BlockType) -> None:
self.blockhash = response['blockhash']
self.parent_slot = response['parent_slot']
self.previous_blockhash = response['previous_blockhash']
self.transactions = [TransactionElement(transaction) for transaction in response['transactions']]
self.transactions = [TransactionElement(
transaction) for transaction in response['transactions']]

def __repr__(self) -> str:
return f"Block(block_height={self.block_height!r}, block_time={self.block_time!r}, blockhash={self.blockhash!r},num_transactions={len(self.transactions)!r})"


class RangeType(TypedDict):
'''
Expand All @@ -158,57 +196,68 @@ class RangeType(TypedDict):
firstSlot: int
lastSlot: int


class Range:
'''
Convert Range JSON to Class
'''

def __init__(self, response: RangeType) -> None:
self.first_slot = response['firstSlot']
self.last_slot = response['lastSlot']

def __repr__(self) -> str:
return f"Range(first_slot={self.first_slot!r}, last_slot={self.last_slot!r})"


class BlockProductionType(TypedDict):
'''
JSON Response type of Block Production Information received by RPC
'''
byIdentity: dict[str, Any]
range: RangeType


class BlockProduction:
'''
Convert Block Production JSON to Class
'''

def __init__(self, response: BlockProductionType) -> None:
self.by_identity = response['byIdentity']
self.range = Range(response['range'])

def __repr__(self) -> str:
return f"BlockProduction(by_identity={self.by_identity!r}, range={self.range!r})"


class BlockCommitmentType(TypedDict):
'''
JSON Response type of Block Commitment Information received by RPC
'''
commitment: List[int]
totalStake: int


class BlockCommitment:
'''
Convert Block Commitment JSON to Class
'''

def __init__(self, response: BlockCommitmentType) -> None:
self.commitment = response['commitment']
self.total_stake = response['totalStake']

def __repr__(self) -> str:
return f"BlockCommitment(commitment={self.commitment!r}, total_stake={self.total_stake!r})"


class FeeCalculatorType(TypedDict):
'''
JSON Response type of Fee Calculator Information received by RPC
'''
lamportsPerSignature: int

class BlockHashType(TypedDict):
'''
JSON Response type of Block Hash Information received by RPC
'''
blockhash: str
feeCalculator: FeeCalculatorType

class FeeCalculator:
'''
Expand All @@ -218,10 +267,26 @@ class FeeCalculator:
def __init__(self, response: FeeCalculatorType) -> None:
self.lamports_per_signature = response['lamportsPerSignature']

def __repr__(self) -> str:
return f"FeeCalculator(lamports_per_signature={self.lamports_per_signature!r})"


class BlockHashType(TypedDict):
'''
JSON Response type of Block Hash Information received by RPC
'''
blockhash: str
feeCalculator: FeeCalculatorType


class BlockHash:
'''
Convert Block Hash JSON to Class
'''

def __init__(self, response: BlockHashType) -> None:
self.blockhash = response['blockhash']
self.fee_calculator = FeeCalculator(response['feeCalculator'])
self.fee_calculator = FeeCalculator(response['feeCalculator'])

def __repr__(self) -> str:
return f"BlockHash(blockhash={self.blockhash!r}, fee_calculator={self.fee_calculator!r})"
8 changes: 7 additions & 1 deletion solathon/core/types/cluster_node.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from typing import Optional, TypedDict


class ClusterNodeType(TypedDict):
'''
JSON Response type of Cluster Node Information received by RPC
Expand All @@ -12,15 +13,20 @@ class ClusterNodeType(TypedDict):
featureSet: Optional[int]
shredVersion: Optional[int]


class ClusterNode:
'''
Convert Cluster Node Information JSON to Class
'''

def __init__(self, response: ClusterNodeType) -> None:
self.pubkey = response['pubkey']
self.gossip = response['gossip']
self.tpu = response['tpu']
self.rpc = response['rpc']
self.version = response['version']
self.feature_set = response['featureSet']
self.shred_version = response['shredVersion']
self.shred_version = response['shredVersion']

def __repr__(self) -> str:
return f"ClusterNode(pubkey={self.pubkey!r}, tpu={self.tpu!r}, rpc={self.rpc!r}, version={self.version!r},, shred_version={self.shred_version!r})"
12 changes: 11 additions & 1 deletion solathon/core/types/epoch.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,12 @@ class EpochType(TypedDict):
slotsInEpoch: int
transactionCount: Optional[int]


class Epoch:
'''
Convert Epoch Information JSON to Class
'''

def __init__(self, response: EpochType) -> None:
self.epoch = response['epoch']
self.absolute_slot = response['absoluteSlot']
Expand All @@ -24,6 +26,10 @@ def __init__(self, response: EpochType) -> None:
self.slots_in_epoch = response['slotsInEpoch']
self.transaction_count = response['transactionCount']

def __repr__(self) -> str:
return f"Epoch(epoch={self.epoch!r}, absolute_slot={self.absolute_slot!r}, block_height={self.block_height!r}, slot_index={self.slot_index!r})"


class EpochScheduleType(TypedDict):
'''
JSON Response type of Epoch Schedule Information received by RPC
Expand All @@ -34,14 +40,18 @@ class EpochScheduleType(TypedDict):
firstNormalEpoch: int
firstNormalSlot: int


class EpochSchedule:
'''
Convert Epoch Schedule Information JSON to Class
'''

def __init__(self, response: EpochScheduleType) -> None:
self.slots_per_epoch = response['slotsPerEpoch']
self.leader_schedule_slot_offset = response['leaderScheduleSlotOffset']
self.warmup = response['warmup']
self.first_normal_epoch = response['firstNormalEpoch']
self.first_normal_slot = response['firstNormalSlot']


def __repr__(self) -> str:
return f"EpochSchedule(slots_per_epoch={self.slots_per_epoch!r}, leader_schedule_slot_offset={self.leader_schedule_slot_offset!r}, warmup={self.warmup!r})"
Loading

0 comments on commit 618f0e5

Please # to comment.