Initial commit

.gitignore

@@ -0,0 +1,160 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  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/

app.py

@@ -0,0 +1,114 @@
+import os
+import time
+
+import gradio as gr
+import openai
+from dotenv import load_dotenv, find_dotenv
+from simpleaichat import AIChat
+
+from main import weather, search, MODEL
+from utils.tts import TTS, voices
+
+load_dotenv(find_dotenv())
+openai.api_key = os.getenv("OPENAI_API_KEY")
+
+
+def transcribe(audio_file, state=""):
+    time.sleep(5)
+    if audio_file is None:
+        return None
+    prompt = (
+        "The author of this tool is Somto Muotoe. "
+        "Friends: Ire Ireoluwa Adedugbe, Biola Aderiye, Jelson, Raj, Akshay."
+        "Umm, let me think like, hmm... Okay, here's what I'm, like, "
+        "thinking. "
+    )
+    with open(audio_file, "rb") as f:
+        response = openai.Audio.transcribe("whisper-1", f, prompt=prompt)
+        text = response["text"]
+    state += text
+    return state, state
+
+
+def chat_with_gpt(prompt, ai_state):
+    if ai_state is None:
+        params = {"temperature": 0.0, "max_tokens": 200}
+        system_prompt = (
+            "You are a confidante whose response is curt and concise."
+            "You can use tools to give real-time updates on weather and search the internet.  "
+            "Answer all questions empathetically, and ALWAYS ask follow-up questions."
+            "Do NOT say Confidante in any response."
+            "You must TRUST the provided context to inform your response."
+            # "If a question does not make any sense, or is not factually coherent, explain why "
+            # "instead of answering something not correct. If you don't know the answer to a question, "
+            # "please don't share false information."
+        )
+        ai = AIChat(
+            params=params, model=MODEL, system=system_prompt, save_messages=True
+        )
+    else:
+        ai = ai_state
+    tools = [weather, search]
+
+    response = ai(prompt, tools=tools)
+    text_response = response["response"]
+    print(text_response)
+    return text_response, ai
+
+
+def tts(text, voice_id):
+    # Generate audio from the text response
+    tts_ = TTS(voice_id)
+    audio_data = tts_.generate(text=text)
+    return audio_data
+
+
+def transcribe_and_chat(audio_file, voice, history, ai_state):
+    if audio_file is None:
+        raise gr.Error("Empty audio file.")
+    voice_id = voices[voice]
+
+    text, text_state = transcribe(audio_file)
+    gpt_response, ai_state = chat_with_gpt(text, ai_state)
+    audio_data = tts(gpt_response, voice_id)
+
+    # Update the history with the new messages
+    history.append((text, gpt_response))
+
+    return history, audio_data, history, ai_state
+
+
+def clear_chat(history):
+    # Clear the chat history
+    history.clear()
+
+    # Clear the chat for the AIChat object
+    chat_with_gpt("", ai_state=None)
+
+    return history
+
+
+with gr.Blocks(title="JARVIS") as demo:
+    gr.Markdown(
+        "# Talk with GPT-4! You can get real-time weather updates, and can search Google."
+    )
+    audio_input = gr.Audio(source="microphone", type="filepath", visible=True)
+    gr.ClearButton(audio_input)
+
+    voice_select = gr.Radio(choices=list(voices.keys()), label="Voice", value="Bella")
+    history = gr.State(label="History", value=[])
+    ai_state = gr.State(label="AIChat", value=None)
+    # transcription = gr.Textbox(lines=2, label="Transcription")
+    chat_box = gr.Chatbot(label="Response")
+    response_audio = gr.Audio(label="Response Audio", autoplay=True)
+    gr.ClearButton(chat_box, value="Clear Chat")
+    # clear_chat_btn.click(clear_chat, inputs=history, outputs=history)
+
+    audio_input.stop_recording(
+        transcribe_and_chat,
+        inputs=[audio_input, voice_select, history, ai_state],
+        outputs=[chat_box, response_audio, history, ai_state],
+    )
+    audio_input.clear()
+
+demo.launch(server_port=8080, share=True)

environment.yml

@@ -0,0 +1,15 @@
+name: jarvis
+channels:
+  - conda-forge
+dependencies:
+  - python=3.10
+  - pip
+  - python-dotenv
+  - black
+  - pip:
+    - simpleaichat
+    - pyowm>3.0
+    - aiohttp
+    - openai
+    - gradio
+    - faster-whisper

main.py

@@ -0,0 +1,74 @@
+from functools import partial
+from os import getenv
+from pathlib import Path
+
+from dotenv import find_dotenv, load_dotenv
+from simpleaichat import AIChat
+
+from utils.google_serper import GoogleSerper
+from utils.weather_forecast import WeatherAPI
+
+load_dotenv(find_dotenv(raise_error_if_not_found=True))
+
+local_dir = Path(getenv("LOCAL_DIR"))
+
+OPENAI_API_KEY = getenv("OPENAI_API_KEY")
+MODEL = "gpt-4"
+# MODEL = "gpt-3.5-turbo-16k"
+# %%
+
+
+# This uses the Wikipedia Search API.
+# Results from it are nondeterministic, your mileage will vary.
+# def search(query):
+#     """Search the internet."""
+#     wiki_matches = wikipedia_search(query, n=3)
+#     return {"context": ", ".join(wiki_matches), "titles": wiki_matches}
+
+
+# def lookup(query):
+#     """Lookup more information about a topic."""
+#     page = wikipedia_search_lookup(query, sentences=3)
+#     return page
+
+
+def weather(query):
+    """Use this tool to get real-time temperature and weather forecast information"""
+    w = WeatherAPI()
+    weather_info = w.run(query)
+    return weather_info
+
+
+def search(query):
+    """Search the internet."""
+    g = GoogleSerper()
+    search_result = g.run(query)
+    return search_result
+
+
+if __name__ == "__main__":
+    params = {"temperature": 0.1, "max_tokens": 200}
+    system_prompt = (
+        "You are an assistant that likes to use emoticons. "
+        "You are also curt and concise with your responses."
+    )
+    ai = AIChat(console=False, id="chat_1")
+    ai.load_session("chat_session_3.json", id="chat_1")
+    ai.get_session()
+    # AIChat(character="Morgan Freeman")
+    tools = [weather, search]
+    ai("can you summarize the conversation?")
+    ai.save_session("chat_session_3.json", format="json")
+    ai_with_tools = partial(ai, tools=tools)
+
+    while True:
+        response = ai_with_tools(input("User: "))
+        print(response["response"])
+
+    # ai.reset_session()
+
+# TODO: play music, create routines for spotify and/or amazon music; play videos on youtube (i know it will just be the audio)
+#  use faster-whisper instead of the API
+#  using llama.cpp https://github.com/minimaxir/simpleaichat/pull/52
+# checkout swarms: https://github.com/kyegomez/swarms
+# improve chat interface: https://www.gradio.app/guides/creating-a-custom-chatbot-with-blocks

requirements.txt

@@ -0,0 +1,6 @@
+python-dotenv
+simpleaichat
+aiohttp
+openai
+gradio
+faster-whisper

utils/env.py

@@ -0,0 +1,30 @@
+import os
+from typing import Optional, Any, Dict
+
+from dotenv import find_dotenv, load_dotenv
+
+load_dotenv(find_dotenv(raise_error_if_not_found=True))
+
+
+def get_from_dict_or_env(
+    data: Dict[str, Any], key: str, env_key: str, default: Optional[str] = None
+) -> str:
+    """Get a value from a dictionary or an environment variable."""
+    if key in data and data[key]:
+        return data[key]
+    else:
+        return get_from_env(env_key, default=default)
+
+
+def get_from_env(env_key: str, default: Optional[str] = None) -> str:
+    """Get a value from a dictionary or an environment variable."""
+    if env_key in os.environ and os.environ[env_key]:
+        return os.environ[env_key]
+    elif default is not None:
+        return default
+    else:
+        raise ValueError(
+            f"Did not find {env_key}, please add an environment variable"
+            f" `{env_key}` which contains it, or pass"
+            f"  `{env_key}` as a named parameter."
+        )

utils/functions.py

@@ -0,0 +1,80 @@
+from dotenv import find_dotenv, load_dotenv
+import requests
+import os
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Iterator, Optional
+
+load_dotenv(find_dotenv())
+
+
+def get_voices():
+    url = "https://api.elevenlabs.io/v1/voices"
+    api_key = os.getenv("ELEVEN_API_KEY")
+    headers = {"Accept": "application/json", "xi-api-key": api_key}
+
+    response = requests.get(url, headers=headers)
+
+    print(response.text)
+    return response.text
+
+
+def is_installed(lib_name: str) -> bool:
+    lib = shutil.which(lib_name)
+    if lib is None:
+        return False
+    global_path = Path(lib)
+    # else check if path is valid and has the correct access rights
+    return global_path.exists() and os.access(global_path, os.X_OK)
+
+
+def play(audio: bytes, notebook: bool = False) -> None:
+    if notebook:
+        from IPython.display import Audio, display
+
+        display(Audio(audio, rate=44100, autoplay=True))
+    else:
+        if not is_installed("ffplay"):
+            raise ValueError("ffplay from ffmpeg not found, necessary to play audio.")
+        args = ["ffplay", "-autoexit", "-", "-nodisp"]
+        proc = subprocess.Popen(
+            args=args,
+            stdout=subprocess.PIPE,
+            stdin=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        _, err = proc.communicate(input=audio)
+        proc.poll()
+
+
+def save(audio: bytes, filename: str) -> None:
+    with open(filename, "wb") as f:
+        f.write(audio)
+
+
+def stream(audio_stream: Iterator[bytes]) -> bytes:
+    if not is_installed("mpv"):
+        raise ValueError("mpv not found, necessary to stream audio.")
+
+    mpv_command = ["mpv", "--no-cache", "--no-terminal", "--", "fd://0"]
+    mpv_process = subprocess.Popen(
+        mpv_command,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
+    audio = b""
+
+    for chunk in audio_stream:
+        if chunk is not None:
+            mpv_process.stdin.write(chunk)  # type: ignore
+            mpv_process.stdin.flush()  # type: ignore
+            audio += chunk
+
+    if mpv_process.stdin:
+        mpv_process.stdin.close()
+    mpv_process.wait()
+
+    return audio

utils/google_serper.py

@@ -0,0 +1,182 @@
+"""Util that calls Google Search using the Serper.dev API."""
+from typing import Any, Dict, List, Optional
+
+import aiohttp
+import requests
+from typing_extensions import Literal
+
+from pydantic import BaseModel
+from utils.env import get_from_env
+
+
+class GoogleSerper(BaseModel):
+    """Wrapper around the Serper.dev Google Search API.
+
+    You can create a free API key at https://serper.dev.
+
+    To use, you should have the environment variable ``SERPER_API_KEY``
+    set with your API key, or pass `serper_api_key` as a named parameter
+    to the constructor.
+    """
+
+    k: int = 10
+    gl: str = "us"
+    hl: str = "en"
+    # "places" and "images" is available from Serper but not implemented in the
+    # parser of run(). They can be used in results()
+    type: Literal["news", "search", "places", "images"] = "search"
+    result_key_for_type: dict = {
+        "news": "news",
+        "places": "places",
+        "images": "images",
+        "search": "organic",
+    }
+
+    tbs: Optional[str] = None
+    serper_api_key: Optional[str] = get_from_env("SERPER_API_KEY")
+    aiosession: Optional[aiohttp.ClientSession] = None
+
+    class Config:
+        """Configuration for this pydantic object."""
+
+        arbitrary_types_allowed = True
+
+    def results(self, query: str, **kwargs: Any) -> Dict:
+        """Run query through GoogleSearch."""
+        return self._google_serper_api_results(
+            query,
+            gl=self.gl,
+            hl=self.hl,
+            num=self.k,
+            tbs=self.tbs,
+            search_type=self.type,
+            **kwargs,
+        )
+
+    def run(self, query: str, **kwargs: Any) -> str:
+        """Run query through GoogleSearch and parse result."""
+        results = self._google_serper_api_results(
+            query,
+            gl=self.gl,
+            hl=self.hl,
+            num=self.k,
+            tbs=self.tbs,
+            search_type=self.type,
+            **kwargs,
+        )
+
+        return self._parse_results(results)
+
+    async def aresults(self, query: str, **kwargs: Any) -> Dict:
+        """Run query through GoogleSearch."""
+        results = await self._async_google_serper_search_results(
+            query,
+            gl=self.gl,
+            hl=self.hl,
+            num=self.k,
+            search_type=self.type,
+            tbs=self.tbs,
+            **kwargs,
+        )
+        return results
+
+    async def arun(self, query: str, **kwargs: Any) -> str:
+        """Run query through GoogleSearch and parse result async."""
+        results = await self._async_google_serper_search_results(
+            query,
+            gl=self.gl,
+            hl=self.hl,
+            num=self.k,
+            search_type=self.type,
+            tbs=self.tbs,
+            **kwargs,
+        )
+
+        return self._parse_results(results)
+
+    def _parse_snippets(self, results: dict) -> List[str]:
+        snippets = []
+
+        if results.get("answerBox"):
+            answer_box = results.get("answerBox", {})
+            if answer_box.get("answer"):
+                return [answer_box.get("answer")]
+            elif answer_box.get("snippet"):
+                return [answer_box.get("snippet").replace("\n", " ")]
+            elif answer_box.get("snippetHighlighted"):
+                return answer_box.get("snippetHighlighted")
+
+        if results.get("knowledgeGraph"):
+            kg = results.get("knowledgeGraph", {})
+            title = kg.get("title")
+            entity_type = kg.get("type")
+            if entity_type:
+                snippets.append(f"{title}: {entity_type}.")
+            description = kg.get("description")
+            if description:
+                snippets.append(description)
+            for attribute, value in kg.get("attributes", {}).items():
+                snippets.append(f"{title} {attribute}: {value}.")
+
+        for result in results[self.result_key_for_type[self.type]][: self.k]:
+            if "snippet" in result:
+                snippets.append(result["snippet"])
+            for attribute, value in result.get("attributes", {}).items():
+                snippets.append(f"{attribute}: {value}.")
+
+        if len(snippets) == 0:
+            return ["No good Google Search Result was found"]
+        return snippets
+
+    def _parse_results(self, results: dict) -> str:
+        return " ".join(self._parse_snippets(results))
+
+    def _google_serper_api_results(
+        self, search_term: str, search_type: str = "search", **kwargs: Any
+    ) -> dict:
+        headers = {
+            "X-API-KEY": self.serper_api_key or "",
+            "Content-Type": "application/json",
+        }
+        params = {
+            "q": search_term,
+            **{key: value for key, value in kwargs.items() if value is not None},
+        }
+        response = requests.post(
+            f"https://google.serper.dev/{search_type}", headers=headers, params=params
+        )
+        response.raise_for_status()
+        search_results = response.json()
+        return search_results
+
+    async def _async_google_serper_search_results(
+        self, search_term: str, search_type: str = "search", **kwargs: Any
+    ) -> dict:
+        headers = {
+            "X-API-KEY": self.serper_api_key or "",
+            "Content-Type": "application/json",
+        }
+        url = f"https://google.serper.dev/{search_type}"
+        params = {
+            "q": search_term,
+            **{key: value for key, value in kwargs.items() if value is not None},
+        }
+
+        if not self.aiosession:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(
+                    url, params=params, headers=headers, raise_for_status=False
+                ) as response:
+                    search_results = await response.json()
+        else:
+            async with self.aiosession.post(
+                url, params=params, headers=headers, raise_for_status=True
+            ) as response:
+                search_results = await response.json()
+
+        return search_results
+
+
+if __name__ == "__main__":
+    serper = GoogleSerper()
+    serper.run("A langchain alternative")

utils/tts.py

@@ -0,0 +1,58 @@
+from os import getenv
+
+import requests
+
+from utils.functions import play, stream, save
+
+ELEVEN_API_KEY = getenv("ELEVEN_API_KEY")
+CHUNK_SIZE = 1024
+ELEVENLABS_STREAM_ENDPOINT = "https://api.elevenlabs.io/v1/text-to-speech/{voice_id}/stream?optimize_streaming_latency=3"
+ELEVENLABS_ENDPOINT = "https://api.elevenlabs.io/v1/text-to-speech/{voice_id}"
+
+
+voices = {
+    "Bella": "EXAVITQu4vr4xnSDxMaL",
+    "Dorothy": "ThT5KcBeYPX3keUQqHPh",
+    "Male": "onwK4e9ZLuTAKqWW03F9",
+    "Chimamanda": "QSKN4kAq766BnZ0ilL0L",
+    "Ruth": "o9iLaGDMP3YCJcZevdfB",
+    "Ifeanyi": "iQe5hWADpVlprlflH1k8",
+}
+
+
+class TTS:
+    def __init__(self, voice_id):
+        self.voice_id = voice_id
+        self.headers = {
+            "Accept": "audio/mpeg",
+            "Content-Type": "application/json",
+            "xi-api-key": ELEVEN_API_KEY,
+        }
+
+    def generate(self, text, stream_: bool = False, model="eleven_monolingual_v1"):
+        data = {
+            "text": text,
+            "model_id": model,
+            "voice_settings": {"stability": 0.5, "similarity_boost": 0.0},
+        }
+
+        url = (
+            ELEVENLABS_STREAM_ENDPOINT.format(voice_id=self.voice_id)
+            if stream_
+            else ELEVENLABS_STREAM_ENDPOINT.format(voice_id=self.voice_id)
+        )
+        response = requests.post(
+            url,
+            json=data,
+            headers=self.headers,
+            stream=stream_,
+        )
+
+        if stream_:
+            audio_stream = (
+                chunk for chunk in response.iter_content(chunk_size=CHUNK_SIZE) if chunk
+            )
+            return audio_stream
+        else:
+            audio = response.content
+            return audio

utils/weather_forecast.py

@@ -0,0 +1,157 @@
+from datetime import datetime
+from typing import Optional
+
+import httpx
+from pydantic import BaseModel, Field
+from simpleaichat import AIChat
+
+from utils.env import get_from_env
+
+WEATHERAPI_BASE_URL = "http://api.weatherapi.com/v1"
+
+
+class WeatherAPI(BaseModel):
+    """Wrapper for WeatherAPI
+
+    Docs for using:
+
+    1. Go to https://www.weatherapi.com/ and sign up for an API key
+    2. Save your API WEATHERAPI_API_KEY env variable
+    """
+
+    weatherapi_key: Optional[str] = get_from_env("WEATHERAPI_API_KEY")
+    forecast_response: dict = None
+    warning: str = ""
+
+    def _format_weather_info(self):
+        current = self.forecast_response["current"]
+        forecast = self.forecast_response["forecast"]["forecastday"][0]
+        location_name = self.forecast_response["location"]["name"]
+        current_status = current["condition"]["text"]
+        wind_speed = current["wind_kph"]
+        wind_degree = current["wind_degree"]
+        humidity = current["humidity"]
+        cloud = current["cloud"]
+        feels_like = current["feelslike_c"]
+        current_temp = current["temp_c"]
+        min_temp = forecast["day"]["mintemp_c"]
+        max_temp = forecast["day"]["maxtemp_c"]
+        total_precip_mm = forecast["day"]["totalprecip_mm"]
+        total_snow_cm = forecast["day"]["totalsnow_cm"]
+        chance_of_rain = forecast["day"]["daily_chance_of_rain"]
+        chance_of_snow = forecast["day"]["daily_chance_of_snow"]
+        status = forecast["day"]["condition"]["text"]
+
+        is_day = current["is_day"]
+        sunrise = forecast["astro"]["sunrise"]
+        sunset = forecast["astro"]["sunset"]
+        # Convert the sunrise and sunset times to datetime objects
+        sunrise_time = datetime.strptime(sunrise, "%H:%M %p").time()
+        # sunset_time = datetime.strptime(sunset, "%I:%M %p").time()
+
+        # Get the current time
+        now = datetime.now().time()
+
+        # Check if the current time is before or after sunrise
+        if now < sunrise_time:
+            next_event = f"The sun will rise at {sunrise}."
+        else:
+            next_event = f"The sun will set at {sunset}."
+        rain_times = []
+        snow_times = []
+        for i, c in enumerate(forecast["hour"]):
+            if c["will_it_rain"]:
+                rain_times.append(c["time"])
+            if c["will_it_snow"]:
+                snow_times.append(c["time"])
+
+        results = (
+            f"In {location_name}, the current weather is as follows:\n"
+            f"Detailed status: {current_status}\n"
+            f"Wind speed: {wind_speed} kph, direction: {wind_degree}°\n"
+            f"Humidity: {humidity}%\n"
+            f"Temperature: \n"
+            f"  - Current: {current_temp}°C\n"
+            f"  - High: {max_temp}°C\n"
+            f"  - Low: {min_temp}°C\n"
+            f"  - Feels like: {feels_like}°C\n"
+            f"Cloud cover: {cloud}%\n"
+            f"Precipitation:\n"
+            f"  - Total precipitation: {total_precip_mm} mm\n"
+            f"  - Total snowfall: {total_snow_cm} cm\n"
+            f"Chance of precipitation:\n"
+            f"  - Chance of rain: {chance_of_rain}%\n"
+            f"  - Chance of snow: {chance_of_snow}%\n"
+            f"Weather status for the day: {status}\n"
+            + (
+                f"It is currently daytime.\n"
+                if is_day
+                else f"It is currently nighttime.\n"
+            )
+            + next_event
+        )
+
+        if rain_times:
+            results += "There is a chance of rain at the following times:\n"
+            for time in rain_times:
+                results += f"- {time}\n"
+
+        if snow_times:
+            results += "There is a chance of snow at the following times:\n"
+            for time in snow_times:
+                results += f"- {time}\n"
+
+        return f"{self.warning}\n{results}"
+
+    def get_forecast(self, location: dict):
+        """Get the current weather information for a specified location."""
+        forecast_url = f"{WEATHERAPI_BASE_URL}/forecast.json"
+        forecast_params = {
+            "key": self.weatherapi_key,
+            "q": location["city"],
+            "format": "json",
+            "days": 1,
+        }
+
+        r = httpx.get(forecast_url, params=forecast_params).json()
+        self.forecast_response = r
+
+    def run(self, query) -> str | None:
+        location = get_location(query)
+        if not location:
+            self.warning = (
+                "Could not identify any location in the query. Defaulted to Halifax."
+            )
+            location = {"city": "Halifax", "country": "CA"}
+        self.get_forecast(location)
+        weather_info = self._format_weather_info()
+        return weather_info
+
+
+class GetLocationMetadata(BaseModel):
+    """Location information"""
+
+    city: str = Field(description="The city of the location.")
+    state: int = Field(
+        description="The state or province of the location. Must be the full state name"
+    )
+    state_code: int = Field(
+        description="The state or province of the location. Must be a 2-char string."
+    )
+    country: str = Field(
+        description="The country of the location. Country must be a 2-char string"
+    )
+
+
+def get_location(query: str) -> dict | None:
+    params = {"temperature": 0.0, "max_tokens": 100}
+    system_prompt = (
+        "You reply ONLY with the location information. If no location is detected, respond with None in "
+        "each field"
+    )
+    ai = AIChat(system=system_prompt, params=params)
+    # noinspection PyTypeChecker
+    location: dict = ai(query, output_schema=GetLocationMetadata)
+    if location["city"] == "None":
+        return None
+    return location