Image mapping contract support (#282)

prediction_market_agent_tooling/abis/omen_thumbnailmapping.abi.json

@@ -0,0 +1,52 @@
+[
+  {
+    "type": "function",
+    "name": "get",
+    "inputs": [
+      {
+        "name": "marketAddress",
+        "type": "address",
+        "internalType": "address"
+      }
+    ],
+    "outputs": [
+      {
+        "name": "",
+        "type": "bytes32",
+        "internalType": "bytes32"
+      }
+    ],
+    "stateMutability": "view"
+  },
+  {
+    "type": "function",
+    "name": "remove",
+    "inputs": [
+      {
+        "name": "marketAddress",
+        "type": "address",
+        "internalType": "address"
+      }
+    ],
+    "outputs": [],
+    "stateMutability": "nonpayable"
+  },
+  {
+    "type": "function",
+    "name": "set",
+    "inputs": [
+      {
+        "name": "marketAddress",
+        "type": "address",
+        "internalType": "address"
+      },
+      {
+        "name": "image_hash",
+        "type": "bytes32",
+        "internalType": "bytes32"
+      }
+    ],
+    "outputs": [],
+    "stateMutability": "nonpayable"
+  }
+]

prediction_market_agent_tooling/gtypes.py

@@ -34,6 +34,7 @@
 USDC = NewType("USDC", float)
 DatetimeWithTimezone = NewType("DatetimeWithTimezone", datetime)
 ChainID = NewType("ChainID", int)
+IPFSCIDVersion0 = NewType("IPFSCIDVersion0", str)
 
 
 def usd_type(amount: Union[str, int, float]) -> USD:

prediction_market_agent_tooling/markets/omen/omen_contracts.py

@@ -13,6 +13,7 @@
     HexAddress,
     HexBytes,
     HexStr,
+    IPFSCIDVersion0,
     OmenOutcomeToken,
     TxParams,
     TxReceipt,
@@ -26,7 +27,12 @@
     ContractOnGnosisChain,
     abi_field_validator,
 )
-from prediction_market_agent_tooling.tools.web3_utils import xdai_to_wei
+from prediction_market_agent_tooling.tools.web3_utils import (
+    ZERO_BYTES,
+    byte32_to_ipfscidv0,
+    ipfscidv0_to_byte32,
+    xdai_to_wei,
+)
 
 
 class OmenOracleContract(ContractOnGnosisChain):
@@ -610,3 +616,61 @@ def withdraw(
         web3: Web3 | None = None,
     ) -> TxReceipt:
         return self.send(api_keys=api_keys, function_name="withdraw", web3=web3)
+
+
+class OmenThumbnailMapping(ContractOnGnosisChain):
+    # Contract ABI taken from built https://github.com/gnosis/labs-contracts.
+    abi: ABI = abi_field_validator(
+        os.path.join(
+            os.path.dirname(os.path.realpath(__file__)),
+            "../../abis/omen_thumbnailmapping.abi.json",
+        )
+    )
+    address: ChecksumAddress = Web3.to_checksum_address(
+        "0x5D8B7B619EcdE05B8A94C0a0E99E0A0727A0e2e7"
+    )
+
+    def get(
+        self,
+        market_address: ChecksumAddress,
+        web3: Web3 | None = None,
+    ) -> IPFSCIDVersion0 | None:
+        hash_bytes = HexBytes(
+            self.call("get", function_params=[market_address], web3=web3)
+        )
+        return byte32_to_ipfscidv0(hash_bytes) if hash_bytes != ZERO_BYTES else None
+
+    def get_url(
+        self,
+        market_address: ChecksumAddress,
+        web3: Web3 | None = None,
+    ) -> str | None:
+        hash_ = self.get(market_address, web3)
+        return f"https://ipfs.io/ipfs/{hash_}" if hash_ is not None else None
+
+    def set(
+        self,
+        api_keys: APIKeys,
+        market_address: ChecksumAddress,
+        image_hash: IPFSCIDVersion0,
+        web3: Web3 | None = None,
+    ) -> TxReceipt:
+        return self.send(
+            api_keys=api_keys,
+            function_name="set",
+            function_params=[market_address, ipfscidv0_to_byte32(image_hash)],
+            web3=web3,
+        )
+
+    def remove(
+        self,
+        api_keys: APIKeys,
+        market_address: ChecksumAddress,
+        web3: Web3 | None = None,
+    ) -> TxReceipt:
+        return self.send(
+            api_keys=api_keys,
+            function_name="remove",
+            function_params=[market_address],
+            web3=web3,
+        )

prediction_market_agent_tooling/tools/web3_utils.py

@@ -1,5 +1,7 @@
+import binascii
 from typing import Any, Optional, TypeVar
 
+import base58
 import tenacity
 from eth_account import Account
 from eth_typing import URI
@@ -16,6 +18,7 @@
     HexAddress,
     HexBytes,
     HexStr,
+    IPFSCIDVersion0,
     PrivateKey,
     xDai,
     xdai_type,
@@ -287,3 +290,22 @@ def send_xdai_to(
         web3, tx_params_new, from_private_key, timeout
     )
     return receipt_tx
+
+
+def ipfscidv0_to_byte32(cid: IPFSCIDVersion0) -> HexBytes:
+    """
+    Convert ipfscidv0 to 32 bytes.
+    Modified from https://github.com/emg110/ipfs2bytes32/blob/main/python/ipfs2bytes32.py
+    """
+    decoded = base58.b58decode(cid)
+    sliced_decoded = decoded[2:]
+    return HexBytes(binascii.b2a_hex(sliced_decoded).decode("utf-8"))
+
+
+def byte32_to_ipfscidv0(hex: HexBytes) -> IPFSCIDVersion0:
+    """
+    Convert 32 bytes hex to ipfscidv0.
+    Modified from https://github.com/emg110/ipfs2bytes32/blob/main/python/ipfs2bytes32.py
+    """
+    completed_binary_str = b"\x12 " + hex
+    return IPFSCIDVersion0(base58.b58encode(completed_binary_str).decode("utf-8"))

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "prediction-market-agent-tooling"
-version = "0.37.1"
+version = "0.38.0"
 description = "Tools to benchmark, deploy and monitor prediction market agents."
 authors = ["Gnosis"]
 readme = "README.md"
@@ -42,6 +42,7 @@ prompt-toolkit = "^3.0.43"
 safe-cli = "^1.0.0"
 langfuse = "^2.27.1"
 openai = { version = "^1.0.0", optional = true}
+base58 = "^2.1.1"
 
 [tool.poetry.extras]
 openai = ["openai"]

tests/tools/test_web3_utils.py

@@ -1,6 +1,11 @@
 from pydantic.types import SecretStr
 
-from prediction_market_agent_tooling.tools.web3_utils import private_key_to_public_key
+from prediction_market_agent_tooling.gtypes import IPFSCIDVersion0
+from prediction_market_agent_tooling.tools.web3_utils import (
+    byte32_to_ipfscidv0,
+    ipfscidv0_to_byte32,
+    private_key_to_public_key,
+)
 
 
 def test_private_key_to_public_key() -> None:
@@ -12,3 +17,13 @@ def test_private_key_to_public_key() -> None:
         SecretStr(ganache_private_key_example)
     )
     assert actual_public_key == ganache_public_key_example
+
+
+def test_ipfs_hash_conversion() -> None:
+    ipfs = IPFSCIDVersion0("QmRUkBx3FQHrMrt3bACh5XCSLwRjNcTpJreJp4p2qL3in3")
+
+    as_bytes32 = ipfscidv0_to_byte32(ipfs)
+    assert len(as_bytes32) == 32, "The length of the bytes32 should be 32"
+
+    as_ipfs = byte32_to_ipfscidv0(as_bytes32)
+    assert as_ipfs == ipfs, "The IPFS hash should be the same after conversion back"