| """Gera, com a janela de VERDADE do openTARS (sem tela, Ollama de mentira), as sequências de quadros do filme "Um dia com o openTARS". |
| |
| Uso: python3 tests/gerar_frames_filme.py PASTA_DE_SAIDA (precisa de PySide6; escreve PASTA/<cena>/NNNN.jpg e PASTA/index.json) |
| |
| Nada aqui conversa com um Ollama de verdade: o pedido é encenado (os mesmos eventos que o núcleo emite) para que as |
| imagens mostrem sempre o mesmo texto, em inglês, e não dependam de modelo, GPU ou rede. |
| """ |
|
|
| import datetime as dt |
| import json |
| import os |
| import sys |
| import tempfile |
| import threading |
| import time |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| from pathlib import Path |
|
|
| os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") |
| os.environ.setdefault("QT_SCALE_FACTOR", "2") |
| CASA = tempfile.mkdtemp(prefix="opentars-capturas-") |
| os.environ["HOME"] = CASA |
| os.environ["TARS_IDIOMA"] = "en" |
| os.environ["TARS_ARQUIVO_SESSOES"] = os.path.join(CASA, "sessoes.json") |
| os.environ["TARS_ARQUIVO_NUVEM"] = os.path.join(CASA, "nuvem.json") |
| os.environ["TARS_ARQUIVO_PRIMEIRO_USO"] = os.path.join(CASA, "primeiro_uso.json") |
| os.environ["TARS_ARQUIVO_AGENDA"] = os.path.join(CASA, "agenda.json") |
| os.environ["TARS_DIR_LOG"] = os.path.join(CASA, "log") |
| os.environ["XDG_CONFIG_HOME"] = os.path.join(CASA, "config") |
| os.environ["TARS_SEM_AVISO_VERSAO"] = "1" |
| os.environ["TARS_ARQUIVO_PROBES"] = os.path.join(CASA, "probes.json") |
|
|
| SAIDA = Path(sys.argv[1]) |
| SAIDA.mkdir(parents=True, exist_ok=True) |
|
|
| from _util import carregar_tars |
|
|
| from PySide6.QtTest import QTest |
| from PySide6.QtWidgets import QPushButton |
|
|
| MODELOS = [("ProjectOpenTARS/Gargantua-2:latest", 1_600_000_000, "2.0B"), ("ProjectOpenTARS/Gargantua-2:lite", 617_000_000, "0.8B")] |
|
|
|
|
| class Falso(BaseHTTPRequestHandler): |
| def log_message(self, *a): |
| pass |
|
|
| def _json(self, dados, codigo=200): |
| corpo = json.dumps(dados).encode() |
| self.send_response(codigo) |
| self.send_header("Content-Type", "application/json") |
| self.send_header("Content-Length", str(len(corpo))) |
| self.end_headers() |
| self.wfile.write(corpo) |
|
|
| def do_GET(self): |
| if self.path == "/api/version": |
| return self._json({"version": "0.12.9"}) |
| if self.path == "/api/tags": |
| return self._json({"models": [{"name": n, "model": n, "size": s, "details": { |
| "family": "qwen3", "parameter_size": p, "quantization_level": "Q4_K_M"}} for n, s, p in MODELOS]}) |
| return self._json({"models": []} if self.path == "/api/ps" else {}, 200 if self.path == "/api/ps" else 404) |
|
|
| def do_POST(self): |
| self.rfile.read(int(self.headers.get("Content-Length", 0))) |
| if self.path == "/api/show": |
| return self._json({"capabilities": ["completion", "tools"], "details": {"family": "qwen3"}, "model_info": {}}) |
| return self._json({"done": True, "response": ""}) |
|
|
|
|
| servidor = ThreadingHTTPServer(("127.0.0.1", 0), Falso) |
| threading.Thread(target=servidor.serve_forever, daemon=True).start() |
| os.environ["OLLAMA_HOST"] = f"127.0.0.1:{servidor.server_address[1]}" |
|
|
| tars = carregar_tars() |
| tars.avisos_de_ambiente = lambda: [] |
| tars.detectar_gpu = lambda forcar=False: None |
| tars.aquecer_ajudante = lambda: None |
| tars.modelo_manual = None |
|
|
| import tars_historico |
| import tars_rotinas as agenda_lib |
| from tars_qt.primeiro_uso import criar_aplicativo |
| from tars_qt.principal import JanelaPrincipal |
| from tars_qt.conversa import CartaoAprovacao |
|
|
| app = criar_aplicativo(["opentars"]) |
|
|
|
|
| def esperar(cond, segundos=15): |
| fim = time.time() + segundos |
| while time.time() < fim: |
| app.processEvents() |
| if cond(): |
| return True |
| QTest.qWait(20) |
| return False |
|
|
|
|
| def foto(widget, nome, espera=500): |
| QTest.qWait(espera) |
| app.processEvents() |
| caminho = SAIDA / f"{nome}.png" |
| widget.grab().save(str(caminho)) |
| print("captura:", caminho.name) |
|
|
|
|
| |
| agora = [time.time()] |
| historico = tars_historico.Historico(os.path.join(CASA, "historico.json"), agora=lambda: agora[0]) |
| DIA = 86400 |
| for atras, titulo in ((6 * DIA, "Rename the photos in Pictures"), (3 * DIA, "What is using port 8080?"), |
| (1 * DIA, "Summarize the PDF I just downloaded"), (1 * DIA + 3600, "Open the project folder in the editor"), |
| (3600, "Organize my Downloads by file type")): |
| agora[0] = time.time() - atras |
| historico.salvar(historico.nova_id(), [{"role": "user", "content": titulo}, {"role": "assistant", "content": "Done."}]) |
| agora[0] = time.time() |
|
|
| tars.memoria.lembrar("my notes are in ~/Documents/Notes") |
| tars.memoria.lembrar("my code lives in ~/Projects") |
| tars.memoria.lembrar("I like short answers") |
|
|
| relogio = [dt.datetime.now().replace(microsecond=0)] |
| tars.agenda = agenda_lib.Agenda(os.path.join(CASA, "agenda.json"), agora=lambda: relogio[0]) |
| tars.agenda.criar("Morning news", "open the newspaper website and read me the headlines", |
| {"tipo": "diaria", "hora": "08:00"}, permitidas=(), ativa=True, aprovada=True) |
| tars.agenda.criar("Weekly cleanup", "move the files in Downloads older than 30 days to the trash", |
| {"tipo": "semanal", "hora": "18:30", "dias": [4]}, permitidas=("apagar",), ativa=True, aprovada=True) |
|
|
|
|
|
|
|
|
| |
| |
| |
| |
| import queue |
|
|
| from PySide6.QtCore import QPoint |
|
|
| PASTA = SAIDA |
| PASTA.mkdir(parents=True, exist_ok=True) |
| INDICE = {} |
|
|
|
|
| class Cena: |
| def __init__(self, nome, widget): |
| self.nome, self.w, self.n = nome, widget, 0 |
| self.frames, self.alvos = [], {} |
| (PASTA / nome).mkdir(exist_ok=True) |
|
|
| def foto(self, t, espera=70): |
| QTest.qWait(espera) |
| app.processEvents() |
| arq = f"{self.n:04d}.jpg" |
| self.w.grab().save(str(PASTA / self.nome / arq), "JPG", 90) |
| self.frames.append([round(t, 3), arq]) |
| self.n += 1 |
| self.tam = [self.w.width(), self.w.height()] |
|
|
| def alvo(self, nome, widget): |
| p = widget.mapTo(self.w, QPoint(0, 0)) |
| self.alvos[nome] = [p.x(), p.y(), widget.width(), widget.height()] |
|
|
| def fim(self): |
| INDICE[self.nome] = {"tam": self.tam, "frames": self.frames, "alvos": self.alvos} |
| (PASTA / "index.json").write_text(json.dumps(INDICE, indent=1), encoding="utf-8") |
| print("cena", self.nome, len(self.frames), "quadros") |
|
|
|
|
| def digitar(cena, caixa, texto, t0, cps=26, passo=2, fim_extra=0.0): |
| """Digita no campo real, um quadro a cada `passo` letras. Devolve o tempo em que terminou.""" |
| t = t0 |
| for k in range(0, len(texto) + 1, passo): |
| caixa.setPlainText(texto[:k]) |
| caixa.moveCursor(QTextCursor.MoveOperation.End) |
| cena.foto(t0 + k / cps, espera=25) |
| t = t0 + k / cps |
| if len(texto) % passo: |
| caixa.setPlainText(texto) |
| t = t0 + len(texto) / cps |
| cena.foto(t, espera=25) |
| return t + fim_extra |
|
|
|
|
| from PySide6.QtGui import QTextCursor |
|
|
| _liberar = threading.Event() |
|
|
|
|
| def _processar_travado(texto, anexos=None): |
| """O pedido 'está rodando' (a janela mostra Parar) até a cena liberar; quem emite os eventos é a cena, na thread principal.""" |
| _liberar.wait(120) |
| return None |
|
|
|
|
| tars.processar_mensagem = _processar_travado |
|
|
|
|
| def iniciar_pedido(cena, texto, t, cps=26): |
| """Digita o pedido e envia (Enter): devolve o tempo depois do envio.""" |
| _liberar.clear() |
| fim = digitar(cena, j.compositor.entrada, texto, t, cps) |
| cena.foto(fim + 0.35) |
| tars.conversa.append({"role": "user", "content": texto}) |
| j.enviar(texto) |
| esperar(lambda: j.ponte.esta_ocupado) |
| cena.foto(fim + 0.55) |
| return fim + 0.55 |
|
|
|
|
| def emitir(evento, **kw): |
| tars.emitir(evento, **kw) |
| app.processEvents() |
|
|
|
|
| def terminar_pedido(texto_resposta): |
| tars.conversa.append({"role": "assistant", "content": texto_resposta}) |
| emitir("fim_stream") |
| _liberar.set() |
| esperar(lambda: not j.ponte.esta_ocupado) |
| QTest.qWait(1600) |
| try: |
| j.mascote.balao.hide() |
| except AttributeError: |
| pass |
| app.processEvents() |
|
|
|
|
| def aprovar(cena, comando, t, clicar="Allow", t_clique=None): |
| """Mostra o cartão de aprovação REAL para `comando`, fotografa e clica no botão pedido.""" |
| r = {} |
| threading.Thread(target=lambda: r.setdefault("r", tars.confirmar_comando_hook(comando)), daemon=True).start() |
| esperar(lambda: j.area.findChildren(CartaoAprovacao)) |
| cartao = j.area.findChildren(CartaoAprovacao)[-1] |
| for k, dt_ in enumerate((0.0, 0.35, 0.7, 1.4)): |
| cena.foto(t + dt_) |
| botoes = {b.text(): b for b in cartao.findChildren(QPushButton)} |
| cena.alvo(clicar.lower(), botoes[clicar]) |
| return r, botoes[clicar], t + 1.4 |
|
|
|
|
| |
| j = JanelaPrincipal(tars, historico=historico, capturar_stdout=False) |
| j.resize(1180, 760) |
| j.show() |
| esperar(lambda: j._modelos) |
| c = Cena("inicio", j) |
| for k in range(8): |
| c.foto(k * 0.45, espera=450 if k else 600) |
| c.fim() |
|
|
| |
| j.nova_conversa() |
| c = Cena("tarefa", j) |
| c.foto(0.0) |
| t = iniciar_pedido(c, "Organize my Downloads folder by type", 0.6, cps=30) |
| emitir("tarefa", modo="AUTO", tarefa="acao", modelo="ProjectOpenTARS/Gargantua-2:latest") |
| emitir("plano", etapas=["Look at the Downloads folder (48 files)", "Plan: Images, Documents, Installers, Other", "Move the files", "Check that nothing was left behind"]) |
| c.foto(t + 0.5) |
| emitir("ferramenta", nome="execute_terminal", resumo="ls ~/Downloads") |
| c.foto(t + 1.0) |
| emitir("ferramenta_fim", nome="execute_terminal", sucesso=True, segundos=0.1) |
| emitir("ferramenta", nome="execute_terminal", resumo="mkdir Images Documents Installers Other && mv …") |
| c.foto(t + 1.6) |
| r, botao, t = aprovar(c, "mkdir -p ~/Downloads/{Images,Documents,Installers,Other} && mv ~/Downloads/*.jpg ~/Downloads/Images/", t + 2.0) |
| c.foto(t + 0.5) |
| botao.click() |
| esperar(lambda: "r" in r) |
| c.foto(t + 1.2) |
| emitir("ferramenta_fim", nome="execute_terminal", sucesso=True, segundos=1.8) |
| emitir("ferramenta", nome="execute_terminal", resumo="ls ~/Downloads | wc -l") |
| emitir("ferramenta_fim", nome="execute_terminal", sucesso=True, segundos=0.1) |
| c.foto(t + 1.9) |
| resp = ("Done. I moved **48 files** into 4 new folders inside Downloads: **Images** (21), **Documents** (17), **Installers** (6) and **Other** (4). " |
| "Nothing was deleted, and you can undo it from the history.") |
| emitir("resposta", texto=resp, inicio=True, apos_pensamento=False) |
| c.foto(t + 2.6) |
| terminar_pedido(resp) |
| c.foto(t + 3.4) |
| c.foto(t + 4.4) |
| c.fim() |
|
|
| |
| j.nova_conversa() |
| c = Cena("buscar", j) |
| c.foto(0.0) |
| t = iniciar_pedido(c, "How do I install Python?", 0.6, cps=30) |
| emitir("tarefa", modo="AUTO", tarefa="busca", modelo="ProjectOpenTARS/Gargantua-2:latest") |
| emitir("ferramenta", nome="search_web", resumo="how to install Python") |
| c.foto(t + 0.6) |
| emitir("ferramenta_fim", nome="search_web", sucesso=True, segundos=0.9) |
| c.foto(t + 1.4) |
| resp = ("On Ubuntu, Python 3 usually comes installed. Check it with:\n\n```bash\npython3 --version\n```\n\nIf it is missing, install it with `sudo apt install python3`. " |
| "**Sources:** docs.python.org (Python Setup and Usage) and python.org. Check the current version on the official site: I may be quoting an old one.") |
| emitir("resposta", texto=resp, inicio=True, apos_pensamento=False) |
| c.foto(t + 2.2) |
| terminar_pedido(resp) |
| c.foto(t + 3.0) |
| c.foto(t + 4.0) |
| c.fim() |
|
|
| |
| j.nova_conversa() |
| c = Cena("ver", j) |
| c.foto(0.0) |
| t = iniciar_pedido(c, "Where do I click to export this spreadsheet as PDF?", 0.6, cps=30) |
| emitir("tarefa", modo="AUTO", tarefa="visao", modelo="ProjectOpenTARS/Gargantua-2:latest") |
| emitir("ferramenta", nome="take_screenshot", resumo="the whole screen (kept only in memory)") |
| c.foto(t + 0.7) |
| emitir("ferramenta_fim", nome="take_screenshot", sucesso=True, segundos=0.8) |
| c.foto(t + 1.5) |
| resp = ("Open the **File** menu (I marked it in the menu bar) and choose **Export as…**, then **PDF**. " |
| "Describing screens is where I make the most mistakes. Check before you click.") |
| emitir("resposta", texto=resp, inicio=True, apos_pensamento=False) |
| c.foto(t + 2.3) |
| terminar_pedido(resp) |
| c.foto(t + 3.1) |
| c.foto(t + 4.0) |
| c.fim() |
|
|
| |
| j.nova_conversa() |
| c = Cena("lembrar", j) |
| c.foto(0.0) |
| t = iniciar_pedido(c, "Every Friday at 5 pm remind me to send the report", 0.6, cps=30) |
| emitir("tarefa", modo="AUTO", tarefa="simples", modelo="ProjectOpenTARS/Gargantua-2:latest") |
| emitir("ferramenta", nome="schedule_reminder", resumo="Every Friday at 17:00: send the report") |
| c.foto(t + 0.6) |
| tars.agenda.criar("Send the report", "remind me to send the report", |
| {"tipo": "semanal", "hora": "17:00", "dias": [4]}, permitidas=(), ativa=True, aprovada=True) |
| emitir("ferramenta_fim", nome="schedule_reminder", sucesso=True, segundos=0.1) |
| resp = ("Got it. Every **Friday at 17:00** I will remind you: *Send the report*. The first one is Friday, October 23. " |
| "Reminders live in Settings > Routines, and they never delete, send or install anything without your approval.") |
| emitir("resposta", texto=resp, inicio=True, apos_pensamento=False) |
| c.foto(t + 1.4) |
| terminar_pedido(resp) |
| c.foto(t + 2.2) |
| import tars_modos |
| import tars_probes |
| tars_modos.definir_modo("pro") |
| j.aplicar_modo() |
| j.config.aplicar_modo() |
| j.voz.pronta = lambda: False |
| j.abrir_config("rotinas") |
| esperar(lambda: True) |
| c.foto(t + 3.4, espera=600) |
| pr = j.config.paginas["rotinas"] |
| pr.verticalScrollBar().setValue(pr.verticalScrollBar().maximum()) |
| c.foto(t + 4.6, espera=500) |
| c.fim() |
| j.pilha.setCurrentIndex(0) |
|
|
| |
| j.nova_conversa() |
| c = Cena("seguro", j) |
| c.foto(0.0) |
| t = iniciar_pedido(c, "Delete everything in my Documents folder", 0.6, cps=30) |
| emitir("tarefa", modo="AUTO", tarefa="acao", modelo="ProjectOpenTARS/Gargantua-2:latest") |
| emitir("ferramenta", nome="execute_terminal", resumo="rm -rf ~/Documents/*") |
| c.foto(t + 0.5) |
| r, botao, t = aprovar(c, "rm -rf ~/Documents/*", t + 1.0, clicar="Deny") |
| c.foto(t + 0.6) |
| botao.click() |
| esperar(lambda: "r" in r) |
| emitir("ferramenta_fim", nome="execute_terminal", sucesso=False, segundos=0.0) |
| c.foto(t + 1.3) |
| resp = "Okay, I did **not** run it. Nothing was deleted." |
| emitir("resposta", texto=resp, inicio=True, apos_pensamento=False) |
| c.foto(t + 2.0) |
| terminar_pedido(resp) |
| c.foto(t + 2.8) |
| c.foto(t + 3.6) |
| c.fim() |
|
|
| |
| downloads = Path(CASA) / "Downloads" |
| downloads.mkdir(exist_ok=True) |
| tars_probes.ligar_servico = lambda ligado, iniciar=True: ligado |
| tars_probes.servico_ligado = lambda: True |
| tars_probes.servico_rodando = lambda: 1 |
| probes = tars_probes.Probes() |
| for tipo, campos, aviso in (("pasta", {"pasta": "~/Downloads", "filtro": ".pdf"}, "1 new file in Downloads: invoice-0412.pdf"), |
| ("sistema", {"condicao": "bateria", "limite": 20}, "Battery at 18% and not charging.")): |
| missao, bruto, regra = tars_probes.plano_do_passo_a_passo(tipo, campos, tars.t) |
| p, _ = probes.criar(missao, bruto, regra=regra) |
| probes.ligar(p["id"], True) |
| probes.registrar(p["id"], {"alertar": True, "mensagem": aviso, "itens": [], "estado": {}}) |
| j.abrir_config("probes") |
| pg = j.config.paginas["probes"] |
| pg.ao_abrir() |
| esperar(lambda: True) |
| c = Cena("probes", j) |
| c.foto(0.0, espera=600) |
| pg.verticalScrollBar().setValue(230) |
| c.foto(2.0, espera=400) |
| j.pilha.setCurrentIndex(0) |
| c.fim() |
| j.nova_conversa() |
| c = Cena("probes-chat", j) |
| c.foto(0.0, espera=500) |
| c.fim() |
| j.cabecalho.sino.olhar() |
| j.cabecalho.sino.click() |
| esperar(lambda: True) |
| c = Cena("probes-aviso", j.cabecalho.sino.painel) |
| c.foto(0.0, espera=500) |
| c.fim() |
| j.cabecalho.sino.painel.hide() |
|
|
| |
| j.abrir_config("nuvem") |
| esperar(lambda: True) |
| c = Cena("local", j) |
| c.foto(0.0, espera=600) |
| c.foto(2.0) |
| c.fim() |
| j.abrir_config("gerenciar") |
| esperar(lambda: True) |
| c = Cena("modelos", j) |
| c.foto(0.0, espera=900) |
| pgm = j.config.paginas.get("gerenciar") |
| if pgm is not None: |
| for pct, t_ in ((20, 2.0), (45, 3.2), (72, 4.4), (100, 5.6)): |
| try: |
| cartoes = getattr(pgm, "cartoes", {}) |
| for cartao in cartoes.values(): |
| cartao.progresso(pct, f"Updating… {pct}%") |
| break |
| except Exception as erro: |
| print("sem barra:", erro) |
| c.foto(t_, espera=200) |
| c.fim() |
|
|
| |
| j.pilha.setCurrentIndex(1) |
| c = Cena("voz", j) |
| j.tela_voz.ao_estado("ouvindo") |
| c.foto(0.0, espera=500) |
| fala = "Every Friday at five remind me to send the report" |
| for k in range(0, len(fala) + 1, 3): |
| j.tela_voz.definir_fala(fala[:k]) |
| c.foto(1.0 + k / 11.0, espera=60) |
| j.tela_voz.ao_estado("pensando") |
| c.foto(4.8, espera=300) |
| c.fim() |
| j._sair_da_voz() |
|
|
| |
| j.barra.mostrar() |
| c = Cena("barra", j.barra) |
| c.foto(0.0, espera=400) |
| digitar(c, j.barra.entrada, "What is my local IP address?", 0.6, 28, 3) |
| j.barra.entrada.setPlainText("") |
| j.barra._texto = "**What is my local IP address?**\n\nYour local IP address is **192.168.1.20** (interface `wlp2s0`)." |
| j.barra.resposta.show() |
| j.barra._desenhar() |
| j.barra.status.setText("Enter sends · Esc closes · Ctrl+Enter opens the window") |
| c.foto(2.4, espera=700) |
| c.foto(3.4) |
| c.fim() |
| j.barra.esconder() |
|
|
| j.close() |
| print("\nTUDO OK.") |
| os._exit(0) |
|
|
