91 lines
11 KiB
Python
91 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import gi
|
|
gi.require_version('Gtk', '4.0')
|
|
gi.require_version('Adw', '1')
|
|
from gi.repository import Adw, Gio, GLib, Gtk
|
|
|
|
CONFIG = Path(os.environ.get('XDG_CONFIG_HOME', Path.home() / '.config')) / 'jarvis' / 'config.json'
|
|
PAGES = [('general', 'General', 'Wake phrase, language, startup'), ('voice', 'Voice', 'Wake model, voice enrollment, preview'), ('models', 'Models', 'GPU runtime, fit, downloads'), ('memory', 'Memory', 'RAG workspaces and retention'), ('skills', 'Skills', 'Harness tools and confirmations'), ('computer', 'Computer use', 'Portal permissions and budgets'), ('privacy', 'Privacy', 'Storage, traces, deletion'), ('lab', 'Lab', 'BCI, VLA, ABot-World'), ('about', 'About', 'Versions and diagnostics')]
|
|
|
|
class Store:
|
|
def __init__(self):
|
|
try: self.data = json.loads(CONFIG.read_text())
|
|
except Exception: self.data = {}
|
|
def set(self, key, value):
|
|
self.data[key] = value; CONFIG.parent.mkdir(parents=True, exist_ok=True); CONFIG.write_text(json.dumps(self.data, indent=2) + '\n')
|
|
|
|
class ControlCenter(Adw.Application):
|
|
def __init__(self): super().__init__(application_id='io.qvac.Jarvis.Control', flags=Gio.ApplicationFlags.DEFAULT); self.store = Store(); self.rows = {}; self.proxy = None
|
|
def do_activate(self):
|
|
if getattr(self, 'window', None): self.window.present(); return
|
|
self.window = Adw.ApplicationWindow(application=self, title='Jarvis Control Center', default_width=1080, default_height=700)
|
|
self.window.set_content(self.build_shell()); self.connect_daemon(); self.window.present()
|
|
def build_shell(self):
|
|
split = Adw.NavigationSplitView(); sidebar = Adw.NavigationPage(title='Jarvis', child=self.build_sidebar(split)); content = Adw.NavigationPage(title='General', child=self.build_page('general')); split.set_sidebar(sidebar); split.set_content(content); self.content = content; self.split = split; return split
|
|
def build_sidebar(self, split):
|
|
box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12); box.set_margin_top(18); box.set_margin_bottom(18); box.set_margin_start(12); box.set_margin_end(12)
|
|
heading = Gtk.Label(label='JARVIS', xalign=0); heading.add_css_class('title-2'); box.append(heading)
|
|
local = Gtk.Label(label='LOCAL · QVAC MASTER', xalign=0); local.add_css_class('dim-label'); box.append(local)
|
|
listbox = Gtk.ListBox(selection_mode=Gtk.SelectionMode.SINGLE); listbox.add_css_class('navigation-sidebar');
|
|
for key, label, _subtitle in PAGES:
|
|
row = Gtk.ListBoxRow(); row.set_child(Gtk.Label(label=label, xalign=0, margin_top=10, margin_bottom=10, margin_start=12, margin_end=12)); row.set_name(key); listbox.append(row)
|
|
listbox.connect('row-selected', lambda _list, row: self.select_page(split, row.get_name() if row else 'general')); listbox.select_row(listbox.get_row_at_index(0)); box.append(listbox)
|
|
self.runtime = Gtk.Label(label='Connecting to jarvisd…', xalign=0, wrap=True); self.runtime.add_css_class('dim-label'); box.append(self.runtime); return box
|
|
def select_page(self, split, key): self.content.set_child(self.build_page(key)); self.content.set_title(dict((p[0], p[1]) for p in PAGES)[key])
|
|
def build_page(self, key):
|
|
page = Gtk.ScrolledWindow(); page.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC); root = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=18); root.set_margin_top(28); root.set_margin_bottom(28); root.set_margin_start(32); root.set_margin_end(32); page.set_child(root)
|
|
title = next(p for p in PAGES if p[0] == key); header = Gtk.Label(label=title[1], xalign=0); header.add_css_class('title-1'); root.append(header); sub = Gtk.Label(label=title[2], xalign=0); sub.add_css_class('dim-label'); root.append(sub)
|
|
builders = {'general': self.general, 'voice': self.voice, 'models': self.models, 'memory': self.memory, 'skills': self.skills, 'computer': self.computer, 'privacy': self.privacy, 'lab': self.lab, 'about': self.about}; builders[key](root); return page
|
|
def group(self, root, title, subtitle=''):
|
|
group = Adw.PreferencesGroup(title=title, description=subtitle); root.append(group); return group
|
|
def entry(self, group, key, title, value=''):
|
|
row = Adw.EntryRow(title=title, text=str(self.store.data.get(key, value))); row.connect('changed', lambda item: self.store.set(key, item.get_text())); group.add(row); return row
|
|
def switch(self, group, key, title, active=True):
|
|
row = Adw.SwitchRow(title=title, active=bool(self.store.data.get(key, active))); row.connect('notify::active', lambda item, _pspec: self.store.set(key, item.get_active())); group.add(row); return row
|
|
def button(self, group, title, label, callback):
|
|
row = Adw.ActionRow(title=title); button = Gtk.Button(label=label, valign=Gtk.Align.CENTER); button.connect('clicked', callback); row.add_suffix(button); group.add(row); return row
|
|
def general(self, root):
|
|
group = self.group(root, 'General'); self.entry(group, 'wake_phrase', 'Wake phrase', 'hey jarvis'); self.entry(group, 'language', 'Language', 'en-US'); self.entry(group, 'hotkey', 'Hotkey', '<Super><Space>'); self.switch(group, 'startup', 'Start Jarvis when I log in', True)
|
|
def voice(self, root):
|
|
group = self.group(root, 'Voice'); self.entry(group, 'voice_id', 'Voice ID'); self.entry(group, 'voice_reference', 'Reference WAV path'); self.entry(group, 'wake_command', 'Local wake bridge command'); self.switch(group, 'tts_enabled', 'Enable spoken replies', True); self.button(group, 'Voice enrollment', 'Enroll reference', lambda _b: self.call('IngestPath', '(s)', [self.store.data.get('voice_reference', '')])); self.button(group, 'Preview voice', 'Speak preview', lambda _b: self.call('Say', '(s)', ['Jarvis voice preview.'])); self.button(group, 'Wake model', 'Refresh runtime', lambda _b: self.call('GetRuntimeStatus'))
|
|
def models(self, root):
|
|
group = self.group(root, 'Single QVAC master'); self.model_status = Gtk.Label(label='Loading master status…', xalign=0, wrap=True); group.add(Adw.ActionRow(title='Runtime status', child=self.model_status)); self.entry(group, 'model', 'Active model', 'qwen3.5-4b'); self.button(group, 'Assess model fit', 'Assess', lambda _b: self.assess()); self.button(group, 'Model lifecycle', 'Download / resume', lambda _b: self.download()); self.button(group, 'Model lifecycle', 'Pause / cancel', lambda _b: self.call('CancelModel', '(s)', [self.store.data.get('model', 'qwen3.5-4b')]))
|
|
capabilities = self.group(root, 'Capabilities', 'Every capability remains visible when its model does not fit the GPU profile.')
|
|
for label in ['Chat · Plan · Summarize · Rewrite · Code', 'Embeddings · RAG · Batch prompts', 'Vision · OCR · Classification', 'Image · Video · Music generation', 'ASR · TTS · Translation · Voice clone', 'LoRA · BCI · VLA · ABot-World']:
|
|
capabilities.add(Adw.ActionRow(title=label, subtitle='Status supplied by the single QVAC master'))
|
|
def memory(self, root):
|
|
group = self.group(root, 'RAG workspaces'); self.entry(group, 'rag_workspace', 'Workspace name', 'home'); self.entry(group, 'rag_path', 'Folder to ingest'); self.button(group, 'Ingest local folder', 'Ingest', lambda _b: self.call('IngestPath', '(s)', [self.store.data.get('rag_path', '')])); self.button(group, 'Memory controls', 'Refresh workspaces', lambda _b: self.call('GetRuntimeStatus'))
|
|
def skills(self, root):
|
|
group = self.group(root, 'Harness permissions'); self.switch(group, 'confirm_destructive', 'Confirm destructive actions', True); self.switch(group, 'skills_computer', 'Enable computer-use tools', False); self.switch(group, 'skills_media', 'Enable media tools', True)
|
|
def computer(self, root):
|
|
group = self.group(root, 'Computer use'); self.switch(group, 'computer_observe', 'Observe-only mode', False); self.switch(group, 'computer_full', 'Full computer-use mode', False); self.entry(group, 'computer_step_budget', 'Step budget', '20'); self.button(group, 'Portal grant', 'Grant computer access', lambda _b: self.call('ComputerGrant', '(b)', [True])); self.button(group, 'Portal grant', 'Revoke immediately', lambda _b: self.call('ComputerRevoke'))
|
|
def privacy(self, root):
|
|
group = self.group(root, 'Privacy'); self.switch(group, 'save_transcripts', 'Save transcripts locally', False); self.switch(group, 'save_cu_traces', 'Save computer-use traces', False); self.switch(group, 'mute_schedule', 'Honor mute schedule', True); self.button(group, 'Computer traces', 'Delete temporary frames', lambda _b: self.call('WipeComputerTraces'))
|
|
def lab(self, root):
|
|
group = self.group(root, 'Research capabilities'); self.switch(group, 'lab_bci', 'BCI transcription slot', False); self.switch(group, 'lab_vla', 'VLA desktop actuator slot', False); self.switch(group, 'lab_world', 'ABot-World sandbox', False); self.entry(group, 'p2p_relays', 'Optional P2P relays')
|
|
def about(self, root):
|
|
group = self.group(root, 'About'); group.add(Adw.ActionRow(title='QVAC', subtitle='0.19.x · single local master')); group.add(Adw.ActionRow(title='Harness', subtitle=str(Path.home() / 'dev/agent-harness'))); group.add(Adw.ActionRow(title='Computer-use', subtitle='Wayland portal · AT-SPI · libei')); self.button(group, 'Diagnostics', 'Refresh status', lambda _b: self.call('GetRuntimeStatus'))
|
|
def connect_daemon(self):
|
|
try: self.proxy = Gio.DBusProxy.new_for_bus_sync(Gio.BusType.SESSION, Gio.DBusProxyFlags.NONE, None, 'io.qvac.Jarvis', '/io/qvac/Jarvis', 'io.qvac.Jarvis.Session'); self.call('GetRuntimeStatus')
|
|
except Exception as exc: self.runtime.set_text(f'jarvisd unavailable: {exc}')
|
|
def call(self, method, signature=None, args=None):
|
|
if not self.proxy: return
|
|
try:
|
|
self.runtime.set_text(f'{method}…')
|
|
self.proxy.call(method, GLib.Variant(signature, args) if signature else None, Gio.DBusCallFlags.NONE, 5000, None, self._call_done, method)
|
|
except Exception as exc: self.runtime.set_text(f'{method}: {exc}')
|
|
def _call_done(self, proxy, result, method):
|
|
try: self.update_status(proxy.call_finish(result))
|
|
except Exception as exc: self.runtime.set_text(f'{method}: {exc}')
|
|
def update_status(self, result):
|
|
value = result.unpack()[0] if result else ''; self.runtime.set_text('Connected · local QVAC master');
|
|
if hasattr(self, 'model_status'): self.model_status.set_text(value)
|
|
def assess(self): self.call('AssessModelFit', '(s)', [self.store.data.get('model', 'qwen3.5-4b')])
|
|
def download(self): self.call('DownloadModel', '(s)', [self.store.data.get('model', 'qwen3.5-4b')])
|
|
|
|
def main(): return ControlCenter().run(None)
|
|
if __name__ == '__main__': main()
|