Skip to main content
4 of 5
added 8 characters in body
Asta
  • 1
  • 1

when using call function in web3.py im getting error

Error: Could not transact with/call contract function, is contract deployed correctly and chain synced?

import streamlit as st
from web3 import Web3
import hashlib
import json

# Connect to local Ethereum node (Ganache)
ganache_url = "http://127.0.0.1:7545"  # Adjust if needed
web3 = Web3(Web3.HTTPProvider(ganache_url))

# Check if connection is successful
if not web3.is_connected():
    st.error("Failed to connect to the Ethereum node.")
    st.stop()

# Set the default account (first account in Ganache)
web3.eth.default_account = web3.eth.accounts[0]

# Load ABI from abi.json
def load_abi(file_path):
    with open(file_path, 'r') as abi_file:
        return json.load(abi_file)

# Load ABI and Contract Address (Replace with your actual contract address)
contract_abi = load_abi('abi.json')
contract_address = "0x5B38Da6a701c568545dCfcB03FcB875f56beddC4"  # Replace with your contract address

# Create a contract object
contract = web3.eth.contract(address=contract_address, abi=contract_abi)

# Streamlit UI
st.title("Deepfake Detection - Blockchain Integration")
st.write("Latest block number:", web3.eth.block_number)

def to_bytes32(hexstr):
    """ Convert a hex string to a bytes32 format. """
    return Web3.to_bytes(hexstr=hexstr)

def hash_file(file):
    """ Compute the SHA-256 hash of a file. """
    file_bytes = file.read()
    return hashlib.sha256(file_bytes).hexdigest()

# Upload video file
uploaded_file = st.file_uploader("Upload a video file", type=["mp4", "avi", "mov"])

if uploaded_file:
    video_hash = hash_file(uploaded_file)
    st.write(f"Video Hash: {video_hash}")

    option = st.selectbox("Choose an action", ["Check Hash", "Store Result", "Retrieve Result"])

    if option == "Check Hash":
        if st.button("Check"):
            if video_hash:
                try:
                    is_present = contract.functions.isVideoDetected(to_bytes32(video_hash)).call()
                    if is_present:
                        st.success("The hash is detected as a deepfake.")
                    else:
                        st.warning("The hash is not detected as a deepfake.")
                except Exception as e:
                    st.error(f"Error: {str(e)}")
            else:
                st.error("Please enter a valid video hash.")

    elif option == "Store Result":
        result = st.number_input("Deepfake detection result (0 = Not Detected, >0 = Detected)", min_value=0)
        if st.button("Store"):
            if video_hash:
                try:
                    tx_hash = contract.functions.updateVideoResult(to_bytes32(video_hash), result).transact({'from': web3.eth.default_account})
                    web3.eth.wait_for_transaction_receipt(tx_hash)
                    st.success("Result stored successfully on the blockchain.")
                except Exception as e:
                    st.error(f"Error: {str(e)}")
            else:
                st.error("Please enter a valid video hash.")

    elif option == "Retrieve Result":
        if st.button("Retrieve"):
            if video_hash:
                try:
                    result = contract.functions.getVideoResult(to_bytes32(video_hash)).call()
                    st.success(f"Deepfake Detection Result: {result}")
                except Exception as e:
                    st.error(f"Error: {str(e)}")
            else:
                st.error("Please enter a valid video hash.")
else:
    st.info("Please upload a video file to proceed.")

solidity code is as:

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.4;

contract DeepfakeDetection {

    // State variable with underscore prefix for clarity
    mapping(bytes32 => uint256) private _videoHashToResult; // Mapping to store results by video hash

    // Event for logging results
    event VideoResultUpdated(bytes32 indexed videoHash, uint256 result);

    // Function to get video result
    function getVideoResult(bytes32 videoHash) public view returns (uint256) {
        return _videoHashToResult[videoHash];
    }

    // Function to update video result with gas optimization
    function updateVideoResult(bytes32 videoHash, uint256 newResult) public {
        // Cache the state variable in memory to reduce SLOAD costs
        uint256 cachedResult = _videoHashToResult[videoHash];

        // Check if the new result is different to avoid unnecessary storage writes
        if (cachedResult != newResult) {
            _videoHashToResult[videoHash] = newResult;

            // Emit an event to log the update
            emit VideoResultUpdated(videoHash, newResult);
        }
    }

    // Function to check video detection status and return deepfakeness level
    function isVideoDetected(bytes32 videoHash) public view returns (bool isDetected, uint256 result) {
        // Get the result from the mapping
        uint256 detectionResult = _videoHashToResult[videoHash];

        // Check if the video is present
        if (detectionResult > 0) {
            // Video is present, return detected status and result
            return (true, detectionResult);
        } else {
            // Video is not present, return not detected status and zero result
            return (false, 0);
        }
    }
}

web3.py part for easy debugging:

from web3 import Web3
import json

# Step 1: Connect to local Ethereum node (Ganache)
ganache_url = "http://127.0.0.1:7545"
web3 = Web3(Web3.HTTPProvider(ganache_url))

# Step 2: Verify the connection
if not web3.is_connected():
    print("Failed to connect to the Ethereum node.")
    exit()
else:
    print("Connected to Ethereum node successfully.")

# Step 3: Set the default account (first account in Ganache)
web3.eth.default_account = web3.eth.accounts[0]

# Step 4: Load ABI from a file
def load_abi(file_path):
    with open(file_path, 'r') as abi_file:
        return json.load(abi_file)

# Load the contract's ABI and address
contract_abi = load_abi('abi.json')
contract_address = "0x5B38Da6a701c568545dCfcB03FcB875f56beddC4"  # Replace with your contract address

# Step 5: Create a contract object
contract = web3.eth.contract(address=contract_address, abi=contract_abi)

# Helper Function: Convert hex string to bytes32
def to_bytes32(hexstr):
    if len(hexstr) != 64:
        raise ValueError("Hex string must be exactly 64 characters long for bytes32.")
    return Web3.to_bytes(hexstr=hexstr)

# Step 6: Define a hardcoded video hash (64-character hex string)
hardcoded_video_hash = "0x" + "a" * 64  # Replace with actual hash if needed
video_bytes32 = to_bytes32(hardcoded_video_hash[2:])

# Step 7: Test "Check Hash" functionality
try:
    is_detected, result = contract.functions.isVideoDetected(video_bytes32).call()
    print(f"Check Hash Function - is_detected: {is_detected}, result: {result}")
except Exception as e:
    print(f"Error during 'Check Hash' function call: {str(e)}")

# Step 8: Test "Store Result" functionality
try:
    hardcoded_result = 1  # Assume 1 indicates detected as deepfake
    tx_hash = contract.functions.updateVideoResult(video_bytes32, hardcoded_result).transact({'from': web3.eth.default_account})
    receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
    print(f"Store Result Function - Transaction hash: {tx_hash.hex()}")
    print(f"Transaction receipt: {receipt}")
except Exception as e:
    print(f"Error during 'Store Result' function call: {str(e)}")

# Step 9: Test "Retrieve Result" functionality
try:
    result = contract.functions.getVideoResult(video_bytes32).call()
    print(f"Retrieve Result Function - result: {result}")
except Exception as e:
    print(f"Error during 'Retrieve Result' function call: {str(e)}")

Asta
  • 1
  • 1