Sources

Each source is a protocol adapter configured under adapters (or nats/mqtt for streaming). All normalize to CanonicalMachineData.

Schedule source

A schedule:{id} source node runs a flow on a clock, with no device and no caller: the nightly sales summary, an hourly sync. It is a flow source (set on the canvas or in flow.json), not an adapters: entry, so it deploys like any other node.

{ "id": "nightly", "type": "source", "match": "schedule:nightly-sales",
  "cron": "30 0 * * *", "timezone": "Asia/Jakarta",
  "payload": { "report": "daily-sales" } }

Each tick enqueues one record under route key schedule:{id} (source schedule, machine id {id}): your payload plus a reserved _schedule object.

{ "report": "daily-sales",
  "_schedule": { "id": "nightly-sales", "cron": "30 0 * * *", "timezone": "Asia/Jakarta",
                 "scheduled_for": "2026-10-31T00:30:00+07:00",
                 "scheduled_for_utc": "2026-10-30T17:30:00Z",
                 "date": "2026-10-31", "previous_date": "2026-10-30" } }

date is the date in the zone, which differs from the UTC date around midnight. A job that summarises the day that just ended reads _schedule.previous_date.

• Deploy checks: an invalid cron, an unknown zone, a wildcard id, a payload that is not an object (or uses _schedule), or a cron with no future fire time fail the deploy with the error on the node.
• No catch-up: ticks that pass while the connector is stopped are not replayed. A tick observed more than 60 s after its scheduled time (the host was suspended, the clock was stepped) is skipped and logged. After a restart the schedule resumes at the next future tick.
• Overlap: every tick is enqueued on time. If the previous run is still going, the new tick queues behind it (records of one route key run in order, one at a time).
• Failures end like any record, so a failing transform raises the pipeline-failure alert (one per node per window, see Alerts).
• Visibility: the editor’s source panel shows the next runs and the last tick’s outcome; the same data is at GET /api/schedules, and POST /api/schedules/preview previews any cron. POST /api/ingest/schedule/{id} also runs the pipeline by hand: the body is the record as is, without _schedule.
• Testing: Test on the node dry-runs the pipeline with a tick for now (no side effects).

CNC adapters: FOCAS vs MTConnect

OpexFlow ships both a FOCAS adapter and an MTConnect adapter, and normalizes them to the same CanonicalMachineData shape — so downstream transforms and sinks don’t care which one produced the event.

• FOCAS (FANUC Open CNC API Specifications) talks to the control over FOCAS/Ethernet (TCP 8193) and reads run state, axis positions, spindle data, and alarms directly. It gives depth (controller internals the standardized protocols don’t expose) but is Fanuc-specific.
• MTConnect is an open standard (OpexFlow reads the agent’s JSON /current endpoint). Many modern controls — Fanuc included — expose an MTConnect agent that streams a standardized data-item model (spindle speed, feed, program, axis positions, controller mode). One consumer works across vendors.

For a mixed floor: read Fanuc machines that need deep data with FOCAS and everything else with MTConnect. Both land in the same normalized schema.

FOCAS modes

FOCAS has two backends behind one adapter. adapters.focas.mode picks one:

adapters:
  focas:
    enabled: true
    mode: auto                 # auto | native | library
    # lib_path: libfwlib32.so  # your Fanuc library; unset = native unless it sits beside the binary
    devices:
      - id: fanuc-mill-1
        name: Fanuc VF-2SS
        host: 192.168.1.10
        port: 8193             # default
        timeout_ms: 5000       # default: connect and per-request timeout
        max_axes: 3
        macros: [500, 501]     # native only: macro variables read on every poll

Both modes build the record the same way: a poll reads status, axes, spindle speed and alarm bits, and the control’s state maps to the machine state as follows.

axes hold the absolute position per axis in mm, spindle.speed the actual spindle speed. metadata carries the raw status words (run, motion, mode, edit, emergency, alarm) and alarm_bits. Record quality is good whenever the status read worked. A control that does not answer, or refuses the status read, yields an offline record (metadata.error). Any other item the control refuses (an option it lacks, an older series) is left out and named in metadata.read_warnings; the rest of the record stays valid.

Extra reads in native mode

The native client reads more than the library path does. These land in metadata:

A poll opens one connection to the control, reads, and closes it, as the library path does.

Demo simulator

The native client ships with a CNC simulator that speaks the same protocol, so the UI can show a machine without hardware:

• simulator: true on a device (no host needed) starts a simulated 3-axis mill inside the connector on a loopback port. It runs a demo program in a loop (positions, feed, spindle and the part counter in macro #500 move) and is always polled with the native client, in any mode.
• opex-flow focas-sim --port 8193 runs the same simulator as a separate process (--bind, --profile 31i|0i, --idle); point a device’s host and port at it.

Cross-checked, and still unverified

The two backends were compared field by field against fwlib32.h and a test checks that both produce the same record for the same simulated controller state (idle, cutting, feed hold, alarm, emergency stop, suffixed axis name).

• Status words. fwlib32.h declares ODBST in three variants. The library path uses the default one: nine shorts, hdck, tmmode, then aut, run, motion, mstb, emergency, alarm, edit. The native call returns the seven words from aut on (14 bytes), which is the same order, so both use the same six words (mstb is not mapped). Native also accepts an 18-byte payload as that nine-word layout. Not confirmed: that a real control sends 14 bytes (pyfanuc only decodes it for CNC types 16 and 31, so other families such as 0i-F or 30i-B are untested). The 15-series variant of the header has a different word order and is not supported.
• Axis names. The library returns two characters per axis (name, suffix). The native call returns four bytes per axis; the name is the first, a suffix is the first non-zero byte after it. Labels agree (X, Z2) for names of one or two characters; the captures show only suffix-less names.
• Position units. The library returns the control’s least input increment and the adapter divides by 1000 (0.001 mm). Native uses the decimals the control sends with each number. They agree on a metric control with three decimals. With another resolution (for example 0.0001 mm) or inch input the library path is off by a power of ten and both modes still label the value mm, also for rotary axes.
• Alarm bits and list. The library reads cnc_alarm2, native the cnc_alarm bit field; the same bit meaning is assumed. Alarm type names follow the 30i/0i-D/F table; older series number the bits differently. Native lists alarms through cnc_rdalmmsg, whose first entry is assumed to be the library’s first entry. Native asks for 32 characters of text per alarm (the ODBALMMSG size), the library path accepts up to 64, so a long message can be shorter in native mode.
• Spindle and feed. Decimal places of feed and spindle values were not in the captures (the simulator sends none).
• Servo load, program name. The servo-load request argument is inferred (the capture shows the request and the response, no reference function); the program-name call exists on series 30i and later only.
• Open response. The simulator answers the open request with an empty payload; real controls send about 360 bytes that no client reads. A control that insists on something in the reply is not covered.
• Limits. The client sends at most nine sub-packets per request (the largest seen in captures); the control’s own limit for simultaneous FOCAS connections applies.

FANUC and FOCAS are trademarks of FANUC CORPORATION; focas-proto is independent, contains no FANUC code or library, and is not endorsed by FANUC.

MQTT semantics

Each mqtt.sources[] entry has mode: latest | stream (default latest, same as NATS devices[].mode). latest keeps one value per subscription (messages between two polls overwrite each other); stream queues every message. Both MQTT and NATS emit a record only on a fresh message (cached values are never replayed); a silent source emits an offline record every 60 s. An unparseable data_type: number payload yields metadata.value: null with record quality: "bad". Every message is wrapped as metadata.value with state: Running (MQTT does not pass canonical records through; NATS does).

• Wildcards: a wildcard topic filter produces one machine per concrete topic (<id>:<topic>, real topic in metadata.topic); subscriptions are re-established on every reconnect. Client ids are <client_id>-<role>-<random>, so the configured client_id is only a prefix.
• NATS: separate durable consumers per mode. Device subjects may use * and >. Unreachable-at-boot servers are retried with backoff. The sink subject is telemetry.<percent-encoded machine id>. See NATS delivery semantics.
• MQTT delivery: messages are acknowledged to the broker on receipt (rumqttc automatic acks). Manual acks after sink success are possible in rumqttc, but the connector uses a clean session with a random client id per connection, so a broker never redelivers an unacknowledged message after a reconnect — holding back the ack would add no durability. Use the NATS MQTT gateway (devices → JetStream) when you need ingest that survives a restart.
• Sinks never block the pipeline: publishes are queued and drops are counted.

NATS delivery semantics

A JetStream message is acknowledged after the record is durably handled downstream, not when it is received or handed to the router. Each record carries a completion handle through the pipeline; the message is settled once every part of it has finished:

• “Durable” for the push sink means stored in the push queue (push_queue.db when connector.push_queue.enabled, otherwise the in-memory buffer), not delivered to the cloud: the push drain retries and dead-letters on its own from there. storage, clickhouse and postgres count a record as handled once it is accepted into their write queue (a batch writer may still hold it for a moment).
• Fan-out: a map over N elements, a branch, or several sinks settle the message only when all children finished; the worst outcome wins (nak > term > ack). One failed child therefore redelivers the whole message, and children that already succeeded run again: delivery is at-least-once and sinks may see duplicates. A device subject matching two configured devices is handled the same way.
• Backpressure: the consumer’s max_ack_pending (1000 for latest; 2 × 1024 × devices for stream) bounds unacked messages across the per-device queues, the pipeline and the sinks. When sinks stall the server stops delivering. A stream queue that overflows (1024 per subject) returns its oldest message to JetStream (nak) instead of dropping it. An in-flight message is kept alive with Progress heartbeats every 40 s (ack_wait is 120 s) so a slow sink does not cause duplicate deliveries.
• Ordering: redelivery can reorder messages relative to newer ones; records carry their JetStream publish timestamp.

OPC-UA security

These keys are for reading from an OPC-UA server. To serve OpexFlow’s own tags to a SCADA or MES client, see OPC-UA server.

security selects the policy: none, basic128rsa15, basic256, basic256sha256, aes128sha256rsaoaep or aes256sha256rsapss. security_mode is none | sign | signandencrypt (default signandencrypt unless the policy is none). username / password are refused on policy none unless allow_insecure_credentials: true.

Server certificates are verified: trust_all_server_certs defaults to false. To trust a server, move its certificate from <data_dir>/pki/rejected to <data_dir>/pki/trusted (or set trust_all_server_certs: true on an isolated network, which disables MITM protection). Other keys: connect_timeout_ms (10000), request_timeout_ms (5000), interval_ms (1000), max_nodes_per_read (100). node_id must be real NodeId syntax (ns=2;s=Name); nodes returning a bad status are skipped and listed in metadata.opcua_bad_nodes; the record timestamp is the server source timestamp.

OPC-UA subscribe mode

mode: subscribe replaces the poll with one OPC-UA subscription per endpoint: the server pushes data changes, so a fast-changing tag is not limited by interval_ms and an idle one costs no traffic. The default stays mode: poll.

opcua:
  enabled: true
  endpoints:
    - id: "CNC-01"
      endpoint: "opc.tcp://192.168.1.201:4840"
      mode: subscribe
      publishing_interval_ms: 250     # default: interval_ms, else 1000
      nodes:
        - node_id: "ns=2;s=Spindle.Speed"
          alias: "spindle_speed"
          mapping: "spindleSpeed"
          sampling_interval_ms: 100   # default: the publishing interval; 0 = server's fastest
          deadband: { type: absolute, value: 5 }   # or { type: percent, value: 1 }
          queue_size: 10              # default 1
        - node_id: "ns=2;s=Machine.Status"
          alias: "status"
          mapping: "state"

These options are rejected at config load unless mode: subscribe. max_nodes_per_read also bounds the items per CreateMonitoredItems call.

Record semantics. Records go through the same normalisation as polling (mappings, quality, valueQuality, opcua_bad_nodes, state unknown for an unreadable state node, offline record when no node is usable), so downstream flows see one shape. Each notification message produces one record carrying the merged snapshot of the latest value of every node, not one record per tag; a tag that did not change keeps its last value. The record timestamp is the newest server source timestamp in the snapshot. If a node reports several queued samples in one message (queue_size > 1), one record per sample round is emitted, oldest first. Records reach the pipeline on the next tick of the connector poll loop (connector.poll_interval_ms), which drains them; at most 1000 are buffered per endpoint (oldest dropped, logged). A subscription emits nothing while nothing changes, so an idle machine is quiet rather than re-reported every interval; the first publish after subscribing delivers the initial value of every node.

Robustness. The client library reconnects the session and moves (TransferSubscriptions) or recreates the subscription and its monitored items; initial values arrive again. The adapter additionally rebuilds the session and subscribes again when the event loop gives up, the server ends the subscription (status change notification), the items were not restored after a reconnect, or no publish response (data or keep-alive) arrives for 3 keep-alive periods (at least 10 s). An offline record is emitted when a subscription is lost.

Fallback. If the server refuses the subscription or any monitored item (too many subscriptions, unsupported service, unsupported filter such as percent without EURange), that endpoint is polled every interval_ms instead, and the reason is shown as the adapter/source last_error (subscription refused (…); polling every 1000ms). It tries to subscribe again every 60 s and on every new session.

Status. In subscribe mode last_read is the time of the last data change notification, data_points counts values delivered, connected follows the session (false while the library reconnects).

Modbus and EtherNet/IP notes

Modbus devices accept parity, stop_bits, data_bits, timeout_ms (3000), word_order / byte_order (big default; CDAB = word_order: little). Registers take function (1 coil, 2 discrete, 3 holding default, 4 input) and types uint16, int16, uint32, int32, uint64, int64, float, float64, bool, bit (+bit 0-15), string (+length words). EtherNet/IP tag type is ignored: values decode by the type the PLC returns (BOOL, SINT, INT, DINT, LINT, REAL, DWORD); timeout_ms defaults to 5000.

Database sources

clickhouse.sources[] and postgres.sources[] accept an optional cursor_column: an ascending column used as a persisted watermark (<data_dir>/cursors.json) so only newer rows are emitted (max 10 000 rows per poll). A failed fetch yields an offline record.

Postgres TLS

The Postgres source and sink share one pool and speak TLS through rustls (no OpenSSL; the postgres feature builds on static musl targets). Settings under postgres::

sslmode, sslrootcert, sslcert and sslkey query parameters in postgres.url (postgres://u:p@db/x?sslmode=verify-full&sslrootcert=/etc/ssl/db-ca.pem) are honoured and override the matching keys. Modes follow libpq:

• prefer uses TLS when the server offers it and falls back to plaintext only if the server refuses TLS. It does not check the certificate, so it does not stop an active attacker.
• require always encrypts but, like libpq, does not check the certificate. With ca_cert set it also verifies the chain (same as verify-ca).
• verify-ca checks the certificate chain against the trusted CAs, not the hostname.
• verify-full also checks that the host in the URL matches the certificate (DNS name or IP SAN). Use it for anything beyond a trusted LAN.

Use a server certificate signed by a CA (public or private, passed as ca_cert). A self-signed server certificate works as ca_cert only if it is a valid end-entity certificate (basicConstraints=CA:FALSE, with a SAN); openssl req -x509 defaults to CA:TRUE, which rustls rejects. Certificates are read once at startup, so restart after rotating them. A key=value connection string cannot carry sslmode; use a postgres:// URL or the sslmode key.