Skip to content

Commit

Permalink
Fix
Browse files Browse the repository at this point in the history
  • Loading branch information
nielstron committed Sep 2, 2024
1 parent b829608 commit b0eb69c
Showing 1 changed file with 46 additions and 40 deletions.
86 changes: 46 additions & 40 deletions pycardano/backend/ogmios_v6.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,21 +2,22 @@
from typing import Optional, Union, Dict, List
from cachetools import Cache, LRUCache, TTLCache, func

from ogmios.client import Client
from ogmios.client import Client as OgmiosClient
from ogmios.datatypes import (
ProtocolParameters,
Era,
Tip,
Utxo,
Address,
TxOutputReference,
ProtocolParameters as OgmiosProtocolParameters,
Era as OgmiosEra,
Tip as OgmiosTip,
Utxo as OgmiosUtxo,
Address as OgmiosAddress,
TxOutputReference as OgmiosTxOutputReference,
)
from ogmios.utils import get_current_era, GenesisParameters
from ogmios.utils import get_current_era, GenesisParameters as OgmiosGenesisParameters

from pycardano.backend.kupo import KupoChainContextExtension
from pycardano.backend.base import (
ChainContext,
ProtocolParameters as pycProtocolParameters,
ProtocolParameters,
GenesisParameters,
)
from pycardano.network import Network
from pycardano.transaction import (
Expand All @@ -27,7 +28,7 @@
MultiAsset,
AssetName,
Asset,
Address as pyc_Address,
Address,
)
from pycardano.plutus import (
PLUTUS_V1_COST_MODEL,
Expand All @@ -47,12 +48,12 @@ class OgmiosV6ChainContext(ChainContext):
"""Ogmios chain context for use with PyCardano"""

_network: Network
_client: Client
_client: OgmiosClient
_service_name: str
_last_known_block_slot: int
_last_chain_tip_fetch: float
_genesis_param: Optional[GenesisParameters]
_protocol_param: Optional[ProtocolParameters]
_protocol_param: Optional[OgmiosProtocolParameters]
_utxo_cache: Cache
_datum_cache: Cache

Expand Down Expand Up @@ -86,28 +87,30 @@ def __init__(
)
self._datum_cache = LRUCache(maxsize=datum_cache_size)

Check warning on line 88 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L88

Added line #L88 was not covered by tests

def _query_current_era(self) -> Era:
with Client(self.host, self.port, self.secure) as client:
def _query_current_era(self) -> OgmiosEra:
with OgmiosClient(self.host, self.port, self.secure) as client:
return get_current_era(client)

Check warning on line 92 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L92

Added line #L92 was not covered by tests

def _query_current_epoch(self) -> int:
with Client(self.host, self.port, self.secure) as client:
with OgmiosClient(self.host, self.port, self.secure) as client:
epoch, _ = client.query_epoch.execute()
return epoch

Check warning on line 97 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L96-L97

Added lines #L96 - L97 were not covered by tests

def _query_chain_tip(self) -> Tip:
with Client(self.host, self.port, self.secure) as client:
def _query_chain_tip(self) -> OgmiosTip:
with OgmiosClient(self.host, self.port, self.secure) as client:
tip, _ = client.query_network_tip.execute()
return tip

Check warning on line 102 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L101-L102

Added lines #L101 - L102 were not covered by tests

def _query_utxos_by_address(self, address: pyc_Address) -> List[Utxo]:
with Client(self.host, self.port, self.secure) as client:
def _query_utxos_by_address(self, address: Address) -> List[OgmiosUtxo]:
with OgmiosClient(self.host, self.port, self.secure) as client:
utxos, _ = client.query_utxo.execute([address])
return utxos

Check warning on line 107 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L106-L107

Added lines #L106 - L107 were not covered by tests

def _query_utxos_by_tx_id(self, tx_id: str, index: int) -> List[Utxo]:
with Client(self.host, self.port, self.secure) as client:
utxos, _ = client.query_utxo.execute([TxOutputReference(tx_id, index)])
def _query_utxos_by_tx_id(self, tx_id: str, index: int) -> List[OgmiosUtxo]:
with OgmiosClient(self.host, self.port, self.secure) as client:
utxos, _ = client.query_utxo.execute(

Check warning on line 111 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L111

Added line #L111 was not covered by tests
[OgmiosTxOutputReference(tx_id, index)]
)
return utxos

Check warning on line 114 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L114

Added line #L114 was not covered by tests

def _is_chain_tip_updated(self):
Expand All @@ -128,15 +131,15 @@ def _fraction_parser(fraction: str) -> float:
return int(x) / int(y)

Check warning on line 131 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L130-L131

Added lines #L130 - L131 were not covered by tests

@property
def protocol_param(self) -> pycProtocolParameters:
def protocol_param(self) -> ProtocolParameters:
if not self._protocol_param or self._is_chain_tip_updated():
self._protocol_param = self._fetch_protocol_param()
return self._protocol_param

Check warning on line 137 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L136-L137

Added lines #L136 - L137 were not covered by tests

def _fetch_protocol_param(self) -> ProtocolParameters:
with Client(self.host, self.port, self.secure) as client:
with OgmiosClient(self.host, self.port, self.secure) as client:
protocol_parameters, _ = client.query_protocol_parameters.execute()
pyc_protocol_params = pycProtocolParameters(
pyc_protocol_params = ProtocolParameters(

Check warning on line 142 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L141-L142

Added lines #L141 - L142 were not covered by tests
min_fee_constant=protocol_parameters.min_fee_constant.lovelace,
min_fee_coefficient=protocol_parameters.min_fee_coefficient,
min_pool_cost=protocol_parameters.min_stake_pool_cost.lovelace,
Expand Down Expand Up @@ -185,22 +188,25 @@ def _fetch_protocol_param(self) -> ProtocolParameters:
@property
def genesis_param(self) -> GenesisParameters:
if not self._genesis_param or self._is_chain_tip_updated():
self._genesis_param = self._fetch_genesis_param()
# TODO transform to PyCardano GenesisParameters?
self._genesis_param = self._fetch_genesis_param() # type: ignore[assignment]

Check warning on line 192 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L192

Added line #L192 was not covered by tests

# Update the refetch interval if we haven't calculated it yet
if (
self._refetch_chain_tip_interval == DEFAULT_REFETCH_INTERVAL
and self._genesis_param is not None
and self._genesis_param.slot_length is not None
and self._genesis_param.active_slots_coefficient is not None
):
self._refetch_chain_tip_interval = self._genesis_param.slot_length.get(
"milliseconds"
) / eval(self._genesis_param.active_slots_coefficient)
return self._genesis_param
self._refetch_chain_tip_interval = (

Check warning on line 201 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L201

Added line #L201 was not covered by tests
self._genesis_param.slot_length
/ float(self._genesis_param.active_slots_coefficient)
)
return self._genesis_param # type: ignore[return-value]

Check warning on line 205 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L205

Added line #L205 was not covered by tests

def _fetch_genesis_param(self) -> GenesisParameters:
with Client(self.host, self.port, self.secure) as client:
return GenesisParameters(client, self._query_current_era())
def _fetch_genesis_param(self) -> OgmiosGenesisParameters:
with OgmiosClient(self.host, self.port, self.secure) as client:
return OgmiosGenesisParameters(client, self._query_current_era())

Check warning on line 209 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L209

Added line #L209 was not covered by tests

@property
def network(self) -> Network:
Expand All @@ -221,7 +227,7 @@ def _utxos(self, address: str) -> List[UTxO]:
if key in self._utxo_cache:
return self._utxo_cache[key]

Check warning on line 228 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L228

Added line #L228 was not covered by tests

utxos = self._utxos_ogmios(Address(address=address))
utxos = self._utxos_ogmios(OgmiosAddress(address=address))

Check warning on line 230 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L230

Added line #L230 was not covered by tests

self._utxo_cache[key] = utxos

Check warning on line 232 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L232

Added line #L232 was not covered by tests

Expand All @@ -231,7 +237,7 @@ def _check_utxo_unspent(self, tx_id: str, index: int) -> bool:
results = self._query_utxos_by_tx_id(tx_id, index)
return len(results) > 0

Check warning on line 238 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L237-L238

Added lines #L237 - L238 were not covered by tests

def _utxos_ogmios(self, address: pyc_Address) -> List[Utxo]:
def _utxos_ogmios(self, address: Address) -> List[OgmiosUtxo]:
"""Get all UTxOs associated with an address with Ogmios.
Args:
Expand All @@ -248,7 +254,7 @@ def _utxos_ogmios(self, address: pyc_Address) -> List[Utxo]:

return utxos

Check warning on line 255 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L255

Added line #L255 was not covered by tests

def _utxo_from_ogmios_result(self, utxo: Utxo) -> UTxO:
def _utxo_from_ogmios_result(self, utxo: OgmiosUtxo) -> UTxO:
"""Convert an Ogmios UTxO result to a PyCardano UTxO."""
tx_in = TransactionInput.from_primitive([utxo.tx_id, utxo.index])
lovelace_amount = utxo.value.get("ada").get("lovelace", 0)
Expand All @@ -269,7 +275,7 @@ def _utxo_from_ogmios_result(self, utxo: Utxo) -> UTxO:
datum = RawCBOR(bytes.fromhex(utxo.datum))

Check warning on line 275 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L275

Added line #L275 was not covered by tests
if set(utxo.value.keys()) == {"ada"}:
tx_out = TransactionOutput(

Check warning on line 277 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L277

Added line #L277 was not covered by tests
pyc_Address.from_primitive(utxo.address),
Address.from_primitive(utxo.address),
amount=lovelace_amount,
datum_hash=datum_hash,
datum=datum,
Expand All @@ -285,7 +291,7 @@ def _utxo_from_ogmios_result(self, utxo: Utxo) -> UTxO:
multi_assets.setdefault(policy, Asset())[token_name] = quantity

Check warning on line 291 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L289-L291

Added lines #L289 - L291 were not covered by tests

tx_out = TransactionOutput(

Check warning on line 293 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L293

Added line #L293 was not covered by tests
pyc_Address.from_primitive(utxo.address),
Address.from_primitive(utxo.address),
amount=Value(lovelace_amount, multi_assets),
datum_hash=datum_hash,
datum=datum,
Expand All @@ -303,13 +309,13 @@ def utxo_by_tx_id(self, tx_id: str, index: int) -> Optional[UTxO]:
def submit_tx_cbor(self, cbor: Union[bytes, str]):
if isinstance(cbor, bytes):
cbor = cbor.hex()

Check warning on line 311 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L311

Added line #L311 was not covered by tests
with Client(self.host, self.port, self.secure) as client:
with OgmiosClient(self.host, self.port, self.secure) as client:
client.submit_transaction.execute(cbor)

Check warning on line 313 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L313

Added line #L313 was not covered by tests

def evaluate_tx_cbor(self, cbor: Union[bytes, str]) -> Dict[str, ExecutionUnits]:
if isinstance(cbor, bytes):
cbor = cbor.hex()

Check warning on line 317 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L317

Added line #L317 was not covered by tests
with Client(self.host, self.port, self.secure) as client:
with OgmiosClient(self.host, self.port, self.secure) as client:
result, _ = client.evaluate_transaction.execute(cbor)
result_dict = {}

Check warning on line 320 in pycardano/backend/ogmios_v6.py

View check run for this annotation

Codecov / codecov/patch

pycardano/backend/ogmios_v6.py#L319-L320

Added lines #L319 - L320 were not covered by tests
for res in result:
Expand Down

0 comments on commit b0eb69c

Please sign in to comment.