"""Electrum/Fulcrum-based network provider for BCH."""
from bitcash.network.APIs.FulcrumProtocolAPI import FulcrumProtocolAPI
from typing_extensions import override
from cashscript_py.helpers.crypto import hash256
from cashscript_py.interfaces import Utxo
from cashscript_py.network.network_provider import (
BroadcastFailed,
Network,
NetworkProvider,
RequestFailed,
UnexpectedResponse,
)
[docs]
class ElectrumNetworkProvider(NetworkProvider):
"""NetworkProvider backed by bitcash FulcrumProtocolAPI.
Connects to specific servers per network and provides UTXO queries,
block height, raw transaction fetching, and broadcasting.
"""
_servers = {
Network.MAINNET: "bch.imaginary.cash",
Network.TESTNET3: "testnet.imaginary.cash",
Network.TESTNET4: "testnet4.imaginary.cash",
Network.CHIPNET: "chipnet.bch.ninja",
}
_port = 50002 # Default port used for Electrum servers
def __init__(self, network: Network = Network.MAINNET):
"""Initialize the provider for a given network.
Args:
network: Target network (mainnet/testnet/chipnet/...).
Raises:
ValueError: If the network is unsupported.
"""
server = self._get_server_for_network(network)
self.network_provider: FulcrumProtocolAPI = FulcrumProtocolAPI(f"{server}:{self._port}")
self._network = network
@property
@override
def network(self) -> Network:
"""Return the configured network."""
return self._network
def _get_server_for_network(self, network: Network) -> str:
"""Resolve the Electrum/Fulcrum server hostname for a given network.
Args:
network: Target network.
Returns:
Hostname string.
Raises:
ValueError: If the network is unsupported.
"""
if network not in self._servers:
raise ValueError(f"Unsupported network: {network}")
return self._servers[network]
[docs]
@override
async def get_utxos(self, address: str) -> list[Utxo]:
"""Fetch UTXOs for a CashAddress.
Args:
address: CashAddress string.
Returns:
List of Utxo objects.
Raises:
RequestFailed: If the provider returns an error.
UnexpectedResponse: If the provider returns a malformed response.
"""
try:
response = self.network_provider.get_unspent(address)
if isinstance(response, list):
return [Utxo.from_unspent(unspent) for unspent in response]
raise UnexpectedResponse(response)
except Exception as e:
raise RequestFailed(e) from e
[docs]
@override
async def get_block_height(self) -> int:
"""Return the current block height.
Raises:
RequestFailed: If the provider returns an error.
UnexpectedResponse: If the provider returns a malformed response.
"""
try:
response = self.network_provider.get_blockheight()
if isinstance(response, int):
return response
raise UnexpectedResponse(response)
except Exception as e:
raise RequestFailed(e) from e
[docs]
@override
async def get_raw_transaction(self, txid: str) -> str:
"""Retrieve the raw transaction hex for a txid.
Args:
txid: Transaction ID (hex).
Returns:
Raw transaction hex.
Raises:
RequestFailed: If the provider returns an error.
UnexpectedResponse: If the provider returns a malformed response.
"""
try:
response = self.network_provider.get_raw_transaction(txid)["hex"]
if isinstance(response, str):
return response
raise UnexpectedResponse(response)
except Exception as e:
raise RequestFailed(e) from e
[docs]
@override
async def send_raw_transaction(self, tx_hex: str) -> str:
"""Broadcast a raw transaction to the network.
Args:
tx_hex: Hex-encoded transaction.
Returns:
Transaction ID computed for the broadcast transaction.
Raises:
BroadcastFailed: If the transaction was not accepted by the network.
"""
try:
result = self.network_provider.broadcast_tx(tx_hex)
except Exception as e:
raise BroadcastFailed(e) from e
if result is True:
tx_bytes = bytes.fromhex(tx_hex)
return hash256(tx_bytes)[::-1].hex()
# This should never happen
raise BroadcastFailed("broadcast_tx returned False")