import dataclasses import datetime import email.utils import json import logging from collections.abc import Mapping from decimal import Decimal import fin_defs import requests from frozendict import frozendict from .data import DepoGroup, DepoSingle logger = logging.getLogger(__name__) # mainnet: https://reader.partisiablockchain.com # testnet: https://node1.testnet.partisiablockchain.com HOSTNAME = 'reader.partisiablockchain.com' URL_ACCOUNT_PLUGIN = 'https://{hostname}/{shard}blockchain/accountPlugin/local' URL_ACCOUNT_PLUGIN_GLOBAL = 'https://{hostname}/{shard}blockchain/accountPlugin/global' URL_CONTRACT_STATE = 'https://{hostname}/{shard}blockchain/contracts/{address}?requireContractState=false' MPC_DECIMALS = 10000 def shard_id_for_address(address: str) -> str: assert address is not None, 'missing address' if address.endswith('a'): return 'shards/Shard0/' # TODO elif address.endswith('2'): return 'shards/Shard1/' # TODO else: return 'shards/Shard2/' # TODO @dataclasses.dataclass(frozen=True) class Balances: update_time: datetime.datetime mpc: Decimal byoc: Mapping[str, Decimal] @dataclasses.dataclass(frozen=True) class PbcClient: session: requests.Session def get_json( self, url: str, data: Mapping[str, str] = frozendict(), method='POST', ) -> tuple[dict, datetime.datetime]: headers = { 'Content-Type': 'application/json', 'Accept': 'application/json', } response = self.session.request( method, url, headers=headers, data=json.dumps(data), ) response.raise_for_status() date_text = response.headers.get('last-modified') or response.headers.get( 'date', ) date = email.utils.parsedate_to_datetime(date_text) json_data = response.json() if json_data is None: msg = 'No result data for ' + url raise Exception(msg) return (json_data, date) def determine_coins(self) -> list[dict[str, str]]: data: dict = {'path': []} url = URL_ACCOUNT_PLUGIN_GLOBAL.format( hostname=HOSTNAME, shard='', ) json_data, date = self.get_json(url, data=data) return json_data['coins']['coins'] def get_account_balances(self, address: str) -> Balances: coins = self.determine_coins() url = URL_ACCOUNT_PLUGIN.format( hostname=HOSTNAME, shard=shard_id_for_address(address), ) data: dict = { 'path': [ {'type': 'field', 'name': 'accounts'}, {'type': 'avl', 'keyType': 'BLOCKCHAIN_ADDRESS', 'key': address}, ], } account_data, date = self.get_json(url, data=data) byoc: dict[str, Decimal] = {} mpc = Decimal(account_data['mpcTokens']) / MPC_DECIMALS for coin_idx, amount_data in enumerate(account_data['accountCoins']): coin_data = coins[coin_idx] byoc_balance = Decimal(amount_data['balance']) denominator = Decimal(coin_data['conversionRate']['denominator']) native_balance = byoc_balance / denominator byoc[coin_data['symbol']] = native_balance del coin_idx, coin_data return Balances(date, mpc, byoc) def get_contract_state(self, address: str) -> tuple[dict, datetime.datetime]: url = URL_CONTRACT_STATE.format( hostname=HOSTNAME, shard=shard_id_for_address(address), address=address, ) data: dict = {'path': []} return self.get_json(url, data=data) BYOC_ASSETS = { 'ETH': fin_defs.WELL_KNOWN_SYMBOLS['ETH'], 'POLYGON_USDC': fin_defs.WELL_KNOWN_SYMBOLS['USDC'], 'BNB': fin_defs.WELL_KNOWN_SYMBOLS['BNB'], 'MATIC': fin_defs.WELL_KNOWN_SYMBOLS['MATIC'], } class PartisiaBlockchainAccountDepoFetcher: def __init__(self, session, blockchain_address: str): assert session is not None, 'missing session' assert blockchain_address is not None, 'missing blockchain_address' self.client = PbcClient(session) self.blockchain_address = blockchain_address def get_depo(self) -> DepoGroup: balances = self.client.get_account_balances(self.blockchain_address) depo_mpc = DepoSingle('Native', balances.update_time, {fin_defs.MPC: balances.mpc}) depo_byoc = DepoSingle( 'BYOC', balances.update_time, {BYOC_ASSETS[k]: v for k, v in balances.byoc.items()}, ) return DepoGroup( 'Partisia Blockchain', balances.update_time, [depo_mpc, depo_byoc], )