commit c72bd08e88d00e4a6ce22691ce850b1effd55f73
Author: Jon Michael Aanes <jonjmaa@gmail.com>
Date:   Thu Apr 3 22:10:39 2025 +0200

    Initial PBC client moved out of fin-depo

diff --git a/README.md b/README.md
new file mode 100644
index 0000000..e69de29
diff --git a/pbc_client/__init__.py b/pbc_client/__init__.py
new file mode 100644
index 0000000..2a4c706
--- /dev/null
+++ b/pbc_client/__init__.py
@@ -0,0 +1,129 @@
+import dataclasses
+import datetime
+import email.utils
+import json
+import logging
+from collections.abc import Mapping
+from decimal import Decimal
+from typing import Any
+
+import requests
+from frozendict import frozendict
+
+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:
+    # Very rough implementation
+    if address is None:
+        msg = 'Address must not be None'
+        raise TypeError(msg)
+    if address.endswith('a'):
+        return 'shards/Shard0/'
+    if address.endswith('2'):
+        return 'shards/Shard1/'
+    return 'shards/Shard2/'
+
+
+@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, Any] = frozendict(),
+        method='POST',
+    ) -> tuple[Any, 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, Any]]:
+        data: dict = {'path': []}
+
+        url: str = 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)
+