| |
| 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() |
|
|
