8712931c23
- frontend/app.js: use wss:// when page is served over HTTPS; add HTTP polling
fallback (/state every 2s) when WebSocket is down, amber dot = polling mode;
stop polling on WS reconnect; clearer state labels (LOG OFFLINE, LOAD ERROR,
DISCONNECTED); fix replace('_',' ') to use regex for multi-underscore states
- frontend/style.css: add #ws-dot.polling (amber); min-height 280px on screen;
shrink llama-wrap/SVG; tighten screen padding and gap
- log_parser.py: detect cancelled requests (stop: cancel task / Connection
handling canceled) → idle; fix: switch to generating at progress=1.00
- server.py: add GET /state endpoint for polling fallback
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
import argparse
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import uvicorn
|
|
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from log_parser import LogParser, ServerState
|
|
from log_watcher import LogWatcher
|
|
|
|
FRONTEND_DIR = Path(__file__).parent / "frontend"
|
|
|
|
# ── App factory ───────────────────────────────────────────────────────────────
|
|
|
|
def create_app(log_path: str, reopen_log: bool = False) -> FastAPI:
|
|
app = FastAPI()
|
|
app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")
|
|
|
|
clients: set[WebSocket] = set()
|
|
parser = LogParser()
|
|
last_state: dict = parser.state.to_dict()
|
|
|
|
@app.get("/")
|
|
async def index():
|
|
return FileResponse(FRONTEND_DIR / "index.html")
|
|
|
|
@app.get("/state")
|
|
async def get_state():
|
|
return last_state
|
|
|
|
@app.websocket("/ws")
|
|
async def ws_endpoint(ws: WebSocket):
|
|
await ws.accept()
|
|
clients.add(ws)
|
|
await ws.send_text(json.dumps(last_state))
|
|
try:
|
|
while True:
|
|
await ws.receive_text()
|
|
except WebSocketDisconnect:
|
|
clients.discard(ws)
|
|
|
|
async def broadcast(state: ServerState):
|
|
nonlocal last_state, clients
|
|
last_state = state.to_dict()
|
|
payload = json.dumps(last_state)
|
|
dead: set[WebSocket] = set()
|
|
for ws in list(clients):
|
|
try:
|
|
await ws.send_text(payload)
|
|
except Exception:
|
|
dead.add(ws)
|
|
clients -= dead
|
|
|
|
async def on_line(line: str):
|
|
result = parser.feed(line)
|
|
if result is not None:
|
|
await broadcast(result)
|
|
|
|
async def on_offline():
|
|
parser._state.state = "offline"
|
|
await broadcast(parser.state)
|
|
|
|
@app.on_event("startup")
|
|
async def startup():
|
|
watcher = LogWatcher(log_path, on_line, on_offline=on_offline, reopen=reopen_log)
|
|
watcher.start()
|
|
|
|
return app
|
|
|
|
|
|
# ── CLI entry point ───────────────────────────────────────────────────────────
|
|
|
|
def _config_bool(key: str) -> bool:
|
|
config_file = Path("config.json")
|
|
if config_file.exists():
|
|
return bool(json.loads(config_file.read_text()).get(key, False))
|
|
return False
|
|
|
|
|
|
def resolve_log_path(cli_log: Optional[str]) -> str:
|
|
path = cli_log or os.environ.get("LLAMAGOCHI_LOG")
|
|
if not path:
|
|
config_file = Path("config.json")
|
|
if config_file.exists():
|
|
data = json.loads(config_file.read_text())
|
|
path = data.get("log")
|
|
if not path:
|
|
raise SystemExit(
|
|
"Error: log file path required.\n"
|
|
" Use --log /path/to/llama-server.log\n"
|
|
" or set LLAMAGOCHI_LOG env var\n"
|
|
" or set \"log\" key in config.json"
|
|
)
|
|
return path
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="Llamagochi — llama-server virtual pet")
|
|
ap.add_argument("--log", help="Path to llama-server log file")
|
|
ap.add_argument("--port", type=int, default=8080, help="HTTP port (default: 8080)")
|
|
ap.add_argument("--reopen-log", action="store_true",
|
|
help="Close and reopen log file each poll cycle (for sshfs/network filesystems)")
|
|
args = ap.parse_args()
|
|
log_path = resolve_log_path(args.log)
|
|
reopen_log = args.reopen_log or _config_bool("reopen_log")
|
|
app = create_app(log_path, reopen_log=reopen_log)
|
|
uvicorn.run(app, host="0.0.0.0", port=args.port)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|