from concrete import fhe from config import circuit_filepath, number_of_neighbors from shapely.geometry import Point import geopandas as gpd import streamlit as st import folium from streamlit_folium import st_folium 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 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 transform_point(longitude, latitude): """Transform coordinates into an integer to be processed by the FHE circuit Args: longitude (float): longitude of the point latitude (float): latitude of the point Returns: int, int: integers to be processed by the FHE circuit """ gdf = gpd.GeoDataFrame({"geometry": [Point(longitude, latitude)]}, crs="EPSG:4326") gdf = gdf.to_crs("EPSG:2154") x, y = gdf.geometry.iloc[0].x, gdf.geometry.iloc[0].y x = int(x) % 10000 y = int(y) % 10000 return x, y def process_result(rest, result): """Add the nearest restaurants in the map and in the client view Args: rest (geopandas.DataFrame): list of restaurants result (list[(int, int)]): list of the nearest neighbors returned by the algorithm """ add_to_client_side( f"The {number_of_neighbors} closest restaurant to your location are:" ) for index, res in enumerate(result): mask1 = ( rest["geometry"] .to_crs("epsg:2154") .apply(lambda geom: int(geom.x) % 10000 == res[0]) ) mask2 = ( rest["geometry"] .to_crs("epsg:2154") .apply(lambda geom: int(geom.y) % 10000 == res[1]) ) final_mask = mask1 & mask2 result_df = rest[final_mask] restaurant_info = f"{result_df.name.iloc[0]}, {result_df.cuisine.iloc[0]}" add_marker(result_df.geometry, restaurant_info) add_to_client_side(f"{index+1}. {restaurant_info}.") 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(restaurants, returned_objects=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. Returns: Streamlit.FoliumMap: An interactive map displaying nodes and markers """ if "decrypted_result" in st.session_state: m = restaurants.explore( scheme="naturalbreaks", tooltip="name", popup=["name"], name="Quadratic-Paris", color="red", marker_kwds=dict(radius=5, fill=True, name="node_id"), ) else: m = restaurants.explore( scheme="naturalbreaks", tooltip="name", popup=["name"], name="Quadratic-Paris", ) if "position" in st.session_state: position = st.session_state["position"] folium.Marker( [position.y, position.x], popup="Starting point", tooltip="Starting point" ).add_to(m) 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"], icon=folium.Icon(color="black", icon_color="#FFFF00"), ).add_to(m) 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()