Architecture

This page describes the internal architecture of ankerctl (Python). For the line-by-line reference of every package, file, and class, see CLAUDE.md in the repo root and .claude/agent-memory/INDEX.md.

Three-tier layout

┌──────────────────────────────────────────────────────────────┐
│  UI Layer       static/    HTML / JS / CSS (Cash.js)         │
├──────────────────────────────────────────────────────────────┤
│  API Layer      cli/       Click CLI commands                │
│                 web/       Flask routes, services            │
├──────────────────────────────────────────────────────────────┤
│  Protocol       libflagship/  MQTT, PPPP, HTTP clients       │
└──────────────────────────────────────────────────────────────┘

Layer	Folder	Purpose
UI	`static/`	HTML templates, JS, vendor bundles, screenshots
API	`cli/`	All `./ankerctl.py <group> <cmd>` implementations
API	`web/`	Flask app, REST + WebSocket routes, services
Protocol	`libflagship/`	Encrypted MQTT, PPPP UDP, Anker cloud HTTP

Repository tour

ankerctl.py              Main CLI entry point (Click)
cli/                     CLI command implementations
  config.py              config import / login / show / set-password
  mqtt.py                mqtt monitor / gcode / send / gcode-dump
  pppp.py                pppp lan-search / print-file / capture-video
  http.py                http calc-check-code / calc-sec-code
  util.py                shared helpers (patch_gcode_time, extract_layer_count)
  logfmt.py              logging setup (single source of truth)
libflagship/             Protocol clients
  mqtt.py                MQTT message types (auto-generated)
  mqttapi.py             MQTT client (encryption, topic routing)
  pppp.py                PPPP packet types (auto-generated)
  ppppapi.py             PPPP API (sockets, channels, file transfer)
  amtypes.py             Common AnkerMake types (auto-generated)
  megajank.py            AES, ECDH, PPPP curse/decurse
  httpapi.py             Anker cloud HTTP API
  seccode.py             Security code computation (v1 + v2)
  logincache.py          login.json / .ldb cache parsing
  notifications/         Apprise client
web/                     Flask web server
  __init__.py            Routes (REST + WebSocket), service registration
  notifications.py       AppriseNotifier
  service/               Background services
    mqtt.py              MqttQueue (heart of the app)
    pppp.py              PPPPService (LAN connection)
    video.py             VideoQueue (H.264 over PPPP channel 1)
    filetransfer.py      FileTransferService (G-code upload pipeline)
    history.py           PrintHistory (SQLite log)
    timelapse.py         TimelapseService (snapshot → MP4)
    homeassistant.py     HomeAssistantService (HA MQTT Discovery)
    filament.py          FilamentStore (SQLite profiles)
  lib/
    service.py           Service / ServiceManager framework
specification/           .stf protocol specs (input to codegen)
templates/               Codegen templates (Jinja2)
static/                  Web UI assets (HTML, JS, CSS, libflagship.js)
examples/                Standalone scripts for protocol experiments
documentation/           Markdown docs (see also: this Wiki)

Service framework

All long-running work lives in services registered in register_services() in web/__init__.py and managed by ServiceManager (in web/lib/service.py).

Lifecycle

Service extends threading.Thread. The background thread calls four lifecycle methods:

Method	When
`worker_init()`	Once on thread start, before any state transitions
`worker_start()`	Each time service transitions Stopped → Running
`worker_run(timeout)`	Repeatedly while Running
`worker_stop()`	Each time service transitions Running → Stopped

States: Starting → Running → Stopping → Stopped.

A service may raise ServiceRestartSignal from worker_run() to trigger a clean restart. This is used for example by VideoQueue after 3 consecutive stall recoveries.

Reference counting

ServiceManager tracks per-service refcounts. Services start on first borrow and stop when the count returns to zero (with the documented exceptions like VideoQueue).

# Idiomatic access pattern from a Flask route
with app.svc.borrow("mqttqueue") as mqtt:
    state = mqtt.get_state()
    mqtt.send_gcode("G28")
# refcount auto-decremented at end of with-block

# Non-blocking read (does not wait for the service to start)
pppp = app.svc.get("pppp", ready=False)
try:
    ...
finally:
    app.svc.put("pppp")

# Raw dict access (no refcount effect — read-only)
vq = app.svc.svcs.get("videoqueue")

Warning Accessing a service attribute outside the with borrow() block is a use-after-release bug. The service may have been stopped.

Notify / stream pattern

Services broadcast events via self.notify(data). ServiceManager.stream(name) returns a generator backed by a Queue tapped into service.handlers — it is what feeds the WebSocket endpoints.

@sock.route("/ws/mqtt")
def mqtt_ws(sock):
    for data in app.svc.stream("mqttqueue"):
        sock.send(json.dumps(data))

The stream uses a 1-second timeout to avoid blocking forever when the service stops.

Service catalog

Service	Registered as	Purpose	Started by
`MqttQueue`	`"mqttqueue"`	Cloud MQTT — drives the entire app	`register_services()`
`PPPPService`	`"pppp"`	LAN connection (UDP P2P)	First `borrow()`
`VideoQueue`	`"videoqueue"`	H.264 camera over PPPP channel 1	First `borrow()`
`FileTransferService`	`"filetransfer"`	G-code upload pipeline	First `borrow()`
`PrintHistory`	(sub-service of `MqttQueue`)	SQLite print log	`MqttQueue.worker_init()`
`TimelapseService`	(sub-service of `MqttQueue`)	Snapshot → MP4 assembly	`MqttQueue.worker_init()`
`HomeAssistantService`	(sub-service of `MqttQueue`)	HA MQTT Discovery	`MqttQueue.worker_init()`
`FilamentStore`	`app.filaments` (not a `Service`)	SQLite filament profiles	App startup

PrintHistory, TimelapseService, and HomeAssistantService are not independent services — they are owned by MqttQueue and accessed via mqtt.history, mqtt.timelapse, mqtt.ha. This way the print state machine (MqttQueue) can drive history records, timelapse triggers, and HA payloads from the same notification handler.

For per-method and per-attribute details see .claude/agent-memory/INDEX.md → "Key Classes Summary".

MQTT message flow (printer → frontend)

Printer publishes a message to /phone/maker/{SN}/notice
MqttQueue.worker_run() calls client.fetch() — decrypts the AES-256-CBC payload and returns a Python dict
For ct=1052 (layer info), total_layer is overridden with _gcode_layer_count if set (extracted from the G-code header at upload time)
self.notify(obj) broadcasts the raw dict to all handlers → WebSocket /ws/mqtt → frontend
self._forward_to_ha(obj) processes the same payload for Home Assistant Discovery updates
self._handle_notification(obj) processes for Apprise notifications, print history, and timelapse

The frontend receives raw JSON and dispatches by commandType. The 0–10000 progress scale (ct=1001) is not normalized server-side for the WebSocket — the frontend divides by 100 directly.

Key MQTT command types

For the full list and detailed descriptions, see documentation/MQTT_COMMANDS.md. Most-used:

Constant	`ct` value	Purpose
`ZZ_MQTT_CMD_GCODE_COMMAND`	1043	Send raw G-code
`ZZ_MQTT_CMD_PRINT_CONTROL`	1008	Pause (2) / Resume (3) / Stop (4) / Restart (0)
`ZZ_MQTT_CMD_AUTO_LEVELING`	1007	Start G29
`ZZ_MQTT_CMD_FIRMWARE_VERSION`	1002	Query firmware version

Notification types (printer → app):

`ct`	Field of interest	Meaning
1000	`value`	State: 0=idle, 1=printing, 2=paused, 8=aborted
1001	`progress` (0-10000), `realSpeed`, `time`, `filename`	Print progress
1003	`currentTemp`, `targetTemp`	Nozzle (1/100 °C units)
1004	`currentTemp`, `targetTemp`	Bed (1/100 °C units)
1007	`value`	Auto-level probe progress (50 total: 1 center + 7×7)
1044	`path`	Filename at print start
1052	`real_print_layer`, `total_layer`	Layer counts

See Protocol Details for encryption details and topic structure.

Web layer

Flask app lives in web/__init__.py. Routes are defined inline (no blueprints).

Authentication middleware

Implemented as _check_api_key() registered with @app.before_request. The rules:

GET requests are unauthenticated by default
POST / DELETE requests always require auth when ANKERCTL_API_KEY is set
A small allow-list of protected GET paths also requires auth (e.g. /api/settings/mqtt, /api/notifications/settings, /api/debug/*, /api/ankerctl/server/reload)
Setup paths (/api/ankerctl/config/upload, /api/ankerctl/config/login) are exempt when no printer is configured yet
All /api/debug/* paths require auth (prefix match)

WebSocket routes do not trigger before_request — /ws/ctrl therefore enforces auth inline (the first message must include the API key).

Frontend

HTML templates use Flask's Jinja2 (in-page, no separate templates/ folder for the app — most pages are static HTML in static/tabs/)
JavaScript uses Cash.js, Chart.js, and a custom AutoWebSocket wrapper with auto-reconnect (in static/ankersrv.js)
The video stream is raw H.264 NAL units rendered with jMuxer

Code generation

Files marked "DO NOT EDIT" are generated by transwarp from .stf specifications:

libflagship/mqtt.py
libflagship/pppp.py
libflagship/amtypes.py
static/libflagship.js

To change them, edit specification/*.stf or templates/*.j2 and run:

make diff      # preview
make update    # apply

See Development Guide for the codegen workflow.

Configuration storage

Configuration lives in JSON, not a database, so it is easy to inspect and edit.

File	Contents
`default.json`	Account (`user_id`, `auth_token`), per-printer (`mqtt_key`, `duid`, `sn`, alias, region), feature flags
`history.db` (SQLite)	Print history records
`filament.db` (SQLite)	Filament profile store

history.db and filament.db schemas auto-migrate on startup (ALTER TABLE ... IF NOT EXISTS).

For storage paths see Configuration.

Concurrency model

Concept	Implementation
Background work	`threading.Thread` subclass (`Service`)
Inter-thread queue	`queue.Queue` (replaced `multiprocessing.Queue` for safety in v1.0.0)
Mutex	`threading.Lock`, `threading.RLock`
Refcount	`ServiceManager` borrow / put pattern
Cancellation	`Service.stop()` flips state, `worker_run` exits on next iteration

The Flask process uses Werkzeug's WSGI server in dev mode and gunicorn-equivalent semantics in Docker (single process, multiple threads). Long-lived MQTT and PPPP sockets live in dedicated service threads, not in request threads.

Logging

cli/logfmt.py is the single source of truth for logging setup. It configures:

Root logger → ankerctl.log + stdout
Named loggers (mqtt, web, history, timelapse, homeassistant) → separate files in ANKERCTL_LOG_DIR

import logging
log = logging.getLogger(__name__)        # preferred
log = logging.getLogger("mqtt")          # named: writes to mqtt.log

Security Never log auth_token, mqtt_key, or api_key. The config show command redacts them; route handlers should follow the same pattern.

Reference: where to look in the source

For a guided reading order:

ankerctl.py — top-level Click groups and main()
cli/config.py — config file format and login flow
web/__init__.py — every REST and WebSocket route
web/service/mqtt.py — MqttQueue (the largest service)
libflagship/mqttapi.py — MQTT client and encryption
libflagship/ppppapi.py — PPPP UDP framing
documentation/MQTT_COMMANDS.md — message type reference

For the complete annotated index see CLAUDE.md and .claude/agent-memory/INDEX.md.

ankerctl Wiki

Links

Navigation Menu

Uh oh!

Architecture

Architecture

Three-tier layout

Repository tour

Service framework

Lifecycle

Reference counting

Notify / stream pattern

Service catalog

MQTT message flow (printer → frontend)

Key MQTT command types

Web layer

Authentication middleware

Frontend

Code generation

Configuration storage

Concurrency model

Logging

Reference: where to look in the source

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally