# Contains a set of useful functions import pandas as pd from concrete import fhe from config import circuit_filepath, keys_filepath import streamlit as st import folium from streamlit_folium import st_folium def set_up_server(): """Load a server instance from a specified circuit file Raises: OSError: If there is an issue loading the FHE server. Returns: concrete.fhe.compilation.server.Server: A server instance loaded from the circuit file. """ try: server = fhe.Server.load(circuit_filepath) except OSError as e: raise OSError(f"Something went wrong with the circuit. Make sure that the circuit \ exists in {circuit_filepath}.If not run python generate_circuit.py.") from e return server def set_up_client(serialized_client_specs): """Generate a client instance from a specified circuit file Args: serialized_client_specs (bytes): A serialized client specs Returns: concrete.fhe.compilation.client.Client: A client instance created from the client specs """ client_specs = fhe.ClientSpecs.deserialize(serialized_client_specs) client = fhe.Client(client_specs) return client def display_encrypted(encrypted_object): """Display a truncated representation of an encrypted object as a hexadecimal string Args: encrypted_object (bytes): A serialized encrypted object to display Returns: str: A truncated hexadecimal representation of the encrypted object """ encoded_text = encrypted_object.hex() res = '...' + encoded_text[-10:] return res def compute_shortest_path(nodes_nb, client, server): """Calculate the shortest path between two nodes Args: nodes_nb (int): The number of nodes in the network client (concrete.fhe.compilation.client.Client): A client instance server (concrete.fhe.compilation.server.Server): A server instance Returns: List[bytes]: A list of encrypted values representing the path """ deserialized_origin = fhe.Value.deserialize(st.session_state['encrypted_origin']) deserialized_destination = fhe.Value.deserialize(st.session_state['encrypted_destination']) deserialized_evaluation_keys = fhe.EvaluationKeys.deserialize(st.session_state['evaluation_key']) client.keys.load_if_exists_generate_and_save_otherwise(keys_filepath) origin = st.session_state['origin_node'] destination = st.session_state['destination_node'] encrypted_path = [st.session_state['encrypted_origin'], ] o, d = deserialized_origin, deserialized_destination for _ in range(nodes_nb): if origin == destination: break o = server.run(o, d, evaluation_keys=deserialized_evaluation_keys) origin = client.decrypt(o) encrypted_path.append(o.serialize()) return encrypted_path def generate_path(shortest_path, roads): """Generate a path from a list of nodes using road data. Args: shortest_path (List[int]): A list of nodes representing the shortest path. roads (geopandas.DataFrame): A DataFrame containing the ways between the nodes. Returns: pandas.DataFrame: A DataFrame representing the road segments that form the shortest path. """ pairs_list = [] for i in range(len(shortest_path) - 1): current_element = shortest_path[i] next_element = shortest_path[i + 1] result = roads.groupby('way_id').filter(lambda x: set([current_element, next_element]).issubset(x['node_id'])) pairs_list.append(result) final_result = pd.concat(pairs_list) return final_result def decrypt_shortest_path(client): """Decrypt and store the shortest path in the session state Args: client (concrete.fhe.compilation.client.Client): A client instance """ client.keys.load_if_exists_generate_and_save_otherwise(keys_filepath) path = [] for enc_value in st.session_state['encrypted_shortest_path']: deserialized_result = fhe.Value.deserialize(enc_value) path.append(client.decrypt(deserialized_result)) st.session_state['decrypted_result'] = path def init_session(): """Initialize the Streamlit session and layout configuration. Returns: Streamlit.columns: A tuple of Streamlit columns for layout customization. """ st.set_page_config(layout="wide") if 'markers' not in st.session_state: st.session_state['markers'] = [] if 'server_side' not in st.session_state: st.session_state['server_side'] = [] if 'client_side' not in st.session_state: st.session_state['client_side'] = [] c1, c2, c3 = st.columns([1, 3, 1]) return c1, c2, c3 def add_marker(coordinates, name): """Add a marker with coordinates and a name to the Streamlit session. Args: coordinates (Point): The coordinates of the marker name (str): The name or label for the marker """ data = {'coordinates': coordinates, 'name': name} st.session_state['markers'].append(data) def display_map(nodes, returned_objects=None, path=None): """Display the map with nodes and optional markers and paths. Args: nodes (geopandas.DataFrame): A dataframe containing the nodes to display returned_objects (List[str], optional): Objects to be returned when interacting with the map. Defaults to None. path (pandas.DataFrame, optional): A DataFrame representing the road segments that form the shortest path. Defaults to None. Returns: Streamlit.FoliumMap: An interactive map displaying nodes, markers, and paths """ m = nodes.explore( color="red", marker_kwds=dict(radius=5, fill=True, name='node_id'), tooltip="node_id", tooltip_kwds=dict(labels=False), zoom_control=False, ) if 'markers' in st.session_state: for mrk in st.session_state['markers']: folium.Marker([mrk['coordinates'].y, mrk['coordinates'].x], popup=mrk['name'], tooltip=mrk['name']).add_to(m) if path is not None: path.explore( m=m, color="green", style_kwds = {"weight":5}, tooltip="name", popup=["name"], name="Quadratic-Paris", ) return st_folium(m, width=725, key="origin", returned_objects=returned_objects) def add_to_server_side(message): """Add a message to the server side of the view Args: message (str): The message to be added to the server side """ st.session_state['server_side'].append(message) def add_to_client_side(message): """Add a message to the client side of the view Args: message (str): The message to be added to the client side """ st.session_state['client_side'].append(message) def display_server_side(): """Display the messages stored in the server-side view. """ st.write("**Server-side**") for message in st.session_state['server_side']: st.write(message) def display_client_side(): """Display the messages stored in the client-side view. """ st.write("**Client-side**") for message in st.session_state['client_side']: st.write(message) def restart_session(): """Clear the session state to restart """ if st.button('Restart'): for key in st.session_state.items(): if key[0] != 'evaluation_key': del st.session_state[key[0]] st.rerun()