Initial commit

.env

@@ -0,0 +1,3 @@
+GOOGLE_GENAI_USE_VERTEXAI=TRUE
+GOOGLE_CLOUD_PROJECT=vibecoder-459110
+GOOGLE_CLOUD_LOCATION=us-central1

.gitignore

@@ -0,0 +1,10 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv

.python-version

@@ -0,0 +1 @@
+3.12

Dockerfile

@@ -0,0 +1,17 @@
+FROM ghcr.io/astral-sh/uv:alpine
+
+RUN apk add curl python3
+
+RUN curl -fsSL https://raw.githubusercontent.com/arduino/arduino-cli/master/install.sh | sh
+
+COPY pyproject.toml .
+
+RUN uv sync
+
+COPY *.py .
+
+ENV GOOGLE_GENAI_USE_VERTEXAI=TRUE
+ENV GOOGLE_CLOUD_LOCATION=us-central1
+ENV GOOGLE_CLOUD_PROJECT=vibecoder-459110
+
+CMD ["uv", "run", "main.py"]

agent.py

@@ -0,0 +1,455 @@
+from google.adk.agents import Agent
+from google.adk.sessions import InMemorySessionService
+from google.adk.runners import Runner
+from google.adk.tools import (
+    ToolContext,
+    FunctionTool,
+    built_in_code_execution,
+    google_search,
+)
+from google.genai import types
+import asyncio
+import dotenv
+import os
+import subprocess
+import shutil
+
+dotenv.load_dotenv()
+
+
+def read_code(tool_context: ToolContext, start_line: int = 0, end_line: int = 9999999):
+    """
+    Read the code written so far. The code is arduino code.
+    You can optionally select which part of the code you would like to get.
+
+    Args:
+        start_line (int): The first line of the code part you want (defaults to 0)
+        end_line (int): The last line of the code part you want (defaults to infinity)
+
+    Returns:
+        str: The (part of the) code written so far.
+    """
+    code: str = tool_context.state.get("code", "")
+    return "\n".join(code.splitlines()[start_line:end_line])
+
+
+def write_code(tool_context: ToolContext, code: str):
+    """
+    Write code to the main file. Only put Arduino code here.
+    This is the only place where you are allowed to put code. Don't put code in the chat, only write it here.
+
+    Args:
+        code (str): The code you want to write to the main file.
+    """
+    tool_context.state["code"] = code
+
+
+COMPONENT_INFO_DICT = {
+    "Arduino Uno R3": {
+        "pins": {
+            "D0": "Serial (USART) RX + GPIO",
+            "D1": "Serial (USART) TX + GPIO",
+            "D2": "External interrupt (INT0) + GPIO",
+            "D3": "External interrupt (INT1) + PWM + GPIO",
+            "D4": "GPIO",
+            "D5": "PWM + GPIO",
+            "D6": "PWM + GPIO",
+            "D7": "GPIO",
+            "D8": "GPIO",
+            "D9": "PWM + GPIO",
+            "D10": "PWM + GPIO",
+            "D11": "PWM + GPIO",
+            "D12": "GPIO",
+            "D13": "GPIO",
+            "GND": "Ground",
+            "D14": "GPIO + A[0]",
+            "D15": "GPIO + A[1]",
+            "D16": "GPIO + A[2]",
+            "D17": "GPIO + A[3]",
+            "D18": "GPIO + I2C SDA + A[4]",
+            "D19": "GPIO + I2C SCL + A[5]",
+            "GND": "Ground",
+            "GND": "Ground",
+            "+5V": "+5V OUT",
+            "+3.3V": "+3.3V OUT",
+        }
+    },
+    "Button": {
+        "pins": {
+            "D0": "Pin 1",
+            "D1": "Pin 2",
+        }
+    },
+    "SSD1306": {
+        "pins": {
+            "VCC": "Supply Voltage",
+            "GND": "Ground",
+            "SCL": "I2C clock line",
+            "SDA": "I2C data line",
+        }
+    },
+    "LED Green": {
+        "pins": {
+            "an": "Anode",
+            "ka": "Kathode",
+        }
+    },
+    "LED Blue": {
+        "pins": {
+            "an": "Anode",
+            "ka": "Kathode",
+        }
+    },
+    "LED Red": {
+        "pins": {
+            "an": "Anode",
+            "ka": "Kathode",
+        }
+    },
+    "Resistor 10k": {
+        "resistance": "10k Ohm",
+        "pins": {
+            "an": "Anode",
+            "ka": "Kathode",
+        },
+    },
+    "Resistor 1k": {
+        "resistance": "1k Ohm",
+        "pins": {
+            "an": "Anode",
+            "ka": "Kathode",
+        },
+    },
+    "Resistor 100": {
+        "resistance": "100 Ohm",
+        "pins": {
+            "an": "Anode",
+            "ka": "Kathode",
+        },
+    },
+}
+
+
+def list_components():
+    """
+    List available components you can use to build complex systems.
+    If you want more information on how to use them, use your web search tool.
+
+    Retruns:
+        list[str]: A list of names of components
+    """
+    return list(COMPONENT_INFO_DICT.keys())
+
+
+def component_info(component_name: str):
+    """
+    Get information on a component.
+    For the moment this only includes the pins of the component.
+
+    Args:
+        component_name (str): Name of the component
+
+    Returns:
+        dict[str, str]: A dictionary of the pins and a description
+    """
+    if not component_name in COMPONENT_INFO_DICT.keys():
+        return {
+            "error": "Component with this name does not exist",
+        }
+    return {"success": True, "component_info": COMPONENT_INFO_DICT[component_name]}
+
+
+def add_component(tool_context: ToolContext, component_name: str, id: str):
+    """
+    Add a component to the diagram. You need to write that component later.
+    Use list components to find out which components you are allowed to use.
+
+    Args:
+        component_name (str): The name of the component you want to add.
+        id (str): The id of the component you use to refer to the component later.
+    """
+    if not component_name in COMPONENT_INFO_DICT.keys():
+        return {
+            "error": "The component you wanted is not available in the list of allowed components"
+        }
+    if not "components" in tool_context.state:
+        tool_context.state["components"] = {}
+    components = tool_context.state["components"]
+    components[id] = component_name
+    tool_context.state["components"] = components
+    return {"success": True}
+
+
+def add_wire(
+    tool_context: ToolContext, start_id: str, start_pin: str, end_id: str, end_pin: str
+):
+    """
+    Add a wire between two components in our schematic.
+    Note that a component has to first be added to the schematic.
+
+    Args:
+        start_id (str): The id of the component where the wire starts.
+        start_pin (str): The pin on the start component of the wire.
+        end_id (str): The id of the component where the wire ends.
+        end_pin (str): The pin on the end component of the wire.
+    """
+    if not "wires" in tool_context.state:
+        tool_context.state["wires"] = []
+    if not "components" in tool_context.state:
+        tool_context.state["components"] = {}
+    if not start_id in tool_context.state["components"]:
+        return {
+            "error": f"Component with id {start_id} not added to list of components"
+        }
+    if not end_id in tool_context.state["components"]:
+        return {"error": f"Component with id {end_id} not added to list of components"}
+    wires = tool_context.state["wires"]
+    wires.append(
+        {
+            "start_id": start_id,
+            "start_pin": start_pin,
+            "end_id": end_id,
+            "end_pin": end_pin,
+        }
+    )
+    tool_context.state["wires"] = wires
+    return {
+        "success": True,
+    }
+
+
+def del_component(
+    tool_context: ToolContext,
+    id: str,
+):
+    """
+    Delete a component from the schematic.
+    If you delete a component, all wires connected to it will be deleted as well.
+
+    Args:
+        id (str): The id of the component you want to delete.
+    """
+    if not "components" in tool_context.state:
+        return {"error": "No components to delete"}
+    components = tool_context.state["components"]
+    if not id in components:
+        return {"error": f"Component with id {id} not found"}
+    del components[id]
+    tool_context.state["components"] = components
+    wires = tool_context.state["wires"]
+    if not wires:
+        return {
+            "success": True,
+        }
+    while True:
+        removed = False
+        for wire in wires:
+            if wire["start_id"] == id or wire["end_id"] == id:
+                removed = True
+                wires.remove(wire)
+                break
+        if not removed:
+            break
+
+    tool_context.state["wires"] = wires
+    return {
+        "success": True,
+    }
+
+
+def del_wire(
+    tool_context: ToolContext,
+    start_id: str,
+    start_pin: str,
+    end_id: str,
+    end_pin: str,
+):
+    """
+    Delete a wire between two components in our schematic.
+
+    Args:
+        start_id (str): The id of the component where the wire starts.
+        start_pin (str): The pin on the start component of the wire.
+        end_id (str): The id of the component where the wire ends.
+        end_pin (str): The pin on the end component of the wire.
+    """
+    if not "wires" in tool_context.state:
+        return {"error": "No wires to delete"}
+    if not "components" in tool_context.state:
+        return {"error": "No components to delete"}
+    wires = tool_context.state["wires"]
+    for wire in wires:
+        if (
+            wire["start_id"] == start_id
+            and wire["start_pin"] == start_pin
+            and wire["end_id"] == end_id
+            and wire["end_pin"] == end_pin
+        ):
+            wires.remove(wire)
+            break
+    else:
+        return {"error": f"Wire with id {start_id} not found"}
+    tool_context.state["wires"] = wires
+    return {
+        "success": True,
+    }
+
+
+def compile_with_arduino(
+    tool_context: ToolContext,
+):
+    """
+    Compile the code with the Arduino CLI.
+    """
+
+    code = tool_context.state.get("code")
+    if not code:
+        return {"error": "No code to compile"}
+
+
+    shutil.rmtree('/tmp/arduino_project')
+    os.mkdir('/tmp/arduino_project')    
+
+    with open("/tmp/arduino_project/arduino_project.ino", "w") as f:
+        f.write(code)
+
+    # Compile with the Arduino CLI
+    result = subprocess.run(
+        ["arduino-cli", "compile", "--fqbn", "arduino:avr:uno", "/tmp/arduino_project", "--build-path", "/tmp/arduino_project/output"],
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        return {
+            "error": "Compilation failed",
+            "output": result.stderr.decode("utf-8"),
+        }
+    return {
+        "success": True,
+        "output": result.stdout.decode("utf-8"),
+    }
+
+def search_library(
+    name: str,
+):
+    """
+    Search for a library with the Arduino CLI.
+    Args:
+        name (str): The name of the library you want to search for.
+    
+    Returns:
+        list[str]: A list of libraries that match the search term.
+    """
+
+    # Search for the library with the Arduino CLI
+    result = subprocess.run(
+        ["arduino-cli", "lib", "search", name],
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        return {
+            "error": "Library search failed",
+            "output": result.stderr.decode("utf-8"),
+        }
+    output = result.stdout.decode("utf-8").splitlines()[:25]
+    return {
+        "success": True,
+        "output": "\n".join(output),
+    }
+
+def add_library(name: str):
+    """
+    Install a library for the Arduino CLI.
+
+    Args:
+        name (str): The name of the library you want to install.
+    """
+    # Install the library with the Arduino CLI
+    result = subprocess.run(
+        ["arduino-cli", "lib", "install", name],
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        return {
+            "error": "Library installation failed",
+            "output": result.stderr.decode("utf-8"),
+        }
+    return {
+        "success": True,
+        "output": result.stdout.decode("utf-8"),
+    }
+
+
+root_agent = Agent(
+    name="coding_agent",
+    model="gemini-2.5-flash-preview-04-17",
+    description=("Agent to write Arduino code."),
+    instruction=(
+        "You are an expert programming agent tasked with designing a system involvin an Arduino and aditional hardware."
+        "You are required to wire up everythinb by yourself. It's not an option to just add the components and write teh code."
+        "When writing code always use the appropriate tools you where given to ouptut the code. Don't output any code in the chat."
+        "When you are done, compile the code with the Arduino CLI, without asking for confirmation."
+    ),
+    tools=[
+        FunctionTool(func=write_code),
+        FunctionTool(func=read_code),
+        FunctionTool(func=list_components),
+        FunctionTool(func=add_component),
+        FunctionTool(func=list_components),
+        FunctionTool(func=component_info),
+        FunctionTool(func=del_component),
+        FunctionTool(func=add_wire),
+        FunctionTool(func=del_wire),
+        FunctionTool(func=compile_with_arduino),
+        FunctionTool(func=search_library),
+        FunctionTool(func=add_library),
+    ],
+)
+
+session_service = InMemorySessionService()
+
+APP_NAME = "weather_tutorial_app"
+USER_ID = "user_1"
+SESSION_ID = "session_001"
+
+session = session_service.create_session(
+    app_name=APP_NAME,
+    user_id=USER_ID,
+    session_id=SESSION_ID,
+)
+
+runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)
+
+
+async def call_agent_async(query: str, runner, user_id, session_id):
+    print(f"\n>>> User Query: {query}")
+
+    content = types.Content(role="user", parts=[types.Part(text=query)])
+    final_response_text = "Agent did not produce a final response."
+
+    async for event in runner.run_async(
+        user_id=user_id, session_id=session_id, new_message=content
+    ):
+        print(
+            f"  [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}"
+        )
+        if event.is_final_response():
+            if event.content and event.content.parts:
+                final_response_text = event.content.parts[0].text
+
+    print(f"<<< Agent Response: {final_response_text}")
+
+
+async def run_conversation():
+    await call_agent_async(
+        "What is the current time in new york?",
+        runner=runner,
+        user_id=USER_ID,
+        session_id=SESSION_ID,
+    )
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(run_conversation())
+    except Exception as e:
+        print(f"An error occured: {e}")

main.py

@@ -0,0 +1,237 @@
+import gradio as gr
+from agent import session_service, root_agent, APP_NAME, runner
+from google.genai import types
+from pprint import pprint, pformat
+from uuid import uuid4
+import os
+import tempfile
+
+
+def generate_mermaid(components, wires=[], direction: str = "LR") -> str:
+    """
+    Build a Mermaid-JS flow-chart for a simple “high-level” schematic.
+
+    Parameters
+    ----------
+    components : list[dict]
+        Each dict needs the keys
+            • 'component_name' – human-readable label (e.g. "Arduino Uno R3")
+            • 'id'             – unique node ID used in the diagram
+    wires : list[dict]
+        Each dict needs the keys
+            • 'start_id', 'start_pin' – origin node & pin
+            • 'end_id',   'end_pin'   – destination node & pin
+    direction : str, optional
+        Flow direction for the chart header; one of "LR", "RL", "TB", "TD".
+        Defaults to "LR" (left-to-right).
+
+    Returns
+    -------
+    str
+        A Markdown-fenced Mermaid block you can drop straight into GitHub,
+        Obsidian, Notion, etc.
+    """
+
+    mermaid = [f"```mermaid", f"flowchart {direction}"]
+
+    # 1) Declare nodes
+    for node_id, node_name in components.items():
+        label = f"{node_name}\n({node_id})"
+        mermaid.append(f'    {node_id}["{label}"]')
+
+    # 2) Draw labelled connections
+    for wire in wires:
+        start = wire["start_id"]
+        end = wire["end_id"]
+        label = f'{wire["start_pin"]} → {wire["end_pin"]}'
+        mermaid.append(f'    {start} -- "{label}" --> {end}')
+
+    mermaid.append("```")
+    return "\n".join(mermaid)
+
+
+async def interact_with_agent(
+    prompt,
+    history: list[gr.ChatMessage],
+    session_id: gr.State,
+    local_storage: gr.BrowserState,
+    request: gr.Request,
+):
+    messages = []
+    messages.append(
+        gr.ChatMessage(
+            role="assistant", content="Thinking...", metadata={"status": "pending"}
+        )
+    )
+    yield messages, None, None, None
+    # yield messages
+
+    user_id = local_storage[0]
+
+    print(user_id)
+    print(session_id)
+
+    session = session_service.get_session(
+        app_name=APP_NAME, user_id=user_id, session_id=session_id
+    )
+
+    if session is None:
+        session = session_service.create_session(
+            app_name=APP_NAME, user_id=user_id, session_id=session_id
+        )
+
+    fetch_code = lambda: session_service.get_session(
+        app_name=APP_NAME, user_id=user_id, session_id=session_id
+    ).state.get("code")
+    fetch_components = lambda: session_service.get_session(
+        app_name=APP_NAME, user_id=user_id, session_id=session_id
+    ).state.get("components")
+    fetch_wires = lambda: session_service.get_session(
+        app_name=APP_NAME, user_id=user_id, session_id=session_id
+    ).state.get("wires")
+
+    def mermaid_assemble():
+        components = fetch_components()
+        wires = fetch_wires()
+        if components == None:
+            return gr.Markdown("No components yet")
+        if wires == None:
+            wires = []
+        mermaid = generate_mermaid(components, wires)
+        return gr.Markdown(mermaid)
+
+    content = types.Content(role="user", parts=[types.Part(text=prompt)])
+
+    temp = None
+
+    async for event in runner.run_async(
+        user_id=user_id, session_id=session_id, new_message=content
+    ):
+        """
+        print(
+            f"  [Event] Author: {event.author}, Type: {type(event).__name__}, Final: {event.is_final_response()}, Content: {event.content}"
+        )
+        """
+        for call in event.get_function_calls():
+            messages.append(
+                gr.ChatMessage(
+                    role="assistant",
+                    content=(
+                        f"Invoking {call.name} with args: \n {pformat(call.args)}"
+                        if not call.name == "write_code"
+                        else gr.Code(call.args["code"], language="cpp")
+                    ),
+                    metadata={"title": f"🛠️ Used tool {call.name}"},
+                )
+            )
+            yield messages, None, None, None
+        for call in event.get_function_responses():
+            if call.name == "compile_with_arduino" and call.response["success"]:
+                temp = tempfile.gettempdir()
+
+                temp_uuid = str(uuid4())
+                temp = os.path.join(temp, f"{temp_uuid}.ino.elf")
+
+
+                with open(temp, "wb") as f:
+                    with open(
+                        "/tmp/arduino_project/output/arduino_project.ino.elf", "rb"
+                    ) as f2:
+                        f.write(f2.read())
+
+                yield messages, None, None, gr.DownloadButton(
+                    "Download ELF",
+                    temp,
+                    variant="primary",
+                    render=False,
+                    interactive=True,
+                )
+        if event.is_final_response():
+            messages.append(
+                gr.ChatMessage(role="assistant", content=event.content.parts[0].text)
+            )
+        yield messages, gr.Code(
+            language="cpp", value=fetch_code()
+        ), mermaid_assemble(), (
+            None
+            if not temp
+            else gr.DownloadButton(
+                "Download ELF",
+                temp,
+                variant="primary",
+                render=False,
+                interactive=True,
+            )
+        )
+
+    session = session_service.get_session(
+        app_name=APP_NAME, user_id=user_id, session_id=session_id
+    )
+
+    pprint(session.state)
+
+
+with gr.Blocks() as demo:
+    session_id = gr.State(str(uuid4()))
+    code = gr.Code(language="cpp", render=False)
+    markdown_schematic = gr.Markdown(render=False)
+    local_storage = gr.BrowserState([str(uuid4())])
+    download_button = gr.DownloadButton(
+        "Download ELF",
+        "/tmp/arduino_project/output/arduino_project.ino.elf",
+        interactive=False,
+        render=False,
+    )
+
+    with gr.Row():
+        gr.Markdown(
+            """
+# Arduino Hardware and Software
+                    """
+        )
+    with gr.Row():
+        with gr.Column():
+            chatbot = gr.Chatbot(label="Agent", type="messages")
+
+            def clear_chat():
+                return [
+                    str(uuid4()),
+                    gr.Code(language="cpp", render=False),
+                    gr.Markdown(render=False),
+                    gr.DownloadButton(
+                        "Download ELF",
+                        "/tmp/arduino_project/output/arduino_project.ino.elf",
+                        render=False,
+                        interactive=False,
+                    ),
+                ]
+
+            chatbot.clear(
+                lambda: clear_chat(),
+                outputs=[session_id, code, markdown_schematic, download_button],
+            )
+            chat_interface = gr.ChatInterface(
+                interact_with_agent,
+                chatbot=chatbot,
+                type="messages",
+                examples=[
+                    [
+                        'Design a schematic and write code for an arduino driven "Joke of the day machine" where the user presses a physical button and then a random joke of the day is displayed on the display.'
+                    ],
+                    ["Write a Hello World program for an Arduino Uno"],
+                    ["Write a program to blink an LED on an Arduino Uno"],
+                ],
+                additional_inputs=[session_id, local_storage],
+                additional_outputs=[code, markdown_schematic, download_button],
+                fill_height=True,
+                show_progress="full",
+            )
+        with gr.Column():
+            code.render()
+            markdown_schematic.render()
+
+    with gr.Row():
+        download_button.render()
+
+    # mm
+demo.launch()

pyproject.toml

@@ -0,0 +1,10 @@
+[project]
+name = "arducoder-prototype"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "google-adk>=0.5.0",
+    "gradio>=5.29.1",
+]