app.py

"""
llmOS intent compiler -- a browser front end for the on-device engine.

This does not re-implement the model in Python. It shells out to `llmos-cli`,
the firmware's own C inference engine, reading the same quantized weights the
SD card carries. So what you see here is what the board does, minus the
wall-clock latency of a 240 MHz Xtensa core.

The one thing Python does is render the English gloss. The firmware describes a
rule in Persian (`dsl_describe_fa`) because that is the language it ships in;
rather than pretend otherwise, the English sentence below is rendered here,
from the DSL the C engine produced. The compilation -- the part that matters --
is C in both tabs.
"""
import json
import os
import re
import subprocess

import gradio as gr

MODEL_DIR = os.environ.get("LLMOS_MODEL_DIR", "model")
CLI = os.environ.get("LLMOS_CLI", "./llmos-cli")
LABELS_PATH = os.environ.get("LLMOS_LABELS", "labels.json")

GITHUB = "https://github.com/AliAkrami1375/llmos-firmware"
MODEL_REPO = "https://huggingface.co/DibaAi/llmOS"

try:
    with open(LABELS_PATH, encoding="utf-8") as f:
        LABELS = {int(k): v for k, v in json.load(f).items()}
except Exception:
    LABELS = {}

EXAMPLES_FA = [
    "اگر سنسور پایه ۱ روشن شد پایه ۲ را خاموش کن",
    "هر ۳۰ ثانیه پایه ۵ را چشمک بزن",
    "وقتی سنسور نور پایه ۴ کمتر از ۸۰۰ شد چراغ پایه ۷ را روشن کن",
    "ساعت ۷ صبح بازر پایه ۹ را ۳۰ ثانیه بزن",
    "پایه ۱۶ رو بذار چراغ آشپزخانه",
    "اتاق علی رو روشن کن",
    "اگر اتاق علی روشن شد چراغ حیاط رو خاموش کن",
    "به وای فای خانه_من با رمز 1a2b3c وصل شو",
    "شبکه های وای فای رو اسکن کن",
    "میشه اگه ورودی پورت ۱۵ غیر فعال شد پیم شماره ۹ صفر کن؟",
    "سلام خوبی؟",
]

# English only gets rows that were verified against this exact checkpoint.
# Pin naming and word-form times ("at 7am") are Persian-only in practice, so
# they are not offered here -- an example that misfires teaches the wrong
# lesson about the model.
EXAMPLES_EN = [
    "if pin 1 goes high, turn off pin 2",
    "toggle pin 5 every 30 seconds",
    "every 5 minutes toggle the led on pin 7",
    "when the light sensor on pin 4 drops below 800, turn on pin 7",
    "at 07:00 turn on pin 9 for 30 seconds",
    "turn on pin 16 now",
    "connect to wifi MyHome with password 1a2b3c",
    "read pin 5",
    "hello, how are you?",
]


# --- English rendering of a rule -------------------------------------------

UNITS = {"ms": ("millisecond", 1), "s": ("second", 1), "m": ("minute", 1),
         "h": ("hour", 1)}


def _pin(n):
    """Pin 7 with a name reads as 'Ali's room', otherwise as 'pin 7'."""
    name = LABELS.get(int(n))
    return f'"{name}" (pin {n})' if name else f"pin {n}"


def _duration(text):
    m = re.fullmatch(r"(\d+)(ms|s|m|h)", text)
    if not m:
        return text
    n, u = int(m.group(1)), m.group(2)
    word = UNITS[u][0]
    return f"{n} {word}" + ("s" if n != 1 else "")


def _args(rest):
    return re.findall(r'"([^"]*)"', rest or "")


def _call_en(rest):
    tool = rest.split(" ", 1)[0].strip()
    args = _args(rest)
    if not args:
        return f"run {tool}"
    return f"run {tool} with " + ", ".join(f'"{a}"' for a in args)


def _action_en(a):
    a = a.strip()
    m = re.fullmatch(r"GPIO(\d+)=([01T])", a)
    if m:
        pin, lvl = m.group(1), m.group(2)
        verb = {"0": "turn off", "1": "turn on", "T": "toggle"}[lvl]
        return f"{verb} {_pin(pin)}"
    m = re.fullmatch(r"WAIT (\S+)", a)
    if m:
        return f"wait {_duration(m.group(1))}"
    if a.startswith("CALL "):
        return _call_en(a[5:])
    return a


def _trigger_en(t):
    t = t.strip()
    m = re.fullmatch(r"GPIO(\d+)=([01])", t)
    if m:
        state = "goes high" if m.group(2) == "1" else "goes low"
        return f"whenever {_pin(m.group(1))} {state}"
    m = re.fullmatch(r"ADC(\d+)([<>])(\d+)", t)
    if m:
        rel = "drops below" if m.group(2) == "<" else "rises above"
        return f"whenever the sensor on {_pin(m.group(1))} {rel} {m.group(3)}"
    m = re.fullmatch(r"EVERY (\S+)", t)
    if m:
        return f"every {_duration(m.group(1))}"
    m = re.fullmatch(r"AT (\d{2}):(\d{2})", t)
    if m:
        return f"every day at {m.group(1)}:{m.group(2)}"
    return t


def _cap(s):
    """Upper-case the first letter only. str.capitalize() would lower-case the
    rest, and the rest can be a WiFi password."""
    return s[:1].upper() + s[1:] if s else s


def describe_en(dsl):
    """Render a DSL rule as an English sentence. Presentation only."""
    dsl = (dsl or "").strip()
    if not dsl or dsl == "ERR":
        return ""

    m = re.fullmatch(r'NAME GPIO(\d+) "(.*)"', dsl)
    if m:
        return f'Pin {m.group(1)} is now called "{m.group(2)}".'

    if dsl.startswith("CALL "):
        return _cap(_call_en(dsl[5:])) + "."

    m = re.fullmatch(r"ON (.+?) DO (.+)", dsl)
    if m:
        head, body = _trigger_en(m.group(1)), m.group(2)
    elif dsl.startswith("DO "):
        head, body = "", dsl[3:]
    else:
        return dsl

    steps = [_action_en(a) for a in body.split(";")]
    joined = ", then ".join(steps)
    return _cap(f"{head}, {joined}." if head else f"{joined}.")


# --- the engine -------------------------------------------------------------

MSG = {
    "en": {
        "empty": "Type something first.",
        "timeout": "The engine did not answer in time.",
        "missing": "The inference engine was not found.",
        "err": "This is not an automation request — and the model worked that "
               "out for itself.",
        "fail": "Not translated: ",
        "conf": "confidence {p}%",
        "ms": "{n} ms on this machine",
    },
    "fa": {
        "empty": "چیزی ننوشتید.",
        "timeout": "پاسخ به موقع نرسید.",
        "missing": "موتور استنتاج پیدا نشد.",
        "err": "این درخواست یک قانون اتوماسیون نیست — و مدل خودش تشخیص داده.",
        "fail": "ترجمه نشد: ",
        "conf": "اطمینان {p}٪",
        "ms": "{n} میلی‌ثانیه روی این ماشین",
    },
}


def compile_text(text, lang):
    t = MSG[lang]
    text = (text or "").strip()
    if not text:
        return "", "", t["empty"]

    cmd = [CLI, "--model", MODEL_DIR, "--text", text]
    if os.path.exists(LABELS_PATH):
        cmd += ["--labels", LABELS_PATH]
    try:
        out = subprocess.run(cmd, capture_output=True, text=True,
                             timeout=30).stdout
    except subprocess.TimeoutExpired:
        return "", "", t["timeout"]
    except FileNotFoundError:
        return "", "", t["missing"]

    def field(name):
        m = re.search(rf"^{name}\s*:\s*(.*)$", out, re.M)
        return m.group(1).strip() if m else ""

    dsl, fa, note, stats = (field("dsl"), field("fa"), field("note"),
                            field("stats"))

    if dsl == "ERR" or note.startswith("the request is not"):
        return "ERR", "", t["err"]
    if note and not fa:
        return dsl, "", t["fail"] + note

    plain = describe_en(dsl) if lang == "en" else fa

    conf = re.search(r"confidence ([\d.]+)", stats)
    ms = re.search(r"(\d+)ms", stats)
    detail = []
    if conf:
        detail.append(t["conf"].format(p=round(float(conf.group(1)) * 100)))
    if ms:
        detail.append(t["ms"].format(n=ms.group(1)))
    return dsl, plain, " • ".join(detail)


def compile_en(text):
    return compile_text(text, "en")


def compile_fa(text):
    return compile_text(text, "fa")


# --- interface --------------------------------------------------------------

CSS = """
.banner img { border-radius: 10px; }
footer { display: none !important; }
"""

INTRO_EN = f"""
### Say it, and watch the device turn it into a rule

Type a sentence in English or Persian and see what the board would do.

This is the **same C engine that runs on the ESP32-S3**, with the same
quantized weights the SD card carries. Syntactically invalid output is
**impossible**, not merely unlikely: at every step the decoder admits only
tokens that keep the text a valid prefix of the rule grammar, and the same test
holds every pin, threshold and duration inside its legal range.

A few pins already have names, so you can see naming work:
**"اتاق علی"** is pin 7, **"چراغ حیاط"** pin 3, **"بازر اتاق خواب"** pin 9.
Try `اگر اتاق علی روشن شد چراغ حیاط رو خاموش کن` in the Persian tab.

Each request here is independent, so a name you create with `NAME` is not
remembered for the next one. On the device it is written to the card and works
from that moment on.

**Persian is the primary language.** Naming a pin and writing a time in words
work there; in English, give the clock time in figures — `at 07:00`, not
"at 7am".

[Firmware and installer]({GITHUB}) · [Weights]({MODEL_REPO})
"""

INTRO_FA = f"""
<div dir="rtl">

### بگویید، و ببینید دستگاه چه قانونی می‌سازد

جمله‌ای به فارسی یا انگلیسی بنویسید و ببینید برد چه کار می‌کند.

این **همان موتور C است که روی ESP32-S3 اجرا می‌شود**، با همان وزن‌های
کوانتیزه‌شده‌ی روی کارت حافظه. خروجی نامعتبر از نظر نحوی **ممکن نیست** — نه
فقط بعید: دیکودر در هر گام تنها توکن‌هایی را مجاز می‌داند که متن را پیشوند
معتبری از گرامر قانون نگه دارند، و همان تست هر پایه و آستانه و مدت را داخل
بازه‌ی مجازش نگه می‌دارد.

چند پایه از پیش نام دارند تا نام‌گذاری را هم ببینید:
**اتاق علی** پایه ۷، **چراغ حیاط** پایه ۳، **بازر اتاق خواب** پایه ۹.

اینجا هر درخواست مستقل است، پس اسمی که با `NAME` می‌سازید برای درخواست بعدی
به یاد نمی‌ماند. روی دستگاه، اسم روی کارت نوشته می‌شود و از همان لحظه کار می‌کند.

[فرم‌ور و نصب‌کننده]({GITHUB}) · [وزن‌ها]({MODEL_REPO})

</div>
"""

FOOT_EN = """
---
2.85M parameters, int8, 3.1 MB. It runs on the microcontroller itself — no
internet, no server, no account. MIT licensed.
"""

FOOT_FA = """
<div dir="rtl">

---

مدل ۲٫۸۵ میلیون پارامتری، int8، ۳٫۱ مگابایت. روی خود میکروکنترلر اجرا
می‌شود — بدون اینترنت، بدون سرور، بدون حساب کاربری. مجوز MIT.

</div>
"""

with gr.Blocks(title="llmOS — intent compiler", theme=gr.themes.Soft(),
               css=CSS) as demo:
    with gr.Tabs():
        with gr.Tab("English"):
            gr.Image("banner-en.png", show_label=False, container=False,
                     show_download_button=False, show_fullscreen_button=False,
                     interactive=False, elem_classes="banner")
            gr.Markdown(INTRO_EN)
            inp_en = gr.Textbox(label="Request", lines=2,
                                placeholder="e.g. if pin 1 goes high, turn off pin 2")
            btn_en = gr.Button("Compile", variant="primary")
            dsl_en = gr.Textbox(label="Rule produced (DSL)", interactive=False)
            say_en = gr.Textbox(label="Which means", interactive=False)
            meta_en = gr.Markdown()
            gr.Examples(examples=EXAMPLES_EN, inputs=inp_en)
            gr.Markdown(FOOT_EN)

            btn_en.click(compile_en, inp_en, [dsl_en, say_en, meta_en])
            inp_en.submit(compile_en, inp_en, [dsl_en, say_en, meta_en])

        with gr.Tab("فارسی"):
            gr.Image("banner-fa.png", show_label=False, container=False,
                     show_download_button=False, show_fullscreen_button=False,
                     interactive=False, elem_classes="banner")
            gr.Markdown(INTRO_FA)
            inp_fa = gr.Textbox(label="درخواست", lines=2, rtl=True,
                                placeholder="مثال: اگر پایه ۱ روشن شد پایه ۲ را خاموش کن")
            btn_fa = gr.Button("ترجمه کن", variant="primary")
            dsl_fa = gr.Textbox(label="قانون تولیدشده (DSL)", interactive=False)
            say_fa = gr.Textbox(label="یعنی", interactive=False, rtl=True)
            meta_fa = gr.Markdown()
            gr.Examples(examples=EXAMPLES_FA, inputs=inp_fa)
            gr.Markdown(FOOT_FA)

            btn_fa.click(compile_fa, inp_fa, [dsl_fa, say_fa, meta_fa])
            inp_fa.submit(compile_fa, inp_fa, [dsl_fa, say_fa, meta_fa])

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860)