Compare commits

..

3 Commits

Author SHA1 Message Date
Adolfo Reyna 1996a123d9 feat: add tactility controls and notes CRUD 2026-08-17 13:19:36 -04:00
Adolfo Reyna c445093186 feat: add reminder edit and delete commands 2026-08-17 11:18:55 -04:00
Adolfo Reyna bdf3b55a98 feat: restore Gitea CLI access 2026-08-16 08:40:47 -04:00
10 changed files with 823 additions and 18 deletions
+39 -13
View File
@@ -4,16 +4,15 @@ On-demand CLI façade for Reyna family MCP services and LAN devices. The goal is
## Phase 2 scope
- **Devices parent command**: All local devices are now under `reyna-cli devices`.
- **Devices parent command**: Non-Tactility local devices remain under `reyna-cli devices`; ESP32 boards use the dedicated `reyna-cli tactility` command group.
- `reyna-cli devices screen ...` (ESP32 screen)
- `reyna-cli devices laptop ...` (personal laptop MCP screen)
- `reyna-cli devices computer ...` (this computer's desktop companion client)
- `reyna-cli devices iphone ...` (iPhone app)
- `reyna-cli devices arm ...` (Robot arm)
- **Robot Arm**: Dynamic tool resolution for `state`, `home`, `wave`, `battery`.
- **Tactility boards**: mDNS-first discovery with reserved-IP fallback, plus direct web API access for sysinfo, apps, app installation/run, and filesystem operations.
- **Immich**: On-demand MCP bridge at `http://127.0.0.1:8626/mcp`.
- **MongoDB**: On-demand MCP bridge at `http://127.0.0.1:8630/mcp`.
- **Top-level aliases**: `reyna-cli screen`, `reyna-cli laptop`, `reyna-cli iphone`, `reyna-cli arm` still work for compatibility.
- **Top-level aliases**: `reyna-cli screen`, `reyna-cli laptop`, and `reyna-cli iphone` still work for compatibility.
## Install / run
@@ -22,6 +21,23 @@ uv sync
uv run reyna-cli doctor
```
## Signed Python launcher (macOS permissions)
The manually built, signed **Reyna CLI.app** owns the macOS privacy identity. Use the `signed` wrapper to launch the current Python CLI through that app:
```bash
# Optional: use a different active checkout or interpreter without rebuilding the app.
cat > ~/.reyna-cli.env <<'EOF'
REYNA_CLI_DIR=/Users/adolforeyna/Projects/platform/reyna-cli
REYNA_CLI_PYTHON=/Users/adolforeyna/Projects/platform/reyna-cli/.venv/bin/python
EOF
uv run reyna-cli signed doctor
uv run reyna-cli signed local-services speech transcribe-file ./sample.wav --json
```
The signed launcher reads `~/.reyna-cli.env`, then `~/.config/reyna-cli/env`; `REYNA_CLI_DIR` and `REYNA_CLI_PYTHON` environment variables override those values. Python-only edits take effect on the next `signed` run—**do not rebuild or re-sign the app for those edits**. Changes to Swift sources, `Info.plist`, or app signing require a new manual signed Xcode build before they can be used by `signed`.
## Device discovery
```bash
@@ -29,6 +45,25 @@ uv run reyna-cli devices list --json
uv run reyna-cli devices ping esp32_screen --json
uv run reyna-cli devices tools esp32_screen --json
uv run reyna-cli devices describe esp32_screen --for-hermes
# Tactility fleet discovery; offline boards remain visible in the JSON report
uv run reyna-cli tactility discover --json
uv run reyna-cli tactility sysinfo kidsos1 --json
uv run reyna-cli tactility apps kidsos1 --json
uv run reyna-cli tactility tools Grace --json
uv run reyna-cli tactility describe Grace --json
uv run reyna-cli tactility call Grace get_screenshot --args '{}' --json
uv run reyna-cli tactility call Grace draw_color_bmp --args '{"bmp_base64":"...","x":0,"y":0}' --json
uv run reyna-cli tactility call Grace play_tone --args '{"frequency":440,"duration_ms":500,"volume":40}' --json
uv run reyna-cli tactility call Grace get_sensors --args '{}' --json
uv run reyna-cli tactility install kidsos1 ./build/my.app --json
uv run reyna-cli tactility run kidsos1 one.tactility.myapp --json
uv run reyna-cli tactility report kidsos1 --json
uv run reyna-cli tactility screen clear kidsos1 --width 320 --height 240 --json
uv run reyna-cli tactility screen text kidsos1 "Grace\\nReady" --json
uv run reyna-cli tactility screen text kidsos1 "Layer intentionally" --no-clear-first --json
uv run reyna-cli tactility fs list kidsos1 --path /sdcard --json
uv run reyna-cli tactility fs upload kidsos1 ./manifest.json /sdcard/manifest.json --json
```
## Generic MCP calls
@@ -37,15 +72,6 @@ uv run reyna-cli devices describe esp32_screen --for-hermes
uv run reyna-cli devices call esp32_screen draw_text --args '{"text":"Hello","x":10,"y":20,"size":2}' --json
```
## Robot Arm
```bash
uv run reyna-cli devices arm state
uv run reyna-cli devices arm home
uv run reyna-cli devices arm wave
uv run reyna-cli devices arm battery --json
```
## ESP32 screen, laptop screen, this computer & iPhone
```bash
+133 -1
View File
@@ -22,10 +22,11 @@ from reyna_cli.apple_llm_client import (
from reyna_cli.deco_direct import DecoDirectClient
from reyna_cli.desktop_service import SERVICE_NAME, service_action, service_status, unit_content, unit_path
from reyna_cli.email_local import ThunderbirdEmailClient, email_tool_specs
from reyna_cli.gitea_direct import GiteaClient, load_credentials
from reyna_cli.immich import ImmichClient
from reyna_cli.mcp import MCPClient
from reyna_cli.mongo_direct import MongoDirectClient
from reyna_cli.notes_direct import NotesAutomationError, create_note, list_notes, read_note
from reyna_cli.notes_direct import NotesAutomationError, create_note, delete_note, list_notes, read_note, update_note
from reyna_cli.privacy_client import default_socket_path as privacy_default_socket_path
from reyna_cli.privacy_host import native_calendar_list, privacy_host_status_payload
from reyna_cli.remarkable import LISTENER_LABEL, listen_forever, listener_service_action, listener_service_status, sync_once
@@ -51,6 +52,7 @@ immich_app = typer.Typer(help="Immich direct REST API commands.")
mongo_app = typer.Typer(help="MongoDB direct driver commands.")
zoom_app = typer.Typer(help="Zoom direct REST API commands.")
email_app = typer.Typer(help="Read-only local Thunderbird email commands.")
gitea_app = typer.Typer(help="Direct Gitea API and credential diagnostics (never prints tokens).")
deco_app = typer.Typer(help="TP-Link Deco direct router commands.")
macmini_app = typer.Typer(help="Mac mini MCP tools (Calendar, Contacts, Reminders, Deco).")
remarkable_app = typer.Typer(help="Paper Pro discovery, local cache, and macOS listener service.")
@@ -79,6 +81,7 @@ app.add_typer(immich_app, name="immich")
app.add_typer(mongo_app, name="mongo")
app.add_typer(zoom_app, name="zoom")
app.add_typer(email_app, name="email")
app.add_typer(gitea_app, name="gitea")
app.add_typer(deco_app, name="deco")
macmini_app.add_typer(macmini_calendar_app, name="calendar")
macmini_app.add_typer(macmini_contacts_app, name="contacts")
@@ -142,6 +145,35 @@ def notes_create(
fail(str(exc), json_output)
@notes_app.command("edit")
def notes_edit(
note_id: str,
title: Optional[str] = typer.Option(None, "--title"),
body: Optional[str] = typer.Option(None, "--body"),
json_output: bool = typer.Option(False, "--json"),
):
"""Replace the title and/or body of one Apple Note by ID."""
try:
emit({"ok": True, "source": "direct_apple_notes", "note": update_note(note_id, title=title, body=body)}, json_output)
except (NotesAutomationError, ValueError) as exc:
fail(str(exc), json_output)
@notes_app.command("delete")
def notes_delete(
note_id: str,
force: bool = typer.Option(False, "--force", help="Required: permanently delete the note."),
json_output: bool = typer.Option(False, "--json"),
):
"""Permanently delete one Apple Note by ID; requires explicit --force."""
if not force:
fail("Deletion requires --force", json_output)
try:
emit({"ok": True, "source": "direct_apple_notes", "note": delete_note(note_id)}, json_output)
except (NotesAutomationError, ValueError) as exc:
fail(str(exc), json_output)
def scrub_sensitive(value: Any) -> Any:
if isinstance(value, dict):
cleaned: Dict[str, Any] = {}
@@ -170,6 +202,56 @@ def fail(message: str, json_output: bool = False, **extra: Any) -> None:
raise typer.Exit(1)
@gitea_app.command("credential")
def gitea_credential(json_output: bool = typer.Option(False, "--json")):
"""Report whether a Gitea credential is configured, without displaying it."""
emit({"ok": True, **load_credentials().public_dict()}, json_output)
@gitea_app.command("access")
def gitea_access(json_output: bool = typer.Option(False, "--json")):
"""Read the authenticated Gitea user endpoint."""
try:
emit({"ok": True, "user": GiteaClient().access()}, json_output)
except Exception as exc:
fail(str(exc), json_output)
@gitea_app.command("repos")
def gitea_repos(
limit: int = typer.Option(50, min=1, max=100),
json_output: bool = typer.Option(False, "--json"),
):
"""List repositories when the configured token has repository-list scope."""
try:
emit({"ok": True, "repos": GiteaClient().repos(limit)}, json_output)
except Exception as exc:
fail(str(exc), json_output)
@gitea_app.command("repo")
def gitea_repo(name: str, json_output: bool = typer.Option(False, "--json")):
"""Read one repository and its permissions, e.g. adolforeyna/reyna-cli."""
try:
emit({"ok": True, "repo": GiteaClient().repo(name)}, json_output)
except Exception as exc:
fail(str(exc), json_output)
def run_remindctl(args: List[str]) -> Any:
"""Execute the supported Reminders CLI without invoking a shell."""
result = subprocess.run(["remindctl", *args], capture_output=True, text=True, check=False)
if result.returncode:
raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "remindctl command failed")
output = result.stdout.strip()
if not output:
return {}
try:
return json.loads(output)
except json.JSONDecodeError:
return {"output": output}
def get_required_device(name: str, json_output: bool = False) -> Device:
device = get_device(name)
if not device:
@@ -2100,6 +2182,56 @@ def macmini_reminders_create(
fail(str(exc), json_output)
@macmini_reminders_app.command("edit")
def macmini_reminders_edit(
reminder_id: str,
title: Optional[str] = typer.Option(None, "--title"),
list_name: Optional[str] = typer.Option(None, "--list"),
due: Optional[str] = typer.Option(None, "--due"),
notes: Optional[str] = typer.Option(None, "--notes"),
priority: Optional[str] = typer.Option(None, "--priority"),
clear_due: bool = typer.Option(False, "--clear-due"),
complete: bool = typer.Option(False, "--complete"),
incomplete: bool = typer.Option(False, "--incomplete"),
json_output: bool = typer.Option(False, "--json"),
):
"""Edit a reminder by its ID or ID prefix returned by ``reminders list``."""
if complete and incomplete:
fail("Use only one of --complete or --incomplete", json_output)
if not any((title is not None, list_name is not None, due is not None, notes is not None, priority is not None, clear_due, complete, incomplete)):
fail("Specify at least one field to edit", json_output)
args = ["edit", reminder_id]
for flag, value in (("--title", title), ("--list", list_name), ("--due", due), ("--notes", notes), ("--priority", priority)):
if value is not None:
args.extend([flag, value])
if clear_due:
args.append("--clear-due")
if complete:
args.append("--complete")
if incomplete:
args.append("--incomplete")
args.extend(["--json", "--no-input"])
try:
emit({"ok": True, "source": "remindctl", "result": run_remindctl(args)}, json_output)
except Exception as exc:
fail(str(exc), json_output)
@macmini_reminders_app.command("delete")
def macmini_reminders_delete(
reminder_ids: List[str] = typer.Argument(..., help="Reminder IDs or ID prefixes returned by reminders list."),
force: bool = typer.Option(False, "--force", help="Required: permanently delete without an interactive prompt."),
json_output: bool = typer.Option(False, "--json"),
):
"""Permanently delete one or more reminders; requires explicit --force."""
if not force:
fail("Deletion requires --force", json_output)
try:
emit({"ok": True, "source": "remindctl", "result": run_remindctl(["delete", *reminder_ids, "--force", "--json", "--no-input"])}, json_output)
except Exception as exc:
fail(str(exc), json_output)
def emit_deco_result(tool: str, result: Dict[str, Any], json_output: bool = False) -> None:
emit({"ok": True, "source": "direct", "tool": tool, "result": result}, json_output)
+27 -3
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import os
import socket
from pathlib import Path
from typing import Any, Dict, List, Optional
@@ -36,6 +37,9 @@ class Device(BaseModel):
urls.append(url)
return urls
def resolved_host(self) -> str:
return resolve_device_host(self)
def public_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
@@ -108,13 +112,12 @@ def load_registry(path: Optional[Path] = None) -> Registry:
def get_device(name: str, registry: Optional[Registry] = None) -> Optional[Device]:
registry = registry or load_registry()
normalized = name.strip()
normalized = name.strip().casefold()
shorthand = {
"screen": "esp32_screen",
"esp32": "esp32_screen",
"iphone": "iphone_mcp",
"phone": "iphone_mcp",
"arm": "robot_arm",
"laptop": "personal_laptop_screen",
"laptop_screen": "personal_laptop_screen",
"personal_laptop": "personal_laptop_screen",
@@ -123,11 +126,32 @@ def get_device(name: str, registry: Optional[Registry] = None) -> Optional[Devic
"local": "local_desktop_screen",
}
normalized = shorthand.get(normalized, normalized)
aliases = {str(key).casefold(): value for key, value in registry.aliases.items()}
normalized = aliases.get(normalized, normalized)
for device in registry.devices:
if normalized in {device.id, device.host, device.display_name}:
candidates = {device.id.casefold(), device.host.casefold(), device.display_name.casefold()}
if normalized in candidates:
return device
display = device.display_name.casefold()
if display.startswith(normalized + " ") or display.startswith(normalized + "'s"):
return device
return None
def resolve_device_host(device: Device) -> str:
"""Resolve mDNS when available, retaining the registry IP for offline boards."""
if device.host:
try:
return socket.gethostbyname(device.host)
except OSError:
pass
return device.reserved_ip or device.host
def is_tactility_device(device: Device) -> bool:
value = f"{device.id} {device.type} {' '.join(device.capabilities)}".lower()
return "tactility" in value or device.id.startswith("kidsos_") or device.id == "esp32_screen"
def cache_path_for(device_id: str) -> Path:
return CACHE_DIR / "devices" / f"{device_id}-tools.json"
+174
View File
@@ -0,0 +1,174 @@
from __future__ import annotations
import base64
import json
import os
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib.parse import unquote, urlsplit
import httpx
from reyna_cli.env import load_hermes_env
DEFAULT_GITEA_URL = "https://git.reynafamily.com"
SENSITIVE_KEYS = {"token", "access_token", "password", "authorization", "refresh_token"}
@dataclass(frozen=True)
class GiteaCredentials:
username: str
token: str
source: str
base_url: str
def public_dict(self) -> dict[str, Any]:
parsed = urlsplit(self.base_url)
return {
"configured": bool(self.token) or self.source == "Pi 5 git credential helper",
"host": parsed.netloc,
"source": self.source,
}
def gitea_url() -> str:
load_hermes_env()
return os.getenv("REYNA_GITEA_URL") or os.getenv("GITEA_URL") or DEFAULT_GITEA_URL
def _credential_from_file(path: Path, host: str) -> tuple[str, str] | None:
try:
lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
except OSError:
return None
for line in lines:
parsed = urlsplit(line.strip())
if parsed.hostname == host and parsed.username and parsed.password:
return unquote(parsed.username), unquote(parsed.password)
return None
def _credential_from_git_helper(host: str) -> tuple[str, str] | None:
try:
result = subprocess.run(
["git", "credential", "fill"],
input=f"protocol=https\nhost={host}\n\n",
text=True,
capture_output=True,
timeout=10,
check=False,
)
except (OSError, subprocess.TimeoutExpired):
return None
if result.returncode:
return None
values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)
username, password = values.get("username", ""), values.get("password", "")
return (username, password) if username and password else None
def load_credentials() -> GiteaCredentials:
load_hermes_env()
base_url = gitea_url().rstrip("/")
host = urlsplit(base_url).hostname or ""
token = os.getenv("REYNA_GITEA_TOKEN") or os.getenv("GITEA_TOKEN")
if token:
return GiteaCredentials(os.getenv("REYNA_GITEA_USERNAME", "adolforeyna"), token, "REYNA_GITEA_TOKEN", base_url)
credentials_path = Path(os.getenv("GIT_CREDENTIALS_FILE", Path.home() / ".git-credentials"))
found = _credential_from_file(credentials_path, host)
if found:
return GiteaCredentials(*found, "GIT_CREDENTIALS_FILE", base_url)
found = _credential_from_git_helper(host)
if found:
return GiteaCredentials(*found, "git credential helper", base_url)
return GiteaCredentials("", "", "Pi 5 git credential helper", base_url)
def redact_sensitive(value: Any) -> Any:
if isinstance(value, dict):
return {key: "[REDACTED]" if key.casefold() in SENSITIVE_KEYS else redact_sensitive(item) for key, item in value.items()}
if isinstance(value, list):
return [redact_sensitive(item) for item in value]
return value
class GiteaClient:
def __init__(self, credentials: GiteaCredentials | None = None, timeout: float = 20.0, transport: httpx.BaseTransport | None = None):
self.credentials = credentials or load_credentials()
self.remote_credential_helper = self.credentials.source == "Pi 5 git credential helper"
if not self.credentials.token and not self.remote_credential_helper:
raise RuntimeError("No Gitea credential is available. Set REYNA_GITEA_TOKEN or configure a supported Git credential source.")
self.client = httpx.Client(
base_url=self.credentials.base_url,
timeout=timeout,
transport=transport,
headers={"Authorization": f"token {self.credentials.token}", "Accept": "application/json"},
)
def _remote_get(self, path: str) -> Any:
"""Use Pi 5's existing Git credential helper without returning its token."""
script = r'''import json, subprocess, sys, urllib.request
from urllib.parse import urlsplit
base_url, path = sys.argv[1:]
host = urlsplit(base_url).hostname
result = subprocess.run(["git", "credential", "fill"], input=f"protocol=https\nhost={host}\n\n", text=True, capture_output=True, check=False)
values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)
token = values.get("password", "")
if result.returncode or not token:
raise SystemExit("Pi 5 Git credential helper did not provide a Gitea credential")
request = urllib.request.Request(base_url.rstrip("/") + path, headers={"Authorization": "token " + token, "Accept": "application/json"})
with urllib.request.urlopen(request, timeout=20) as response:
payload = json.load(response)
def redact(value):
if isinstance(value, dict):
return {key: "[REDACTED]" if key.casefold() in {"token", "access_token", "password", "authorization", "refresh_token"} else redact(item) for key, item in value.items()}
if isinstance(value, list):
return [redact(item) for item in value]
return value
print(json.dumps(redact(payload)))'''
helper_host = os.getenv("REYNA_GITEA_CREDENTIAL_HOST", "pi5")
encoded_script = base64.b64encode(script.encode("utf-8")).decode("ascii")
remote_command = (
"python3 -c \"import base64;exec(base64.b64decode('"
+ encoded_script
+ "'))\" "
+ self.credentials.base_url
+ " "
+ path
)
result = subprocess.run(
["ssh", "-o", "BatchMode=yes", helper_host, remote_command],
text=True,
capture_output=True,
timeout=30,
check=False,
)
if result.returncode:
raise RuntimeError("Pi 5 Gitea credential helper request failed.")
try:
return json.loads(result.stdout)
except json.JSONDecodeError as exc:
raise RuntimeError("Pi 5 Gitea credential helper returned invalid JSON.") from exc
def _get(self, path: str) -> Any:
if self.remote_credential_helper:
return self._remote_get(path)
response = self.client.get(path)
response.raise_for_status()
return redact_sensitive(response.json())
def access(self) -> Any:
return self._get("/api/v1/user")
def repos(self, limit: int = 50) -> Any:
return self._get("/api/v1/user/repos?limit=" + str(limit))
def repo(self, name: str) -> Any:
owner, separator, repo = name.partition("/")
if not separator or not owner or not repo:
raise ValueError("Repository must be in owner/name form.")
return self._get(f"/api/v1/repos/{owner}/{repo}")
+74
View File
@@ -105,6 +105,53 @@ _NOTES_CREATE_SCRIPT = r'''function run(argv) {
}'''
_NOTES_UPDATE_SCRIPT = r'''function run(argv) {
const input = JSON.parse(argv[0]);
const app = Application("/System/Applications/Notes.app");
const escapeHtml = value => String(value).replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;").replace(/"/g, "&quot;");
const accounts = app.accounts();
for (let a = 0; a < accounts.length; a++) {
const folders = accounts[a].folders();
for (let f = 0; f < folders.length; f++) {
const notes = folders[f].notes();
for (let n = 0; n < notes.length; n++) {
const note = notes[n];
if (String(note.id()) !== input.id) continue;
const oldTitle = String(note.name());
const oldText = String(note.plaintext());
const fallbackBody = oldText.startsWith(oldTitle) ? oldText.slice(oldTitle.length).replace(/^\\n+/, "") : oldText;
const title = input.title === null ? oldTitle : input.title;
const body = input.body === null ? fallbackBody : input.body;
note.body = "<h1>" + escapeHtml(title) + "</h1><div>" + escapeHtml(body).replace(/\\n/g, "<br>") + "</div>";
return JSON.stringify({id: String(note.id()), title: String(note.name()), folder: String(folders[f].name())});
}
}
}
throw new Error("Note not found");
}'''
_NOTES_DELETE_SCRIPT = r'''function run(argv) {
const input = JSON.parse(argv[0]);
const app = Application("/System/Applications/Notes.app");
const accounts = app.accounts();
for (let a = 0; a < accounts.length; a++) {
const folders = accounts[a].folders();
for (let f = 0; f < folders.length; f++) {
const notes = folders[f].notes();
for (let n = 0; n < notes.length; n++) {
const note = notes[n];
if (String(note.id()) !== input.id) continue;
const id = String(note.id());
note.delete();
return JSON.stringify({id: id, deleted: true});
}
}
}
throw new Error("Note not found");
}'''
def _run_jxa(script: str, payload: dict[str, Any], *, runner: Runner = subprocess.run) -> Any:
arguments: Sequence[str] = [
"/usr/bin/osascript",
@@ -178,3 +225,30 @@ def create_note(
if not isinstance(result, dict):
raise NotesAutomationError("Apple Notes create response was not an object")
return result
def update_note(
note_id: str,
*,
title: Optional[str] = None,
body: Optional[str] = None,
runner: Runner = subprocess.run,
) -> dict[str, Any]:
note_id = _bounded_text(note_id, "note id", 1024)
if title is None and body is None:
raise ValueError("Specify title or body to update")
if title is not None:
title = _bounded_text(title, "title", 500)
if body is not None and len(body) > 100_000:
raise ValueError("body exceeds maximum length 100000")
result = _run_jxa(_NOTES_UPDATE_SCRIPT, {"id": note_id, "title": title, "body": body}, runner=runner)
if not isinstance(result, dict):
raise NotesAutomationError("Apple Notes update response was not an object")
return result
def delete_note(note_id: str, *, runner: Runner = subprocess.run) -> dict[str, Any]:
result = _run_jxa(_NOTES_DELETE_SCRIPT, {"id": _bounded_text(note_id, "note id", 1024)}, runner=runner)
if not isinstance(result, dict) or result.get("deleted") is not True:
raise NotesAutomationError("Apple Notes delete response was not confirmed")
return result
+123
View File
@@ -0,0 +1,123 @@
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, Optional
from urllib.parse import quote
import httpx
from PIL import Image, ImageDraw, ImageFont
class TactilityClient:
"""Small direct client for the Tactility firmware HTTP API."""
def __init__(self, base_url: str, timeout: float = 30.0, transport: Optional[httpx.BaseTransport] = None):
self.base_url = base_url.rstrip("/")
self.client = httpx.Client(timeout=timeout, transport=transport, follow_redirects=True)
def _response(self, response: httpx.Response) -> Any:
response.raise_for_status()
if not response.content:
return {"ok": True, "status_code": response.status_code}
content_type = response.headers.get("content-type", "")
if "json" in content_type:
return response.json()
try:
return response.json()
except ValueError:
return {"ok": True, "status_code": response.status_code, "response": response.text}
def get(self, path: str, **kwargs: Any) -> Any:
return self._response(self.client.get(f"{self.base_url}{path}", **kwargs))
def post(self, path: str, **kwargs: Any) -> Any:
return self._response(self.client.post(f"{self.base_url}{path}", **kwargs))
def sysinfo(self) -> Any:
return self.get("/api/sysinfo")
def apps(self) -> Any:
return self.get("/api/apps")
def install_app(self, app_path: Path) -> Any:
with app_path.open("rb") as handle:
response = self.client.put(
f"{self.base_url}/api/apps/install",
files={"file": (app_path.name, handle, "application/octet-stream")},
)
return self._response(response)
def run_app(self, app_id: str) -> Any:
return self.post("/api/apps/run", params={"id": app_id})
def fs_list(self, path: str = "/") -> Any:
return self.get("/fs/list", params={"path": path})
def fs_mkdir(self, path: str) -> Any:
return self.post("/fs/mkdir", params={"path": path})
def fs_upload(self, local_path: Path, remote_path: str) -> Any:
data = local_path.read_bytes()
return self.post(
"/fs/upload",
params={"path": remote_path},
content=data,
headers={"Content-Type": "application/octet-stream", "Content-Length": str(len(data))},
)
def fs_download(self, remote_path: str, local_path: Path) -> Dict[str, Any]:
response = self.client.get(f"{self.base_url}/fs/download", params={"path": remote_path})
response.raise_for_status()
local_path.parent.mkdir(parents=True, exist_ok=True)
local_path.write_bytes(response.content)
return {"ok": True, "path": str(local_path), "bytes": len(response.content)}
def fs_read(self, remote_path: str) -> str:
response = self.client.get(f"{self.base_url}/fs/download", params={"path": remote_path})
response.raise_for_status()
return response.content.decode("utf-8", errors="replace")
def fs_delete(self, path: str) -> Any:
return self.post("/fs/delete", params={"path": path})
def fs_rename(self, path: str, new_name: str) -> Any:
return self.post("/fs/rename", params={"path": path, "newName": new_name})
def screen_raw(self, pixels: bytes, width: int, height: int) -> Any:
expected = width * height * 2
if len(pixels) != expected:
raise ValueError(f"RGB565 frame is {len(pixels)} bytes; expected {expected} for {width}x{height}")
response = self.client.post(
f"{self.base_url}/api/screen/raw",
params={"w": width, "h": height},
content=pixels,
headers={"Content-Type": "application/octet-stream", "Content-Length": str(len(pixels))},
)
return self._response(response)
def screen_clear_raw(self, width: int, height: int) -> Any:
return self.screen_raw(bytes(width * height * 2), width, height)
@staticmethod
def _rgb565(image: Image.Image) -> bytes:
pixels = bytearray()
for red, green, blue in image.convert("RGB").get_flattened_data():
pixels.extend((((red & 0xF8) << 8) | ((green & 0xFC) << 3) | (blue >> 3)).to_bytes(2, "big"))
return bytes(pixels)
def screen_text(self, message: str, width: int, height: int, clear_first: bool = True) -> Dict[str, Any]:
clear_result = self.screen_clear_raw(width, height) if clear_first else None
image = Image.new("RGB", (width, height), (14, 18, 38))
draw = ImageDraw.Draw(image)
font_path = next((path for path in (
"/System/Library/Fonts/SFNS.ttf",
"/Library/Fonts/Arial.ttf",
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
) if Path(path).exists()), None)
font_size = max(16, round(width / 16))
font = ImageFont.truetype(font_path, font_size) if font_path else ImageFont.load_default()
margin = max(1, min(8, min(width, height) // 10))
draw.rectangle((margin, margin, width - margin - 1, height - margin - 1), outline=(90, 110, 180), width=2)
draw.multiline_text((width // 2, height // 2), message, font=font, fill=(245, 245, 232), anchor="mm", align="center", spacing=5)
write_result = self.screen_raw(self._rgb565(image), width, height)
return {"clear_first": clear_first, "clear": clear_result, "write": write_result, "width": width, "height": height}
+84
View File
@@ -0,0 +1,84 @@
from __future__ import annotations
from pathlib import Path
import httpx
from reyna_cli.gitea_direct import GiteaClient, GiteaCredentials, load_credentials, redact_sensitive
def test_load_credentials_prefers_environment_without_exposing_token(monkeypatch):
monkeypatch.setenv("REYNA_GITEA_TOKEN", "test-token-value")
monkeypatch.setenv("REYNA_GITEA_URL", "https://git.example.test/")
credentials = load_credentials()
assert credentials.token == "test-token-value"
assert credentials.source == "REYNA_GITEA_TOKEN"
assert credentials.public_dict() == {
"configured": True,
"host": "git.example.test",
"source": "REYNA_GITEA_TOKEN",
}
assert "token" not in credentials.public_dict()
def test_load_credentials_reads_git_credentials_file(monkeypatch, tmp_path: Path):
credentials_file = tmp_path / "credentials"
credentials_file.write_text(
"https://adolforeyna:stored-secret@git.reynafamily.com\n",
encoding="utf-8",
)
monkeypatch.delenv("REYNA_GITEA_TOKEN", raising=False)
monkeypatch.delenv("GITEA_TOKEN", raising=False)
monkeypatch.setenv("GIT_CREDENTIALS_FILE", str(credentials_file))
credentials = load_credentials()
assert credentials.username == "adolforeyna"
assert credentials.token == "stored-secret"
assert credentials.source == "GIT_CREDENTIALS_FILE"
assert "stored-secret" not in str(credentials.public_dict())
def test_repo_response_is_redacted_and_requests_authenticated_endpoint():
requests: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
return httpx.Response(
200,
json={
"full_name": "adolforeyna/reyna-cli",
"permissions": {"push": True},
"token": "response-secret",
},
)
client = GiteaClient(
GiteaCredentials("adolforeyna", "request-secret", "test", "https://git.example.test"),
transport=httpx.MockTransport(handler),
)
result = client.repo("adolforeyna/reyna-cli")
assert requests[0].url.path == "/api/v1/repos/adolforeyna/reyna-cli"
assert requests[0].headers["authorization"] == "token request-secret"
assert result["permissions"]["push"] is True
assert result["token"] == "[REDACTED]"
assert "response-secret" not in str(result)
def test_remote_credential_helper_can_make_a_safe_api_request(monkeypatch):
credentials = GiteaCredentials("", "", "Pi 5 git credential helper", "https://git.example.test")
assert credentials.public_dict()["configured"] is True
client = GiteaClient(credentials)
monkeypatch.setattr(client, "_remote_get", lambda path: {"full_name": "adolforeyna/reyna-cli"})
assert client.repo("adolforeyna/reyna-cli") == {"full_name": "adolforeyna/reyna-cli"}
def test_redact_sensitive_handles_nested_api_values():
assert redact_sensitive(
{"password": "one", "nested": [{"access_token": "two"}], "name": "safe"}
) == {"password": "[REDACTED]", "nested": [{"access_token": "[REDACTED]"}], "name": "safe"}
+20 -1
View File
@@ -12,7 +12,7 @@ from types import SimpleNamespace
from typer.testing import CliRunner
from reyna_cli.cli import app
from reyna_cli.notes_direct import create_note, list_notes, read_note
from reyna_cli.notes_direct import create_note, delete_note, list_notes, read_note, update_note
runner = CliRunner()
@@ -63,6 +63,25 @@ def test_read_and_create_notes_use_json_arguments_without_script_interpolation()
assert created["id"] == "new-1"
def test_update_and_delete_notes_use_fixed_jxa_and_json_arguments():
updated = update_note(
"x-coredata://note-1",
title="Updated",
body="New body",
runner=_success_runner(
{"id": "x-coredata://note-1", "title": "Updated", "body": "New body"},
{"id": "x-coredata://note-1", "title": "Updated"},
),
)
assert updated["title"] == "Updated"
deleted = delete_note(
"x-coredata://note-1",
runner=_success_runner({"id": "x-coredata://note-1"}, {"id": "x-coredata://note-1", "deleted": True}),
)
assert deleted["deleted"] is True
def test_notes_cli_is_available_at_top_level_and_macmini_alias(monkeypatch):
monkeypatch.setattr("reyna_cli.cli.list_notes", lambda **_: [{"id": "n1", "title": "Test"}])
+36
View File
@@ -248,6 +248,42 @@ def test_cli_reminders_no_mcp_tool_call_remaining():
assert 'call_macmini_tool("reminders_create"' not in src
def test_cli_reminders_edit_uses_remindctl(monkeypatch):
captured = {}
def fake_run(args):
captured["args"] = args
return {"updated": {"id": "r1", "title": "New title"}}
monkeypatch.setattr("reyna_cli.cli.run_remindctl", fake_run)
res = runner.invoke(app, ["macmini", "reminders", "edit", "r1", "--title", "New title", "--due", "tomorrow", "--json"])
assert res.exit_code == 0
assert captured["args"] == ["edit", "r1", "--title", "New title", "--due", "tomorrow", "--json", "--no-input"]
assert json.loads(res.stdout)["source"] == "remindctl"
def test_cli_reminders_delete_requires_force(monkeypatch):
res = runner.invoke(app, ["macmini", "reminders", "delete", "r1", "--json"])
assert res.exit_code == 1
assert "--force" in res.stdout
def test_cli_reminders_delete_uses_remindctl_with_force(monkeypatch):
captured = {}
def fake_run(args):
captured["args"] = args
return {"deleted": ["r1"]}
monkeypatch.setattr("reyna_cli.cli.run_remindctl", fake_run)
res = runner.invoke(app, ["macmini", "reminders", "delete", "r1", "--force", "--json"])
assert res.exit_code == 0
assert captured["args"] == ["delete", "r1", "--force", "--json", "--no-input"]
def test_privacy_host_cli_has_reminders_authorize():
res = runner.invoke(app, ["privacy-host", "--help"])
assert res.exit_code == 0
+113
View File
@@ -0,0 +1,113 @@
import json
from pathlib import Path
import httpx
from typer.testing import CliRunner
from reyna_cli.cli import app
from reyna_cli.config import Device, Registry, get_device, resolve_device_host
from reyna_cli.tactility import TactilityClient
runner = CliRunner()
def test_friendly_name_lookup_is_case_insensitive():
registry = Registry(devices=[Device(id="kidsos_5c5c", display_name="Grace's 2.8-inch Tactility Board")])
assert get_device("Grace", registry).id == "kidsos_5c5c"
assert get_device("grace", registry).id == "kidsos_5c5c"
def test_resolve_device_host_prefers_live_mdns(monkeypatch):
monkeypatch.setattr("reyna_cli.config.socket.gethostbyname", lambda host: "192.168.68.141")
device = Device(id="kidsos_1234", host="kidsos-1234.local", reserved_ip="192.168.68.99")
assert resolve_device_host(device) == "192.168.68.141"
def test_resolve_device_host_falls_back_when_mdns_is_offline(monkeypatch):
def fail(_host):
raise OSError("offline")
monkeypatch.setattr("reyna_cli.config.socket.gethostbyname", fail)
device = Device(id="kidsos_1234", host="kidsos-1234.local", reserved_ip="192.168.68.99")
assert resolve_device_host(device) == "192.168.68.99"
def test_tactility_client_uses_web_api(monkeypatch, tmp_path):
requests = []
def handler(request: httpx.Request):
requests.append(request)
if request.url.path == "/api/sysinfo":
return httpx.Response(200, json={"version": "0.8.0-dev"})
if request.url.path == "/api/apps":
return httpx.Response(200, json={"apps": [{"id": "one.tactility.demo"}]})
if request.url.path == "/fs/list":
return httpx.Response(200, json={"path": "/sdcard", "entries": []})
return httpx.Response(200, json={"ok": True})
client = TactilityClient("http://kidsos-1234.local", transport=httpx.MockTransport(handler))
assert client.sysinfo()["version"] == "0.8.0-dev"
assert client.apps()["apps"][0]["id"] == "one.tactility.demo"
assert client.fs_list("/sdcard")["path"] == "/sdcard"
assert requests[0].url.host == "kidsos-1234.local"
def test_tactility_install_and_run_requests(tmp_path):
requests = []
def handler(request: httpx.Request):
requests.append(request)
return httpx.Response(200, json={"ok": True})
app_file = tmp_path / "demo.app"
app_file.write_bytes(b"APP")
client = TactilityClient("http://192.168.68.99", transport=httpx.MockTransport(handler))
client.install_app(app_file)
client.run_app("one.tactility.demo")
assert requests[0].method == "PUT"
assert requests[0].url.path == "/api/apps/install"
assert requests[1].url.path == "/api/apps/run"
assert requests[1].url.params["id"] == "one.tactility.demo"
def test_tactility_screen_text_clears_before_write():
requests = []
def handler(request: httpx.Request):
requests.append(request)
return httpx.Response(200, json={"ok": True})
client = TactilityClient("http://192.168.68.99", transport=httpx.MockTransport(handler))
result = client.screen_text("Grace", 20, 10, clear_first=True)
assert result["clear_first"] is True
assert [request.url.path for request in requests] == ["/api/screen/raw", "/api/screen/raw"]
assert requests[0].content == bytes(20 * 10 * 2)
assert len(requests[1].content) == 20 * 10 * 2
def test_tactility_cli_has_commands(monkeypatch, tmp_path):
registry_path = tmp_path / "devices.yaml"
registry_path.write_text(
"""
devices:
kidsos_1234:
type: esp32-s3-tactility
host: kidsos-1234.local
reserved_ip: 192.168.68.99
""",
encoding="utf-8",
)
monkeypatch.setenv("REYNA_DEVICES_REGISTRY", str(registry_path))
result = runner.invoke(app, ["tactility", "--help"])
assert result.exit_code == 0
for command in ("discover", "sysinfo", "apps", "install", "run", "report", "fs", "screen"):
assert command in result.stdout
def test_robot_arm_is_not_a_cli_surface():
result = runner.invoke(app, ["devices", "--help"])
assert result.exit_code == 0
assert "arm" not in result.stdout.lower()