Re-lint after rebase

ROADMAP.md

@@ -10,4 +10,3 @@ In the coming months, we're going to release:
 - [ ] An open-source language model for computer control
 - [ ] A react-native app for your phone
 - [ ] A hand-held device that runs fully offline.
-      

software/.cursorignore

@@ -1,4 +1,3 @@
 _archive
 __pycache__
 .idea
-

software/source/clients/esp32/README.md

@@ -26,4 +26,3 @@ And build and upload the firmware with a simple command:
 ```bash
 pio run --target upload
 ```
-

software/source/clients/linux/device.py

@@ -2,9 +2,11 @@ from ..base_device import Device
 
 device = Device()
 
+
 def main(server_url):
     device.server_url = server_url
     device.start()
 
+
 if __name__ == "__main__":
     main()

software/source/clients/mac/device.py

@@ -2,9 +2,11 @@ from ..base_device import Device
 
 device = Device()
 
+
 def main(server_url):
     device.server_url = server_url
     device.start()
 
+
 if __name__ == "__main__":
     main()

software/source/clients/rpi/device.py

@@ -2,8 +2,10 @@ from ..base_device import Device
 
 device = Device()
 
+
 def main():
     device.start()
 
+
 if __name__ == "__main__":
     main()

software/source/clients/windows/device.py

@@ -2,9 +2,11 @@ from ..base_device import Device
 
 device = Device()
 
+
 def main(server_url):
     device.server_url = server_url
     device.start()
 
+
 if __name__ == "__main__":
     main()

software/source/server/llm.py

@@ -1,4 +1,5 @@
 from dotenv import load_dotenv
+
 load_dotenv()  # take environment variables from .env.
 
 import os
@@ -8,7 +9,7 @@ from pathlib import Path
 ### LLM SETUP
 
 # Define the path to a llamafile
-llamafile_path = Path(__file__).parent / 'model.llamafile'
+llamafile_path = Path(__file__).parent / "model.llamafile"
 
 # Check if the new llamafile exists, if not download it
 if not os.path.exists(llamafile_path):

software/source/server/services/llm/litellm/llm.py

@@ -1,6 +1,5 @@
 class Llm:
     def __init__(self, config):
-
         # Litellm is used by OI by default, so we just modify OI
 
         interpreter = config["interpreter"]
@@ -10,6 +9,3 @@ class Llm:
             setattr(interpreter, key.replace("-", "_"), value)
 
         self.llm = interpreter.llm.completions
-
-        
-

software/source/server/services/llm/llamaedge/llm.py

@@ -3,29 +3,54 @@ import subprocess
 import requests
 import json
 
+
 class Llm:
     def __init__(self, config):
         self.install(config["service_directory"])
 
     def install(self, service_directory):
         LLM_FOLDER_PATH = service_directory
-        self.llm_directory = os.path.join(LLM_FOLDER_PATH, 'llm')
+        self.llm_directory = os.path.join(LLM_FOLDER_PATH, "llm")
         if not os.path.isdir(self.llm_directory):  # Check if the LLM directory exists
             os.makedirs(LLM_FOLDER_PATH, exist_ok=True)
 
             # Install WasmEdge
-            subprocess.run(['curl', '-sSf', 'https://raw.githubusercontent.com/WasmEdge/WasmEdge/master/utils/install.sh', '|', 'bash', '-s', '--', '--plugin', 'wasi_nn-ggml'])
+            subprocess.run(
+                [
+                    "curl",
+                    "-sSf",
+                    "https://raw.githubusercontent.com/WasmEdge/WasmEdge/master/utils/install.sh",
+                    "|",
+                    "bash",
+                    "-s",
+                    "--",
+                    "--plugin",
+                    "wasi_nn-ggml",
+                ]
+            )
 
             # Download the Qwen1.5-0.5B-Chat model GGUF file
             MODEL_URL = "https://huggingface.co/second-state/Qwen1.5-0.5B-Chat-GGUF/resolve/main/Qwen1.5-0.5B-Chat-Q5_K_M.gguf"
-            subprocess.run(['curl', '-LO', MODEL_URL], cwd=self.llm_directory)
+            subprocess.run(["curl", "-LO", MODEL_URL], cwd=self.llm_directory)
 
             # Download the llama-api-server.wasm app
             APP_URL = "https://github.com/LlamaEdge/LlamaEdge/releases/latest/download/llama-api-server.wasm"
-            subprocess.run(['curl', '-LO', APP_URL], cwd=self.llm_directory)
+            subprocess.run(["curl", "-LO", APP_URL], cwd=self.llm_directory)
 
             # Run the API server
-            subprocess.run(['wasmedge', '--dir', '.:.', '--nn-preload', 'default:GGML:AUTO:Qwen1.5-0.5B-Chat-Q5_K_M.gguf', 'llama-api-server.wasm', '-p', 'llama-2-chat'], cwd=self.llm_directory)
+            subprocess.run(
+                [
+                    "wasmedge",
+                    "--dir",
+                    ".:.",
+                    "--nn-preload",
+                    "default:GGML:AUTO:Qwen1.5-0.5B-Chat-Q5_K_M.gguf",
+                    "llama-api-server.wasm",
+                    "-p",
+                    "llama-2-chat",
+                ],
+                cwd=self.llm_directory,
+            )
 
             print("LLM setup completed.")
         else:
@@ -33,17 +58,11 @@ class Llm:
 
     def llm(self, messages):
         url = "http://localhost:8080/v1/chat/completions"
-        headers = {
-            'accept': 'application/json',
-            'Content-Type': 'application/json'
-        }
-        data = {
-            "messages": messages,
-            "model": "llama-2-chat"
-        }
-        with requests.post(url, headers=headers, data=json.dumps(data), stream=True) as response:
+        headers = {"accept": "application/json", "Content-Type": "application/json"}
+        data = {"messages": messages, "model": "llama-2-chat"}
+        with requests.post(
+            url, headers=headers, data=json.dumps(data), stream=True
+        ) as response:
             for line in response.iter_lines():
                 if line:
                     yield json.loads(line)
-
-

software/source/server/services/stt/openai/stt.py

@@ -6,7 +6,6 @@ class Stt:
         return stt(audio_file_path)
 
 
-
 from datetime import datetime
 import os
 import contextlib
@@ -19,6 +18,7 @@ from openai import OpenAI
 
 client = OpenAI()
 
+
 def convert_mime_type_to_format(mime_type: str) -> str:
     if mime_type == "audio/x-wav" or mime_type == "audio/wav":
         return "wav"
@@ -29,30 +29,37 @@ def convert_mime_type_to_format(mime_type: str) -> str:
 
     return mime_type
 
+
 @contextlib.contextmanager
 def export_audio_to_wav_ffmpeg(audio: bytearray, mime_type: str) -> str:
     temp_dir = tempfile.gettempdir()
 
     # Create a temporary file with the appropriate extension
     input_ext = convert_mime_type_to_format(mime_type)
-    input_path = os.path.join(temp_dir, f"input_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.{input_ext}")
-    with open(input_path, 'wb') as f:
+    input_path = os.path.join(
+        temp_dir, f"input_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.{input_ext}"
+    )
+    with open(input_path, "wb") as f:
         f.write(audio)
 
     # Check if the input file exists
     assert os.path.exists(input_path), f"Input file does not exist: {input_path}"
 
     # Export to wav
-    output_path = os.path.join(temp_dir, f"output_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.wav")
+    output_path = os.path.join(
+        temp_dir, f"output_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.wav"
+    )
     if mime_type == "audio/raw":
         ffmpeg.input(
             input_path,
-            f='s16le',
-            ar='16000',
+            f="s16le",
+            ar="16000",
             ac=1,
-        ).output(output_path, loglevel='panic').run()
+        ).output(output_path, loglevel="panic").run()
     else:
-        ffmpeg.input(input_path).output(output_path, acodec='pcm_s16le', ac=1, ar='16k', loglevel='panic').run()
+        ffmpeg.input(input_path).output(
+            output_path, acodec="pcm_s16le", ac=1, ar="16k", loglevel="panic"
+        ).run()
 
     try:
         yield output_path
@@ -60,39 +67,49 @@ def export_audio_to_wav_ffmpeg(audio: bytearray, mime_type: str) -> str:
         os.remove(input_path)
         os.remove(output_path)
 
+
 def run_command(command):
-    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    result = subprocess.run(
+        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
+    )
     return result.stdout, result.stderr
 
-def get_transcription_file(wav_file_path: str):
-    local_path = os.path.join(os.path.dirname(__file__), 'local_service')
-    whisper_rust_path = os.path.join(os.path.dirname(__file__), 'whisper-rust', 'target', 'release')
-    model_name = os.getenv('WHISPER_MODEL_NAME', 'ggml-tiny.en.bin')
 
-    output, error = run_command([
-        os.path.join(whisper_rust_path, 'whisper-rust'),
-        '--model-path', os.path.join(local_path, model_name),
-        '--file-path', wav_file_path
-    ])
+def get_transcription_file(wav_file_path: str):
+    local_path = os.path.join(os.path.dirname(__file__), "local_service")
+    whisper_rust_path = os.path.join(
+        os.path.dirname(__file__), "whisper-rust", "target", "release"
+    )
+    model_name = os.getenv("WHISPER_MODEL_NAME", "ggml-tiny.en.bin")
+
+    output, error = run_command(
+        [
+            os.path.join(whisper_rust_path, "whisper-rust"),
+            "--model-path",
+            os.path.join(local_path, model_name),
+            "--file-path",
+            wav_file_path,
+        ]
+    )
 
     return output
 
+
 def get_transcription_bytes(audio_bytes: bytearray, mime_type):
     with export_audio_to_wav_ffmpeg(audio_bytes, mime_type) as wav_file_path:
         return get_transcription_file(wav_file_path)
 
+
 def stt_bytes(audio_bytes: bytearray, mime_type="audio/wav"):
     with export_audio_to_wav_ffmpeg(audio_bytes, mime_type) as wav_file_path:
         return stt_wav(wav_file_path)
 
-def stt_wav(wav_file_path: str):
 
+def stt_wav(wav_file_path: str):
     audio_file = open(wav_file_path, "rb")
     try:
         transcript = client.audio.transcriptions.create(
-            model="whisper-1", 
-            file=audio_file,
-            response_format="text"
+            model="whisper-1", file=audio_file, response_format="text"
         )
     except openai.BadRequestError as e:
         print(f"openai.BadRequestError: {e}")
@@ -100,10 +117,13 @@ def stt_wav(wav_file_path: str):
 
     return transcript
 
+
 def stt(input_data, mime_type="audio/wav"):
     if isinstance(input_data, str):
         return stt_wav(input_data)
     elif isinstance(input_data, bytearray):
         return stt_bytes(input_data, mime_type)
     else:
-        raise ValueError("Input data should be either a path to a wav file (str) or audio bytes (bytearray)")
+        raise ValueError(
+            "Input data should be either a path to a wav file (str) or audio bytes (bytearray)"
+        )

software/source/server/services/tts/piper/tts.py

@@ -13,26 +13,40 @@ class Tts:
         self.install(config["service_directory"])
 
     def tts(self, text):
-
         with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
             output_file = temp_file.name
             piper_dir = self.piper_directory
-            subprocess.run([
-                os.path.join(piper_dir, 'piper'),
-                '--model', os.path.join(piper_dir, os.getenv('PIPER_VOICE_NAME', 'en_US-lessac-medium.onnx')),
-                '--output_file', output_file
-            ], input=text, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+            subprocess.run(
+                [
+                    os.path.join(piper_dir, "piper"),
+                    "--model",
+                    os.path.join(
+                        piper_dir,
+                        os.getenv("PIPER_VOICE_NAME", "en_US-lessac-medium.onnx"),
+                    ),
+                    "--output_file",
+                    output_file,
+                ],
+                input=text,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+            )
 
             # TODO: hack to format audio correctly for device
             outfile = tempfile.gettempdir() + "/" + "raw.dat"
-            ffmpeg.input(temp_file.name).output(outfile, f="s16le", ar="16000", ac="1", loglevel='panic').run()
+            ffmpeg.input(temp_file.name).output(
+                outfile, f="s16le", ar="16000", ac="1", loglevel="panic"
+            ).run()
 
             return outfile
 
     def install(self, service_directory):
         PIPER_FOLDER_PATH = service_directory
-        self.piper_directory = os.path.join(PIPER_FOLDER_PATH, 'piper')
-        if not os.path.isdir(self.piper_directory):  # Check if the Piper directory exists
+        self.piper_directory = os.path.join(PIPER_FOLDER_PATH, "piper")
+        if not os.path.isdir(
+            self.piper_directory
+        ):  # Check if the Piper directory exists
             os.makedirs(PIPER_FOLDER_PATH, exist_ok=True)
 
             # Determine OS and architecture
@@ -60,51 +74,91 @@ class Tts:
             asset_url = f"{PIPER_URL}{PIPER_ASSETNAME}"
 
             if OS == "windows":
-
                 asset_url = asset_url.replace(".tar.gz", ".zip")
 
             # Download and extract Piper
-            urllib.request.urlretrieve(asset_url, os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME))
+            urllib.request.urlretrieve(
+                asset_url, os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME)
+            )
 
             # Extract the downloaded file
             if OS == "windows":
                 import zipfile
-                with zipfile.ZipFile(os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME), 'r') as zip_ref:
+
+                with zipfile.ZipFile(
+                    os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME), "r"
+                ) as zip_ref:
                     zip_ref.extractall(path=PIPER_FOLDER_PATH)
             else:
-                with tarfile.open(os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME), 'r:gz') as tar:
+                with tarfile.open(
+                    os.path.join(PIPER_FOLDER_PATH, PIPER_ASSETNAME), "r:gz"
+                ) as tar:
                     tar.extractall(path=PIPER_FOLDER_PATH)
 
-            PIPER_VOICE_URL = os.getenv('PIPER_VOICE_URL',
-                                        'https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/')
-            PIPER_VOICE_NAME = os.getenv('PIPER_VOICE_NAME', 'en_US-lessac-medium.onnx')
+            PIPER_VOICE_URL = os.getenv(
+                "PIPER_VOICE_URL",
+                "https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/lessac/medium/",
+            )
+            PIPER_VOICE_NAME = os.getenv("PIPER_VOICE_NAME", "en_US-lessac-medium.onnx")
 
             # Download voice model and its json file
-            urllib.request.urlretrieve(f"{PIPER_VOICE_URL}{PIPER_VOICE_NAME}",
-                                       os.path.join(self.piper_directory, PIPER_VOICE_NAME))
-            urllib.request.urlretrieve(f"{PIPER_VOICE_URL}{PIPER_VOICE_NAME}.json",
-                                       os.path.join(self.piper_directory, f"{PIPER_VOICE_NAME}.json"))
+            urllib.request.urlretrieve(
+                f"{PIPER_VOICE_URL}{PIPER_VOICE_NAME}",
+                os.path.join(self.piper_directory, PIPER_VOICE_NAME),
+            )
+            urllib.request.urlretrieve(
+                f"{PIPER_VOICE_URL}{PIPER_VOICE_NAME}.json",
+                os.path.join(self.piper_directory, f"{PIPER_VOICE_NAME}.json"),
+            )
 
             # Additional setup for macOS
             if OS == "macos":
                 if ARCH == "x64":
-                    subprocess.run(['softwareupdate', '--install-rosetta', '--agree-to-license'])
+                    subprocess.run(
+                        ["softwareupdate", "--install-rosetta", "--agree-to-license"]
+                    )
 
                 PIPER_PHONEMIZE_ASSETNAME = f"piper-phonemize_{OS}_{ARCH}.tar.gz"
                 PIPER_PHONEMIZE_URL = "https://github.com/rhasspy/piper-phonemize/releases/latest/download/"
-                urllib.request.urlretrieve(f"{PIPER_PHONEMIZE_URL}{PIPER_PHONEMIZE_ASSETNAME}",
-                                           os.path.join(self.piper_directory, PIPER_PHONEMIZE_ASSETNAME))
-
-                with tarfile.open(os.path.join(self.piper_directory, PIPER_PHONEMIZE_ASSETNAME), 'r:gz') as tar:
+                urllib.request.urlretrieve(
+                    f"{PIPER_PHONEMIZE_URL}{PIPER_PHONEMIZE_ASSETNAME}",
+                    os.path.join(self.piper_directory, PIPER_PHONEMIZE_ASSETNAME),
+                )
+
+                with tarfile.open(
+                    os.path.join(self.piper_directory, PIPER_PHONEMIZE_ASSETNAME),
+                    "r:gz",
+                ) as tar:
                     tar.extractall(path=self.piper_directory)
 
                 PIPER_DIR = self.piper_directory
-                subprocess.run(['install_name_tool', '-change', '@rpath/libespeak-ng.1.dylib',
-                                f"{PIPER_DIR}/piper-phonemize/lib/libespeak-ng.1.dylib", f"{PIPER_DIR}/piper"])
-                subprocess.run(['install_name_tool', '-change', '@rpath/libonnxruntime.1.14.1.dylib',
-                                f"{PIPER_DIR}/piper-phonemize/lib/libonnxruntime.1.14.1.dylib", f"{PIPER_DIR}/piper"])
-                subprocess.run(['install_name_tool', '-change', '@rpath/libpiper_phonemize.1.dylib',
-                                f"{PIPER_DIR}/piper-phonemize/lib/libpiper_phonemize.1.dylib", f"{PIPER_DIR}/piper"])
+                subprocess.run(
+                    [
+                        "install_name_tool",
+                        "-change",
+                        "@rpath/libespeak-ng.1.dylib",
+                        f"{PIPER_DIR}/piper-phonemize/lib/libespeak-ng.1.dylib",
+                        f"{PIPER_DIR}/piper",
+                    ]
+                )
+                subprocess.run(
+                    [
+                        "install_name_tool",
+                        "-change",
+                        "@rpath/libonnxruntime.1.14.1.dylib",
+                        f"{PIPER_DIR}/piper-phonemize/lib/libonnxruntime.1.14.1.dylib",
+                        f"{PIPER_DIR}/piper",
+                    ]
+                )
+                subprocess.run(
+                    [
+                        "install_name_tool",
+                        "-change",
+                        "@rpath/libpiper_phonemize.1.dylib",
+                        f"{PIPER_DIR}/piper-phonemize/lib/libpiper_phonemize.1.dylib",
+                        f"{PIPER_DIR}/piper",
+                    ]
+                )
 
             print("Piper setup completed.")
         else:

software/source/server/system_messages/TeachModeSystemMessage.py

@@ -131,4 +131,6 @@ print(output)
 
 Remember: You can run Python code outside a function only to run a Python function; all other code must go in a in Python function if you first write a Python function. ALL imports must go inside the function.
 
-""".strip().replace("OI_SKILLS_DIR", os.path.abspath(os.path.join(os.path.dirname(__file__), "skills")))
+""".strip().replace(
+    "OI_SKILLS_DIR", os.path.abspath(os.path.join(os.path.dirname(__file__), "skills"))
+)

software/source/server/tunnel.py

@@ -1,12 +1,14 @@
 import subprocess
 import re
-import shutil
 import pyqrcode
 import time
 from ..utils.print_markdown import print_markdown
 
-def create_tunnel(tunnel_method='ngrok', server_host='localhost', server_port=10001, qr=False):
-    print_markdown(f"Exposing server to the internet...")
+
+def create_tunnel(
+    tunnel_method="ngrok", server_host="localhost", server_port=10001, qr=False
+):
+    print_markdown("Exposing server to the internet...")
 
     server_url = ""
     if tunnel_method == "bore":
@@ -35,9 +37,11 @@ def create_tunnel(tunnel_method='ngrok', server_host='localhost', server_port=10
             if not line:
                 break
             if "listening at bore.pub:" in line:
-                remote_port = re.search('bore.pub:([0-9]*)', line).group(1)
+                remote_port = re.search("bore.pub:([0-9]*)", line).group(1)
                 server_url = f"bore.pub:{remote_port}"
-                print_markdown(f"Your server is being hosted at the following URL: bore.pub:{remote_port}")
+                print_markdown(
+                    f"Your server is being hosted at the following URL: bore.pub:{remote_port}"
+                )
                 break
 
     elif tunnel_method == "localtunnel":
@@ -69,9 +73,11 @@ def create_tunnel(tunnel_method='ngrok', server_host='localhost', server_port=10
                 match = url_pattern.search(line)
                 if match:
                     found_url = True
-                    remote_url = match.group(0).replace('your url is: ', '')
+                    remote_url = match.group(0).replace("your url is: ", "")
                     server_url = remote_url
-                    print(f"\nYour server is being hosted at the following URL: {remote_url}")
+                    print(
+                        f"\nYour server is being hosted at the following URL: {remote_url}"
+                    )
                     break  # Exit the loop once the URL is found
 
             if not found_url:
@@ -93,7 +99,11 @@ def create_tunnel(tunnel_method='ngrok', server_host='localhost', server_port=10
 
         # If ngrok is installed, start it on the specified port
         # process = subprocess.Popen(f'ngrok http {server_port} --log=stdout', shell=True, stdout=subprocess.PIPE)
-        process = subprocess.Popen(f'ngrok http {server_port} --scheme http,https --domain=marten-advanced-dragon.ngrok-free.app --log=stdout', shell=True, stdout=subprocess.PIPE)
+        process = subprocess.Popen(
+            f"ngrok http {server_port} --scheme http,https --domain=marten-advanced-dragon.ngrok-free.app --log=stdout",
+            shell=True,
+            stdout=subprocess.PIPE,
+        )
 
         # Initially, no URL is found
         found_url = False
@@ -110,15 +120,18 @@ def create_tunnel(tunnel_method='ngrok', server_host='localhost', server_port=10
                 found_url = True
                 remote_url = match.group(0)
                 server_url = remote_url
-                print(f"\nYour server is being hosted at the following URL: {remote_url}")
+                print(
+                    f"\nYour server is being hosted at the following URL: {remote_url}"
+                )
                 break  # Exit the loop once the URL is found
 
         if not found_url:
-            print("Failed to extract the ngrok tunnel URL. Please check ngrok's output for details.")
+            print(
+                "Failed to extract the ngrok tunnel URL. Please check ngrok's output for details."
+            )
 
     if server_url and qr:
         text = pyqrcode.create(remote_url)
         print(text.terminal(quiet_zone=1))
 
     return server_url
-

software/source/server/utils/bytes_to_wav.py

@@ -5,6 +5,7 @@ import tempfile
 import ffmpeg
 import subprocess
 
+
 def convert_mime_type_to_format(mime_type: str) -> str:
     if mime_type == "audio/x-wav" or mime_type == "audio/wav":
         return "wav"
@@ -15,39 +16,49 @@ def convert_mime_type_to_format(mime_type: str) -> str:
 
     return mime_type
 
+
 @contextlib.contextmanager
 def export_audio_to_wav_ffmpeg(audio: bytearray, mime_type: str) -> str:
     temp_dir = tempfile.gettempdir()
 
     # Create a temporary file with the appropriate extension
     input_ext = convert_mime_type_to_format(mime_type)
-    input_path = os.path.join(temp_dir, f"input_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.{input_ext}")
-    with open(input_path, 'wb') as f:
+    input_path = os.path.join(
+        temp_dir, f"input_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.{input_ext}"
+    )
+    with open(input_path, "wb") as f:
         f.write(audio)
 
     # Check if the input file exists
     assert os.path.exists(input_path), f"Input file does not exist: {input_path}"
 
     # Export to wav
-    output_path = os.path.join(temp_dir, f"output_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.wav")
+    output_path = os.path.join(
+        temp_dir, f"output_{datetime.now().strftime('%Y%m%d%H%M%S%f')}.wav"
+    )
     print(mime_type, input_path, output_path)
     if mime_type == "audio/raw":
         ffmpeg.input(
             input_path,
-            f='s16le',
-            ar='16000',
+            f="s16le",
+            ar="16000",
             ac=1,
-        ).output(output_path, loglevel='panic').run()
+        ).output(output_path, loglevel="panic").run()
     else:
-        ffmpeg.input(input_path).output(output_path, acodec='pcm_s16le', ac=1, ar='16k', loglevel='panic').run()
+        ffmpeg.input(input_path).output(
+            output_path, acodec="pcm_s16le", ac=1, ar="16k", loglevel="panic"
+        ).run()
 
     try:
         yield output_path
     finally:
         os.remove(input_path)
 
+
 def run_command(command):
-    result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    result = subprocess.run(
+        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
+    )
     return result.stdout, result.stderr
 
 

software/source/server/utils/kernel.py

@@ -1,4 +1,5 @@
 from dotenv import load_dotenv
+
 load_dotenv()  # take environment variables from .env.
 
 import asyncio
@@ -7,8 +8,10 @@ import platform
 
 from .logs import setup_logging
 from .logs import logger
+
 setup_logging()
 
+
 def get_kernel_messages():
     """
     Is this the way to do this?
@@ -16,20 +19,23 @@ def get_kernel_messages():
     current_platform = platform.system()
 
     if current_platform == "Darwin":
-        process = subprocess.Popen(['syslog'], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
+        process = subprocess.Popen(
+            ["syslog"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL
+        )
         output, _ = process.communicate()
-        return output.decode('utf-8')
+        return output.decode("utf-8")
     elif current_platform == "Linux":
-        with open('/var/log/dmesg', 'r') as file:
+        with open("/var/log/dmesg", "r") as file:
             return file.read()
     else:
         logger.info("Unsupported platform.")
 
+
 def custom_filter(message):
     # Check for {TO_INTERPRETER{ message here }TO_INTERPRETER} pattern
-    if '{TO_INTERPRETER{' in message and '}TO_INTERPRETER}' in message:
-        start = message.find('{TO_INTERPRETER{') + len('{TO_INTERPRETER{')
-        end = message.find('}TO_INTERPRETER}', start)
+    if "{TO_INTERPRETER{" in message and "}TO_INTERPRETER}" in message:
+        start = message.find("{TO_INTERPRETER{") + len("{TO_INTERPRETER{")
+        end = message.find("}TO_INTERPRETER}", start)
         return message[start:end]
     # Check for USB mention
     # elif 'USB' in message:
@@ -41,8 +47,10 @@ def custom_filter(message):
     else:
         return None
 
+
 last_messages = ""
 
+
 def check_filtered_kernel():
     messages = get_kernel_messages()
     if messages is None:
@@ -66,11 +74,25 @@ async def put_kernel_messages_into_queue(queue):
         if text:
             if isinstance(queue, asyncio.Queue):
                 await queue.put({"role": "computer", "type": "console", "start": True})
-                await queue.put({"role": "computer", "type": "console", "format": "output", "content": text})
+                await queue.put(
+                    {
+                        "role": "computer",
+                        "type": "console",
+                        "format": "output",
+                        "content": text,
+                    }
+                )
                 await queue.put({"role": "computer", "type": "console", "end": True})
             else:
                 queue.put({"role": "computer", "type": "console", "start": True})
-                queue.put({"role": "computer", "type": "console", "format": "output", "content": text})
+                queue.put(
+                    {
+                        "role": "computer",
+                        "type": "console",
+                        "format": "output",
+                        "content": text,
+                    }
+                )
                 queue.put({"role": "computer", "type": "console", "end": True})
 
         await asyncio.sleep(5)

software/source/server/utils/logs.py

@@ -1,4 +1,5 @@
 from dotenv import load_dotenv
+
 load_dotenv()  # take environment variables from .env.
 
 import os
@@ -9,9 +10,7 @@ root_logger: logging.Logger = logging.getLogger()
 
 
 def _basic_config() -> None:
-    logging.basicConfig(
-        format="%(message)s"
-    )
+    logging.basicConfig(format="%(message)s")
 
 
 def setup_logging() -> None:

software/source/utils/accumulator.py

@@ -6,7 +6,6 @@ class Accumulator:
     def accumulate(self, chunk):
         # print(str(chunk)[:100])
         if type(chunk) == dict:
-
             if "format" in chunk and chunk["format"] == "active_line":
                 # We don't do anything with these
                 return None
@@ -17,15 +16,20 @@ class Accumulator:
                 return None
 
             if "content" in chunk:
-
-                if any(self.message[key] != chunk[key] for key in self.message if key != "content"):
+                if any(
+                    self.message[key] != chunk[key]
+                    for key in self.message
+                    if key != "content"
+                ):
                     self.message = chunk
                 if "content" not in self.message:
                     self.message["content"] = chunk["content"]
                 else:
                     if type(chunk["content"]) == dict:
                         # dict concatenation cannot happen, so we see if chunk is a dict
-                        self.message["content"]["content"] += chunk["content"]["content"]
+                        self.message["content"]["content"] += chunk["content"][
+                            "content"
+                        ]
                     else:
                         self.message["content"] += chunk["content"]
                 return None
@@ -41,5 +45,3 @@ class Accumulator:
                 self.message["content"] = b""
             self.message["content"] += chunk
             return None
-
-        

software/source/utils/print_markdown.py

@@ -1,6 +1,7 @@
 from rich.console import Console
 from rich.markdown import Markdown
 
+
 def print_markdown(markdown_text):
     console = Console()
     md = Markdown(markdown_text)

software/start.py

@@ -16,34 +16,63 @@ app = typer.Typer()
 @app.command()
 def run(
     server: bool = typer.Option(False, "--server", help="Run server"),
-            server_host: str = typer.Option("0.0.0.0", "--server-host", help="Specify the server host where the server will deploy"),
-            server_port: int = typer.Option(10001, "--server-port", help="Specify the server port where the server will deploy"),
-
-            tunnel_service: str = typer.Option("ngrok", "--tunnel-service", help="Specify the tunnel service"),
+    server_host: str = typer.Option(
+        "0.0.0.0",
+        "--server-host",
+        help="Specify the server host where the server will deploy",
+    ),
+    server_port: int = typer.Option(
+        10001,
+        "--server-port",
+        help="Specify the server port where the server will deploy",
+    ),
+    tunnel_service: str = typer.Option(
+        "ngrok", "--tunnel-service", help="Specify the tunnel service"
+    ),
     expose: bool = typer.Option(False, "--expose", help="Expose server to internet"),
-
     client: bool = typer.Option(False, "--client", help="Run client"),
-            server_url: str = typer.Option(None, "--server-url", help="Specify the server URL that the client should expect. Defaults to server-host and server-port"),
-            client_type: str = typer.Option("auto", "--client-type", help="Specify the client type"),
-
-            llm_service: str = typer.Option("litellm", "--llm-service", help="Specify the LLM service"),
-
+    server_url: str = typer.Option(
+        None,
+        "--server-url",
+        help="Specify the server URL that the client should expect. Defaults to server-host and server-port",
+    ),
+    client_type: str = typer.Option(
+        "auto", "--client-type", help="Specify the client type"
+    ),
+    llm_service: str = typer.Option(
+        "litellm", "--llm-service", help="Specify the LLM service"
+    ),
     model: str = typer.Option("gpt-4", "--model", help="Specify the model"),
-            llm_supports_vision: bool = typer.Option(False, "--llm-supports-vision", help="Specify if the LLM service supports vision"),
-            llm_supports_functions: bool = typer.Option(False, "--llm-supports-functions", help="Specify if the LLM service supports functions"),
-            context_window: int = typer.Option(2048, "--context-window", help="Specify the context window size"),
-            max_tokens: int = typer.Option(4096, "--max-tokens", help="Specify the maximum number of tokens"),
-            temperature: float = typer.Option(0.8, "--temperature", help="Specify the temperature for generation"),
-
-            tts_service: str = typer.Option("openai", "--tts-service", help="Specify the TTS service"),
-
-            stt_service: str = typer.Option("openai", "--stt-service", help="Specify the STT service"),
-
-            local: bool = typer.Option(False, "--local", help="Use recommended local services for LLM, STT, and TTS"),
-
-            qr: bool = typer.Option(False, "--qr", help="Print the QR code for the server URL")
+    llm_supports_vision: bool = typer.Option(
+        False,
+        "--llm-supports-vision",
+        help="Specify if the LLM service supports vision",
+    ),
+    llm_supports_functions: bool = typer.Option(
+        False,
+        "--llm-supports-functions",
+        help="Specify if the LLM service supports functions",
+    ),
+    context_window: int = typer.Option(
+        2048, "--context-window", help="Specify the context window size"
+    ),
+    max_tokens: int = typer.Option(
+        4096, "--max-tokens", help="Specify the maximum number of tokens"
+    ),
+    temperature: float = typer.Option(
+        0.8, "--temperature", help="Specify the temperature for generation"
+    ),
+    tts_service: str = typer.Option(
+        "openai", "--tts-service", help="Specify the TTS service"
+    ),
+    stt_service: str = typer.Option(
+        "openai", "--stt-service", help="Specify the STT service"
+    ),
+    local: bool = typer.Option(
+        False, "--local", help="Use recommended local services for LLM, STT, and TTS"
+    ),
+    qr: bool = typer.Option(False, "--qr", help="Print the QR code for the server URL"),
 ):
-
     _run(
         server=server,
         server_host=server_host,
@@ -63,7 +92,7 @@ def run(
         tts_service=tts_service,
         stt_service=stt_service,
         local=local,
-        qr=qr
+        qr=qr,
     )
 
 
@@ -86,7 +115,7 @@ def _run(
     tts_service: str = "openai",
     stt_service: str = "openai",
     local: bool = False,
-    qr: bool = False
+    qr: bool = False,
 ):
     if local:
         tts_service = "piper"
@@ -130,7 +159,9 @@ def _run(
         server_thread.start()
 
     if expose:
-        tunnel_thread = threading.Thread(target=create_tunnel, args=[tunnel_service, server_host, server_port, qr])
+        tunnel_thread = threading.Thread(
+            target=create_tunnel, args=[tunnel_service, server_host, server_port, qr]
+        )
         tunnel_thread.start()
 
     if client: