A codec for a device's own protocol

The physical worldMQTTLineCodecdecodeencodeMACHINE_CODEC_MODULEcreateCodec

A 3D printer on the lineStep 4 of 7The printer claims and reports

The printer speaks gcode_state, stg_cur and print_error. The gateway speaks dispatch, started and arrived.

← A machine that hosts its own broker Milestones on the escalation →

The codec reads a printer's reports as line-protocol events and writes the gateway's commands as printer requests. Workflows and escalations never see the printer's protocol.

The situation

Every machine has a protocol of its own. The gateway should speak one line protocol, and each machine’s protocol should live in one place that turns it into that line protocol and back. Built by hand, the printer’s states leak into every workflow that touches it. Here a codec does the translation and nothing else.

The setup

A codec names its channels and translates both ways:

return {
  root,
  channels: { events: [BAMBU_CHANNEL.REPORT, BAMBU_CHANNEL.BRIDGE], telemetry: [], commands: [BAMBU_CHANNEL.REQUEST] },
  topicFor: (scope, channel) => `${root}/${scope.siteId ?? '+'}/${scope.lineId ?? '+'}/${scope.address ?? '+'}/${channel}`,
  unitsPerDispatch: 1,
  snapshotOnStart: true,
  decode(topic, payload) { /* a report or a bridge state, as line-protocol events */ },
  encode(ident, command) { /* dispatch, pause, resume, cancel, snapshot, as printer requests */ },
};

The host passes it to start(). The development server loads it from MACHINE_CODEC_MODULE, whose createCodec(line) returns it:

import { start, Machines } from '@hotmeshio/long-tail';
// Copied from the long-tail repository's examples/workflows/bambu-printer
import { bambuCodec } from './bambu-printer/codec';

const line = Machines.loadLine('/etc/site/line.json');

await start({
  database,
  workers,
  machines: {
    lines: [line],
    codec: bambuCodec({ root: 'line/v1', line }),
  },
});

encode turns a dispatch into project_file, naming the print <jobId>~<unitId>:

{
  "print": {
    "sequence_id": "<print escalation id>:dispatch",
    "command": "project_file",
    "param": "Metadata/plate_1.gcode",
    "project_id": "0",
    "profile_id": "0",
    "task_id": "0",
    "subtask_id": "0",
    "subtask_name": "Pmw8druzy-p1s-01~U-Pmw8druzy",
    "url": "file:///sdcard/cache/Pmw8druzy.gcode.3mf",
    "md5": "",
    "timelapse": false,
    "bed_type": "auto",
    "bed_levelling": true,
    "flow_cali": false,
    "vibration_cali": false,
    "layer_inspect": false,
    "use_ams": false,
    "ams_mapping": []
  }
}

decode keeps one merged report per printer and reads the phase from it:

export function phaseOf(r: PrinterReport): Phase {
  switch (r.gcode_state) {
    case PRINT_STATE.PREPARE:
      return PHASE.PREPARING;
    case PRINT_STATE.RUNNING: {
      // The stage says so when the printer reports one; otherwise a layer under way does.
      const laying = typeof r.stg_cur === 'number' ? r.stg_cur === STAGE_PRINTING : typeof r.layer_num === 'number' && r.layer_num >= 1;
      return laying ? PHASE.PRINTING : PHASE.PREPARING;
    }
    case PRINT_STATE.PAUSE:
      return PHASE.PAUSED;
    case PRINT_STATE.FINISH:
      return PHASE.FINISHED;
    case PRINT_STATE.FAILED:
      return PHASE.FAILED;
    default:
      return PHASE.IDLE;
  }
}

On the floor

The printer echoes subtask_name in every report, so each report names its escalation. RUNNING with stg_cur 0 is started. A crossed percent is progress. FINISH is the unit’s arrival and the job’s finished record. FAILED, or a return to idle before printing, is a withdrawal with a reason. An acknowledgement under the request’s sequence_id is the reply.

The guarantee

The workflow is the stock machineUnitRoute, and the escalations are ordinary handoffs. Replace the codec and the same route runs on a different machine. A replayed report yields the same envelope ids, so it is never acted on twice. See Machines.

The physical world