Merge pull request #10 from developmentseed/osmnx-tool

🎉 add osmnx tool to query geometry attributes from OSM & road networks.

Author: srmsoumya

.gitignore

@@ -158,3 +158,6 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+
+# streamlit cache
+cache/

app.py

@@ -0,0 +1,83 @@
+import os
+
+import osmnx as ox
+import geopandas as gpd
+
+from langchain.chat_models import ChatOpenAI
+from langchain.tools import Tool, DuckDuckGoSearchRun
+
+from tools.mercantile_tool import MercantileTool
+from tools.geopy.geocode import GeopyGeocodeTool
+from tools.geopy.distance import GeopyDistanceTool
+from tools.osmnx.geometry import OSMnxGeometryTool
+from tools.osmnx.network import OSMnxNetworkTool
+from agents.l4m_agent import base_agent
+
+import geopandas as gpd
+import streamlit as st
+import folium
+from streamlit_folium import folium_static
+
+
+def get_llm():
+    llm = ChatOpenAI(
+        temperature=0,
+        openai_api_key=os.environ["OPENAI_API_KEY"],
+        model_name="gpt-3.5-turbo",
+    )
+    return llm
+
+
+def get_agent(llm, name="structured-chat-zero-shot-react-description"):
+    # define a set of tools the agent has access to for queries
+    duckduckgo_tool = Tool(
+        name="DuckDuckGo",
+        description="Use this tool to answer questions about current events and places. \
+        Please ask targeted questions.",
+        func=DuckDuckGoSearchRun().run,
+    )
+    geocode_tool = GeopyGeocodeTool()
+    distance_tool = GeopyDistanceTool()
+    mercantile_tool = MercantileTool()
+    geometry_tool = OSMnxGeometryTool()
+    network_tool = OSMnxNetworkTool()
+
+    tools = [
+        duckduckgo_tool,
+        geocode_tool,
+        distance_tool,
+        mercantile_tool,
+        geometry_tool,
+        network_tool,
+    ]
+
+    agent = base_agent(llm, tools, name=name)
+    return agent
+
+
+def run_query(agent, query):
+    response = agent(query)
+    return response
+
+
+def show_on_map(response):
+    gdf = response["output"]
+    center = gdf.centroid.iloc[0]
+    folium_map = folium.Map(location=[center.y, center.x], zoom_start=12)
+    folium.GeoJson(gdf).add_to(folium_map)
+    folium_static(folium_map)
+
+
+st.title("LLLLM")
+
+query = st.text_input(
+    "Ask me stuff about the flat world: ",
+    placeholder="Find all the hospitals in Bangalore",
+)
+
+if st.button("Submit", key="submit", type="primary"):
+    llm = get_llm()
+    agent = get_agent(llm)
+    response = run_query(agent, query)
+    show_on_map(response)
+    st.success("Great, you have the results plotted on the map.")

environment.yaml

@@ -4,6 +4,7 @@ channels:
 dependencies:
   - python=3
   - pip
+  - osmnx=1.3.1
   - pip:
     - openai
     - langchain
@@ -12,3 +13,5 @@ dependencies:
     - geopy
     - ipywidgets
     - jupyterlab
+    - streamlit
+    - streamlit-folium

nbs/23-05-26_osmnx-tool.ipynb

@@ -0,0 +1,32 @@
+from typing import Type, Dict
+
+import osmnx as ox
+import geopandas as gpd
+from pydantic import BaseModel, Field
+from langchain.tools import BaseTool
+
+
+class PlaceWithTags(BaseModel):
+    "Name of a place and tags in OSM."
+
+    place: str = Field(..., description="name of a place")
+    tags: Dict[str, str] = Field(..., description="open street maps tags")
+
+
+class OSMnxGeometryTool(BaseTool):
+    """Custom tool to query geometries from OSM."""
+
+    name: str = "geometry"
+    args_schema: Type[BaseModel] = PlaceWithTags
+    description: str = "Use this tool to get geometry of different features of a place like building footprints, parks, lakes, hospitals, schools etc. \
+    Pass the name of the place & relevant tags of Open Street Map as args."
+    return_direct = True
+
+    def _run(self, place: str, tags: Dict[str, str]) -> gpd.GeoDataFrame:
+        gdf = ox.geometries_from_place(place, tags)
+        gdf = gdf[gdf["geometry"].type.isin({"Polygon", "MultiPolygon"})]
+        gdf = gdf[["name", "geometry"]].reset_index(drop=True).head(100)
+        return gdf
+
+    def _arun(self, place: str):
+        raise NotImplementedError

tools/osmnx/geometry.py

@@ -0,0 +1,34 @@
+from typing import Type, Dict
+
+import osmnx as ox
+from osmnx import utils_graph
+import geopandas as gpd
+from pydantic import BaseModel, Field
+from langchain.tools import BaseTool
+
+
+class PlaceWithNetworktype(BaseModel):
+    "Name of a place on the map"
+    place: str = Field(..., description="name of a place on the map")
+    network_type: str = Field(
+        ..., description="network type: one of walk, bike, drive or all"
+    )
+
+
+class OSMnxNetworkTool(BaseTool):
+    """Custom tool to query road networks from OSM."""
+
+    name: str = "network"
+    args_schema: Type[BaseModel] = PlaceWithNetworktype
+    description: str = "Use this tool to get road network of a place. \
+    Pass the name of the place & type of road network i.e walk, bike, drive or all."
+    return_direct = True
+
+    def _run(self, place: str, network_type: str) -> gpd.GeoDataFrame:
+        G = ox.graph_from_place(place, network_type=network_type, simplify=True)
+        network = utils_graph.graph_to_gdfs(G, nodes=False)
+        network = network[["name", "geometry"]].reset_index(drop=True).head(100)
+        return network
+
+    def _arun(self, place: str):
+        raise NotImplementedError