Spaces:
Build error
Build error
| import streamlit as st | |
| import requests | |
| import json | |
| import web3 | |
| abi = [{"anonymous":False,"inputs":[{"indexed":False,"internalType":"string","name":"ipfsHash","type":"string"}],"name":"Store","type":"event"},{"inputs":[],"name":"getHashes","outputs":[{"internalType":"string[]","name":"","type":"string[]"}],"stateMutability":"view","type":"function"},{"inputs":[{"internalType":"string","name":"ipfsHash","type":"string"}],"name":"storeHash","outputs":[],"stateMutability":"nonpayable","type":"function"}] | |
| # function that uploads the results to ipfs | |
| def upload_json_to_ipfs(data): | |
| try: | |
| url = "https://api.pinata.cloud/pinning/pinJSONToIPFS" | |
| print("starting upload to ipfs") | |
| # check for JWT secret | |
| if "PinataJWT" not in st.secrets: | |
| st.write("No JWT secret found, please add your JWT in a secret titled \"PinataJWT\"") | |
| return | |
| jwt_token = st.secrets["PinataJWT"].strip() | |
| headers = { | |
| "Authorization": f"Bearer {jwt_token}", | |
| "Content-Type": "application/json" | |
| } | |
| # Convert Python dictionary to JSON string | |
| response = requests.post(url, headers=headers, json=data) | |
| if response.status_code == 200: | |
| # Print the IPFS hash from the successful response | |
| ipfs_hash = response.json().get("IpfsHash") | |
| return ipfs_hash | |
| else: | |
| st.write(f"Failed to upload JSON. Status code: {response.status_code}") | |
| st.write(response.text) | |
| return None | |
| except Exception as e: | |
| st.write(f"Error uploading to Pinata: {e}") | |
| # function that uploads to blockchain | |
| def upload_to_blockchain(hash): | |
| print("starting blockchain upload") | |
| w3 = web3.Web3(web3.HTTPProvider(st.secrets["infura"])) | |
| # create an instance of our contract | |
| contract = w3.eth.contract(address=st.secrets["ContractAddress"], abi = abi) | |
| # Call your function: 11155111 is Sepolia's id | |
| call_function = contract.functions.storeHash(hash).build_transaction({"chainId": 11155111, | |
| "from": st.secrets["EthWallet"], | |
| "nonce": w3.eth.get_transaction_count(st.secrets["EthWallet"])}) | |
| # Sign transaction | |
| signed_tx = w3.eth.account.sign_transaction(call_function, private_key=st.secrets["pk"]) | |
| # Send transaction | |
| send_tx = w3.eth.send_raw_transaction(signed_tx.rawTransaction) | |
| # Wait for transaction receipt | |
| tx_receipt = w3.eth.wait_for_transaction_receipt(send_tx) | |
| print("ETH Hash:") | |
| print(tx_receipt.logs[0].transactionHash.hex()) | |
| return tx_receipt.logs[0].transactionHash.hex() | |
| def get_ipfs_hashes(): | |
| w3 = web3.Web3(web3.HTTPProvider(st.secrets["infura"])) | |
| # Create an instance of the contract | |
| contract = w3.eth.contract(address=st.secrets["ContractAddress"], abi=abi) | |
| # Call the getHashes function | |
| try: | |
| ipfs_hashes = contract.functions.getHashes().call() | |
| return ipfs_hashes | |
| except Exception as e: | |
| st.write(f"Error retrieving hashes: {e}") | |
| return [] | |
| def retreive_ipfs_hash_data(hashes): | |
| results = [] | |
| for ipfs_hash in hashes: | |
| url = f"https://gateway.pinata.cloud/ipfs/{ipfs_hash}" | |
| try: | |
| response = requests.get(url) | |
| if response.status_code == 200: | |
| data = response.json() | |
| results.append({"hash": ipfs_hash, "data": data}) | |
| else: | |
| results.append({"hash": ipfs_hash, "error": f"Failed to retrieve data (status {response.status_code})"}) | |
| except Exception as e: | |
| results.append({"hash": ipfs_hash, "error": str(e)}) | |
| return results | |
| # function that handles survey submission | |
| # sets up ipfs and blockchain | |
| def submission(survey_data): | |
| ipfs_hash = upload_json_to_ipfs(survey_data) | |
| if ipfs_hash: | |
| print("IPFS Upload Successful") | |
| print(ipfs_hash) | |
| upload_to_blockchain(ipfs_hash) | |
| total_number_pages = 3 | |
| placeholder_buttons = None | |
| Q2_radio_options = ["pizza","burgers"] | |
| # Function that records radio element changes | |
| def radio_change(element, state, key): | |
| st.session_state[state] = element.index(st.session_state[key]) # Setting previously selected option | |
| def multi_change(element, state, key): | |
| st.session_state[state] = [] | |
| for selected_option in st.session_state[key]: | |
| st.session_state[state].append(selected_option) | |
| # Function that disables the last button while data is uploaded to IPFS | |
| def button_disable(): | |
| st.session_state['disabled'] = True | |
| def answer_change(state, key): | |
| st.session_state[state] = st.session_state[key] | |
| st.set_page_config(page_title='IPFS-Based Survey',) | |
| st.title('Test Survey') | |
| st.markdown("<style>.row-widget.stButton {text-align: center;}</style>", unsafe_allow_html=True) | |
| st.markdown("<style>.big-font {font-size:24px;}</style>", unsafe_allow_html=True) | |
| if "current_page" not in st.session_state: | |
| st.session_state["current_page"] = 1 | |
| st.session_state["Q1"] = None | |
| st.session_state["Q2"] = None | |
| st.session_state["Q3"] = None | |
| st.session_state["disabled"] = False | |
| # Page 1; Video | |
| if st.session_state["current_page"] == 1: | |
| st.markdown("""<p class="big-font">Test</p>""", unsafe_allow_html=True) | |
| st.text_area(label = "How is your day", | |
| value= "" if st.session_state["Q1"] == None else st.session_state["Q1"], | |
| key = 'Q1_text', | |
| on_change = answer_change, | |
| args = ( "Q1", "Q1_text",)) | |
| st.markdown("""<style> div[class*="stText"] > label > div[data-testid="stMarkdownContainer"] > p {font-size: 18px;}</style> <br><br>""", unsafe_allow_html=True) | |
| st.radio(label = "what is your favorite food", | |
| options = Q2_radio_options, | |
| index = None if st.session_state["Q2"] == None else st.session_state["Q2"], | |
| key = 'Q2_radio', | |
| on_change = radio_change, | |
| args = (Q2_radio_options, "Q2", "Q2_radio",)) | |
| st.markdown("""<style> div[class*="stRadio"] > label > div[data-testid="stMarkdownContainer"] > p {font-size: 18px;}</style> <br><br>""", unsafe_allow_html=True) | |
| placeholder = st.empty() | |
| if st.button('Next', key='next_button_page_1'): | |
| all_answered = True | |
| if st.session_state["Q1"] == None or st.session_state["Q1"] == []: | |
| all_answered = False | |
| if st.session_state["Q2"] == None or st.session_state["Q2"] == []: | |
| all_answered = False | |
| if all_answered: | |
| st.session_state["current_page"] += 1 | |
| st.rerun() | |
| else: | |
| with placeholder.container(): | |
| st.warning("Please answer all the questions on this page.", icon="⚠️") | |
| st.progress(st.session_state["current_page"]/total_number_pages, text="Progress") | |
| elif st.session_state["current_page"] == 2: | |
| if st.session_state["Q3"] == None: | |
| st.session_state["Q3"] = 6 | |
| st.slider(label="scale of 1-10",min_value=1,max_value=10, | |
| value= st.session_state["Q3"], | |
| key = "Q3_slider", | |
| on_change = answer_change, | |
| args = ("Q3", "Q3_slider",)) | |
| st.markdown("""<style> div[class*="stSlider"] > label > div[data-testid="stMarkdownContainer"] > p {font-size: 18px;}</style> <br><br>""", unsafe_allow_html=True) | |
| placeholder = st.empty() | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| if st.button('Back'): | |
| st.session_state["current_page"] -= 1 | |
| st.rerun() | |
| with col2: | |
| if st.button('Next'): | |
| all_answered = True | |
| if st.session_state["Q3"] == None or st.session_state["Q3"] == []: | |
| all_answered = False | |
| if all_answered: | |
| st.session_state["current_page"] += 1 | |
| st.rerun() | |
| else: | |
| with placeholder.container(): | |
| st.warning("Please answer all the questions on this page.", icon="⚠️") | |
| st.progress(st.session_state["current_page"]/total_number_pages, text="Progress") | |
| elif st.session_state["current_page"] == 3: # Last Page | |
| st.markdown('<p class="big-font">Thank you for participating! <br> Click on the button below to submit your answers. </p>', unsafe_allow_html=True) | |
| st.button('Submit Responses', disabled = st.session_state["disabled"], on_click = button_disable) | |
| if st.session_state["disabled"]: | |
| with st.spinner(r"$\textsf{\normalsize Storing data on IPFS and Ethereum. This operation might take a few minutes. Please wait to receive your confirmation code!}$"): | |
| try: | |
| response = { | |
| "Q1": st.session_state["Q1"], | |
| "Q2": Q2_radio_options[st.session_state["Q2"]], | |
| "Q3": st.session_state["Q3"], | |
| } | |
| submission(response) | |
| except Exception as e: | |
| print(e) | |
| st.error(f'An error ocurred. Here is the error message: {e}', icon="🚨") | |
| if st.button('Back'): | |
| st.session_state["current_page"] -= 1 | |
| st.rerun() | |
| st.progress(st.session_state["current_page"]/total_number_pages, text="Progress") | |