feat: Add EV charger and car parking agents (#132)

integrations/mobility-integrations/README.md

@@ -0,0 +1,62 @@
+# 🚗 uAgent Mobility Integrations Examples 🚴
+
+This repository contains examples of mobility integrations using two agents: `ev_charger` and `geoapi_car_parking`.
+
+1. `ev_charger`: This agent takes latitude, longitude, and a miles radius as input and returns available EV chargers in that region within the given radius. It utilizes the [OpenChargeMap](https://openchargemap.org/site/develop/api) API to retrieve the desired results.
+
+2. `geoapi_car_parking`: This agent takes latitude, longitude, and a miles radius as input and returns available parking spaces within that radius. The agent leverages [Geoapify](https://www.geoapify.com/) to fetch the desired results.
+
+## Getting Started 🚀
+
+To use these agents, follow the steps below:
+
+### Step 1: Obtain API Keys 🔑
+
+Before running the agents, you need to obtain the required API keys:
+
+#### OPENCHARGEMAP_API_KEY 🔌
+
+1. Visit the OpenChargeMap API website: https://openchargemap.org/site/develop/api.
+2. If you don't have an account, create one by signing up.
+3. Once you are logged in, click on the MY PROFILE > my apps at the top.
+4. Click on REGISTER AN APPLICATION button.
+5. Fill out the required information in the form, including the purpose of using the API, and submit the request.
+6. Once approved, you will see your `OPENCHARGEMAP_API_KEY` under MY API KEYS.
+
+#### GEOAPI_API_KEY 🗺️
+
+1. Go to the Geoapify website: https://www.geoapify.com/.
+2. Sign in to your Google account or create a new one if you don't have an existing account.
+3. Once you are logged in, create a new project by clicking on the "Add a new project" button under the Projects section.
+4. Give your project a name and click "OK" to create the new project.
+5. Your `GEOAPI_API_KEY` will be generated. Copy this key and keep it secure, as it will be used to access Geoapify Projects and other services.
+
+### Step 2: Set Environment Variables 🌐
+
+Create a `.env` file in the `mobility-integrations` directory and export the obtained API keys as environment variables:
+
+```bash
+export OPENCHARGEMAP_API_KEY="{GET THE API KEY}"
+export GEOAPI_API_KEY="{GET THE API KEY}"
+```
+
+### Step 3: Install Dependencies ⚙️
+
+To use the environment variables from the `.env` file and install the project dependencies:
+
+```bash
+source .env
+poetry install
+```
+
+### Step 4: Run the Project 🏃
+
+To run the project and its agents:
+
+```bash
+cd src
+poetry shell
+python main.py
+```
+
+Now you have the agents up and running to perform mobility integrations using the provided APIs. Happy integrating! 🎉

integrations/mobility-integrations/pyproject.toml

@@ -0,0 +1,13 @@
+[tool.poetry]
+name = "mobility-integrations"
+version = "0.1.0"
+description = "A template for creating mobility integrations"
+authors = ["Laxmikant Tripathi <[email protected]>"]
+
+[tool.poetry.dependencies]
+python = ">=3.10,<3.12"
+uagents = "*"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

integrations/mobility-integrations/src/agents/ev_charger/ev_charger.py

@@ -0,0 +1,76 @@
+import os
+import uuid
+
+import requests
+from messages import EVRequest, KeyValue, UAgentResponse, UAgentResponseType
+from uagents import Agent, Context, Protocol
+from uagents.setup import fund_agent_if_low
+
+EV_SEED = os.getenv("EV_SEED", "ev charger service secret phrase")
+
+agent = Agent(
+    name="ev_adaptor",
+    seed=EV_SEED,
+)
+
+fund_agent_if_low(agent.wallet.address())
+
+OPENCHARGEMAP_API_KEY = os.environ.get("OPENCHARGEMAP_API_KEY", "")
+
+assert (
+    OPENCHARGEMAP_API_KEY
+), "OPENCHARGEMAP_API_KEY environment variable is missing from .env"
+
+OPENCHARGEMAP_API_URL = "https://api.openchargemap.io/v3/poi?"
+MAX_RESULTS = 100
+
+
+def get_ev_chargers(latitude: float, longitude: float, miles_radius: float) -> list:
+    """Return ev chargers available within given miles_readius of the latiture and longitude.
+    this information is being retrieved from https://api.openchargemap.io/v3/poi? API
+    """
+    response = requests.get(
+        url=OPENCHARGEMAP_API_URL
+        + f"maxresults={MAX_RESULTS}&latitude={latitude}&longitude={longitude}&distance={miles_radius}",
+        headers={"x-api-key": OPENCHARGEMAP_API_KEY},
+        timeout=5,
+    )
+    if response.status_code == 200:
+        return response.json()
+    return []
+
+
+ev_chargers_protocol = Protocol("EvChargers")
+
+
+@ev_chargers_protocol.on_message(model=EVRequest, replies=UAgentResponse)
+async def ev_chargers(ctx: Context, sender: str, msg: EVRequest):
+    ctx.logger.info(f"Received message from {sender}")
+    try:
+        ev_chargers = get_ev_chargers(msg.latitude, msg.longitude, msg.miles_radius)
+        request_id = str(uuid.uuid4())
+        conn_types = []
+        options = []
+        for idx, ev_station in enumerate(ev_chargers):
+            for conn in ev_station["Connections"]:
+                conn_types.append(conn["ConnectionType"]["Title"])
+                conn_type_str = ", ".join(conn_types)
+                option = f"""● EV charger: {ev_station['AddressInfo']['Title']} , located {round(ev_station['AddressInfo']['Distance'], 2)} miles from your location\n● Usage cost {ev_station['UsageCost']};\n● Type - {conn_type_str}"""
+
+            options.append(KeyValue(key=idx, value=option))
+        await ctx.send(
+            sender,
+            UAgentResponse(
+                options=options,
+                type=UAgentResponseType.SELECT_FROM_OPTIONS,
+                request_id=request_id,
+            ),
+        )
+    except Exception as exc:
+        ctx.logger.error(exc)
+        await ctx.send(
+            sender, UAgentResponse(message=str(exc), type=UAgentResponseType.ERROR)
+        )
+
+
+agent.include(ev_chargers_protocol)

integrations/mobility-integrations/src/agents/geopy_car_parking/geopy_car_parking.py

@@ -0,0 +1,123 @@
+import os
+import uuid
+
+import requests
+from messages import GeoParkingRequest, KeyValue, UAgentResponse, UAgentResponseType
+from uagents import Agent, Context, Protocol
+from uagents.setup import fund_agent_if_low
+
+GEOAPI_PARKING_SEED = os.getenv(
+    "GEOAPI_PARKING_SEED", "geoapi parking adaptor agent secret phrase"
+)
+agent = Agent(name="geoapi_parking_adaptor", seed=GEOAPI_PARKING_SEED)
+geoapi_parking_protocol = Protocol("Geoapi CarParking")
+
+fund_agent_if_low(agent.wallet.address())
+
+GEOAPI_API_KEY = os.getenv("GEOAPI_API_KEY", "")
+
+assert GEOAPI_API_KEY, "GEOAPI_API_KEY environment variable is missing from .env"
+
+PARKING_API_URL = "https://api.geoapify.com/v2/places?"
+
+
+def format_parking_data(api_response) -> list:
+    """
+    By taking the response from the API, this function formats the response
+    to be appropriate for displaying back to the user.
+    """
+    parking_data = []
+    parking_name = "Unknown Parking"
+    parking_capacity = ""
+    for place in api_response["features"]:
+        if "name" in place["properties"]:
+            parking_name = place["properties"]["name"]
+            address = place["properties"]["formatted"].split(",")[1::]
+            parking_address = "".join(list(address))
+        elif "formatted" in place["properties"]:
+            parking_address = place["properties"]["formatted"]
+        else:
+            continue
+        if "capacity" in place["properties"]["datasource"]["raw"]:
+            parking_capacity = (
+                f'{place["properties"]["datasource"]["raw"]["capacity"]} spaces'
+            )
+        elif "parking" in place["properties"]["datasource"]["raw"]:
+            parking_capacity = (
+                f'{place["properties"]["datasource"]["raw"]["parking"]} parking'
+            )
+        elif (
+            "access" in place["properties"]["datasource"]["raw"]
+            and place["properties"]["datasource"]["raw"]["access"] != "yes"
+        ):
+            continue
+        parking_data.append(
+            f"""● Car Parking: {parking_name} has {parking_capacity} at {parking_address}"""
+        )
+    return parking_data
+
+
+def get_parking_from_api(latitude, longitude, radius, max_r) -> list:
+    """
+    With all the user preferences, this function sends the request to the Geoapify Parking API,
+    which returns the response.
+    """
+    try:
+        response = requests.get(
+            url=f"{PARKING_API_URL}categories=parking&filter=circle:{longitude},{latitude},{radius}&bias=proximity:{longitude},{latitude}&limit={max_r}&apiKey={GEOAPI_API_KEY}",
+            timeout=60,
+        )
+        return response.json()
+    except Exception as exc:
+        print("Error: ", exc)
+        return []
+
+
+@geoapi_parking_protocol.on_message(model=GeoParkingRequest, replies=UAgentResponse)
+async def geoapi_parking(ctx: Context, sender: str, msg: GeoParkingRequest):
+    """
+    The function takes the request to search for parking in any location based on user preferences
+    and returns the formatted response to TAGI.
+    """
+    ctx.logger.info(f"Received message from {sender}")
+    try:
+        radius_in_meter = msg.radius * 1609
+        response = get_parking_from_api(
+            msg.latitude, msg.longitude, radius_in_meter, msg.max_result
+        )  # Sending user preferences to find nearby parking spaces.
+        response = format_parking_data(
+            response
+        )  # Sending the API response to be made appropriate to users.
+        request_id = str(uuid.uuid4())
+        if len(response) > 1:
+            option = f"""Here is the list of some Parking spaces nearby:\n"""
+            idx = ""
+            options = [KeyValue(key=idx, value=option)]
+            for parking in response:
+                option = parking
+                options.append(KeyValue(key=idx, value=option))
+            await ctx.send(
+                sender,
+                UAgentResponse(
+                    options=options,
+                    type=UAgentResponseType.SELECT_FROM_OPTIONS,
+                    request_id=request_id,
+                ),
+            )  # Sending the response bach to users.
+        else:
+            await ctx.send(
+                sender,
+                UAgentResponse(
+                    message="No options available for this context",
+                    type=UAgentResponseType.FINAL,
+                    request_id=request_id,
+                ),
+            )
+    except Exception as exc:
+        ctx.logger.error(exc)
+        await ctx.send(
+            sender, UAgentResponse(message=str(exc), type=UAgentResponseType.ERROR)
+        )
+
+
+agent.include(geoapi_parking_protocol)

integrations/mobility-integrations/src/main.py

@@ -0,0 +1,13 @@
+from agents.ev_charger.ev_charger import agent as ev_charger_agent
+from agents.geopy_car_parking.geopy_car_parking import agent as geopy_car_parking_agent
+from uagents import Bureau
+
+if __name__ == "__main__":
+    bureau = Bureau(endpoint="http://127.0.0.1:8000/submit", port=8000)
+    print(f"Adding Ev charger agent to Bureau: {ev_charger_agent.address}")
+    bureau.add(ev_charger_agent)
+    print(
+        f"Adding geopy car parking agent to Bureau: {geopy_car_parking_agent.address}"
+    )
+    bureau.add(geopy_car_parking_agent)
+    bureau.run()

integrations/mobility-integrations/src/messages/__init__.py

@@ -0,0 +1,3 @@
+from .ev_charger import EVRequest
+from .general import *
+from .geoapi_parking import GeoParkingRequest

integrations/mobility-integrations/src/messages/ev_charger.py

@@ -0,0 +1,17 @@
+from uagents import Model
+
+
+class EVRequest(Model):
+    """
+    Represents an Electric Vehicle (EV) Charging Request.
+
+    Attributes:
+        latitude (float): The latitude of the location where EV charging is requested.
+        longitude (float): The longitude of the location where EV charging is requested.
+        miles_radius (float): The search radius (in miles) to find available EV charging stations
+                             around the specified location.
+    """
+
+    latitude: float
+    longitude: float
+    miles_radius: float

integrations/mobility-integrations/src/messages/general.py

@@ -0,0 +1,51 @@
+from enum import Enum
+from typing import List, Optional
+
+from uagents import Model
+
+
+class UAgentResponseType(Enum):
+    """
+    Enumeration representing different types of responses from the User Agent.
+
+    Attributes:
+        ERROR (str): Represents an error response type.
+        SELECT_FROM_OPTIONS (str): Represents a response type where the user needs to select from given options.
+        FINAL_OPTIONS (str): Represents a response type where the user has selected the final option.
+    """
+
+    ERROR = "error"
+    SELECT_FROM_OPTIONS = "select_from_options"
+    FINAL_OPTIONS = "final_options"
+
+
+class KeyValue(Model):
+    """
+    Represents a key-value pair.
+
+    Attributes:
+        key (str): The key of the pair.
+        value (str): The value associated with the key.
+    """
+
+    key: str
+    value: str
+
+
+class UAgentResponse(Model):
+    """
+    Represents a User Agent Response.
+
+    Attributes:
+        type (UAgentResponseType): The type of the response.
+        agent_address (str, optional): The address of the agent to which the response is directed.
+        message (str, optional): A message related to the response.
+        options (List[KeyValue], optional): A list of key-value pairs representing response options.
+        request_id (str, optional): An optional identifier for the corresponding request.
+    """
+
+    type: UAgentResponseType
+    agent_address: Optional[str]
+    message: Optional[str]
+    options: Optional[List[KeyValue]]
+    request_id: Optional[str]

integrations/mobility-integrations/src/messages/geoapi_parking.py

@@ -0,0 +1,18 @@
+from uagents import Model
+
+
+class GeoParkingRequest(Model):
+    """
+    Represents a Geo Parking Request.
+
+    Attributes:
+        latitude (float): The latitude of the location for which parking is requested.
+        longitude (float): The longitude of the location for which parking is requested.
+        radius (int): The search radius (in miles) to find parking spots around the specified location.
+        max_result (int, optional): The maximum number of parking spots to be returned. Default is 5.
+    """
+
+    latitude: float
+    longitude: float
+    radius: int
+    max_result: int = 5