Skip to content

DmxOutput.kt

src/jvmMain/kotlin/afield/io/dmx/DmxOutput.kt · 373 lines

@file:afield.runtime.completion.CompletionCategory("Lighting")

package afield.io.dmx

import java.net.DatagramPacket
import java.net.DatagramSocket
import java.net.InetAddress
import java.net.InetSocketAddress
import java.util.UUID
import java.util.concurrent.ConcurrentSkipListMap
import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicLong

DMX transmission: the frame buffers a box writes into, and the thread that puts them on the wire.

DmxUniverse is one universe of 512 slots, addressed 1-based as consoles and fixture manuals address them. DmxOutput owns a set of those and a daemon sender thread that ticks at up to 44 Hz, sending a universe whose bytes have changed and re-sending an unchanged one often enough that no receiver times it out. DmxProtocol picks the wire format, encoded by ArtNet or Sacn: Art-Net needs a destination host, and sACN goes to each universe's multicast group unless one is given.

Nothing here waits on Field's frame clock. beforeTick runs at the top of each sender tick, before the universes are snapshotted, which is where Patch animate computes the next look, so a rig keeps running whether or not a renderer is ticking.

In a box, reach an output through artnetOut or sacnOut rather than this constructor: those hand back the same output across re-runs and close it when the box stops. The user-facing page is Driving Lights: Art-Net and sACN.

enum class DmxProtocol { ARTNET, SACN }

Which wire protocol a DmxOutput speaks or a DmxInput listens for.

class DmxUniverse(val number: Int) : afield.ui.visualization.ToHTML

One universe of DMX data: 512 slots, written by Field code, read by a DmxOutput's sender thread.

Channels are 1-based at this API, as on every console and fixture manual (u[1] is the first slot); data is the raw 0-based array for bulk copies. Values are clamped into 0..255. Evaluating one in a box draws its slots.

class DmxUniverse(val number: Int) : afield.ui.visualization.ToHTML {

val data = ByteArray(512)

The raw slots, 0-based. Writing here directly is fine; the sender copies it each tick.

var length

How many slots are transmitted; 512 unless a rig wants shorter packets.

    var length = 512
        set(value) { require(value in 1..512); field = value }

operator fun get(channel: Int): Int = data[channel - 1].toInt() and 0xff

Value of 1-based channel, 0..255.

operator fun set(channel: Int, value: Int)

Set 1-based channel to value, clamped to 0..255.

    operator fun set(channel: Int, value: Int) {
        data[channel - 1] = value.coerceIn(0, 255).toByte()
    }

fun setLevel(channel: Int, level: Double)

Set 1-based channel from a 0..1 level.

    fun setLevel(channel: Int, level: Double) = set(channel, Math.round(level.coerceIn(0.0, 1.0) * 255).toInt())

fun set16(channel: Int, value: Int)

Set a 16-bit value (0..65535) into channel (coarse) and channel + 1 (fine).

    fun set16(channel: Int, value: Int) {
        val v = value.coerceIn(0, 65535)
        data[channel - 1] = (v shr 8).toByte()
        data[channel] = v.toByte()
    }

fun clear() = java.util.Arrays.fill(data, 0)

Zero every slot.

override fun toHTML(): String

Evaluating a universe in a box draws its slots as a 32 × 16 grid of gray cells, see universeToFLines.

    override fun toHTML(): String =
        afield.ui.visualization.linesToHTML(universeToFLines(data, length = length), 384.0, 192.0, fitToSize = true) +
            "<div style='font-family:monospace;font-size:11px'>universe $number</div>"

override fun toString()

The universe number and its first twelve slot values, or all of them if length is shorter.

    override fun toString() = "DmxUniverse($number, first slots ${(1..minOf(12, length)).map { get(it) }})"
}

class DmxOutput

A DMX transmitter: a set of DmxUniverse frame buffers and a thread that sends them as Art-Net or sACN.

You write into universes whenever you like — out.universe(1)[1] = 255, or through a Patch — and the sender thread, ticking at maxRateHz, decides what goes on the wire:

  • Send on change. A universe whose bytes differ from what was last sent is sent on the next tick. Ticks are at most 44 Hz by default, the DMX512 full-frame rate that both protocols treat as the ceiling for a universe.
  • Keep-alive. An unchanged universe is re-sent once keepAliveMs (default 900 ms) have passed, on the next tick — so the real interval is 900 ms plus up to one tick period (22.7 ms at 44 Hz). E1.31-2016 §6.6.2 asks a source that suppresses unchanged data to send three packets of the unchanged values and then one keep-alive every 800–1000 ms, and receivers drop a source after E131_NETWORK_DATA_LOSS_TIMEOUT (2.5 s); Art-Net 4 likewise expects an unchanged universe to be refreshed at least every 4 s. The default satisfies both. (It was 1000 ms until a Python decoder timed the keep-alives at 1.022 s — the tick granularity pushed them past the window.) The three post-change repeats are sent for both protocols, because Art-Net nodes do not mind them.
  • Sync (optional). With sync on, every tick that sent anything ends with one ArtSync, or one sACN synchronization packet on syncUniverse (data packets then carry that sync address), so a multi-universe strip updates in one step instead of tearing across universes.
  • Stream terminated. Closing an sACN output sends three packets per universe with the Stream_Terminated option set (E1.31 §6.2.6), so receivers release the universe immediately instead of holding the last look for 2.5 s.

Destinations: Art-Net goes to host (a node's IP, or a broadcast address). sACN goes to host if one is given (unicast), otherwise to each universe's multicast group 239.255.hi.lo.

In a box, use artnetOut / sacnOut rather than the constructor: they reuse the output across re-runs and close it when the box is stopped.

declaration — 80 lines, from line 114 
class DmxOutput(
    /** Which wire format goes out. It also decides the default [port] and whether [host] is required. */
    val protocol: DmxProtocol,
    /**
     * Where packets are sent. Art-Net needs one — a node's IP, or a broadcast address — and the
     * constructor refuses to build without it. sACN unicasts to it when it is given and otherwise
     * sends to each universe's multicast group.
     */
    val host: String? = null,
    /** UDP port to send to: 6454 for Art-Net, 5568 for sACN. */
    val port: Int = if (protocol == DmxProtocol.ARTNET) ArtNet.PORT else Sacn.PORT,
    /**
     * The name this source announces in sACN data and discovery packets, at most 63 bytes of UTF-8
     * in a 64-byte field. ArtDmx has no source-name field, so Art-Net ignores it.
     */
    val sourceName: String = "Field4",
    /**
     * sACN priority, 0..200. A receiver uses only the highest priority live on a universe and merges
     * within it, so raising this takes a universe over from a lower-priority source. Art-Net has no
     * equivalent field.
     */
    val priority: Int = Sacn.DEFAULT_PRIORITY,
    /**
     * This source's sACN component identifier, a fresh UUID per output. It is how a receiver tells
     * two sources on one universe apart, so an output that is closed and rebuilt arrives as a new
     * source rather than as a continuation of the old one.
     */
    val cid: UUID = UUID.randomUUID(),
    /**
     * Sender ticks per second, at most 1000. A tick sends the universes that changed, so this is the
     * ceiling on how often one universe goes out rather than a rate it is held to: an unchanging rig
     * sends only keep-alives. 44 Hz is the DMX512 full-frame rate both protocols treat as the limit.
     */
    val maxRateHz: Double = 44.0,
    /**
     * How long an unchanged universe may go unsent before it is re-sent. The default of 900 ms sits
     * inside E1.31's 800–1000 ms keep-alive window with a tick period to spare, and well inside both
     * protocols' data-loss timeouts.
     */
    val keepAliveMs: Long = 900,
    /**
     * End every tick that sent anything with one synchronization packet, so a rig spread over
     * several universes latches in one step instead of tearing across them.
     */
    val sync: Boolean = false,
    /**
     * The sACN synchronization address: carried by data packets and used for the synchronization
     * packet when [sync] is on. Art-Net's ArtSync addresses nothing, so Art-Net ignores it.
     */
    val syncUniverse: Int = 7962,
    /**
     * Send sACN universe-discovery pages every 10 seconds listing the universes this output has been
     * asked for. They go to one well-known universe, so a tool learns what this source transmits
     * without joining the multicast group of every universe it might be on. On by default for sACN,
     * and unused for Art-Net, where discovery runs the other way round as ArtPoll.
     */
    val discovery: Boolean = protocol == DmxProtocol.SACN,
    autoStart: Boolean = true,
) : AutoCloseable {

    private class Stream(val universe: DmxUniverse) {
        val snapshot = ByteArray(512)
        val lastSent = ByteArray(512)
        val packet = ByteArray(maxOf(ArtNet.DMX_MAX_PACKET, Sacn.DATA_MAX_PACKET))
        var sequence = 0
        var lastSendNanos = 0L
        var repeatsLeft = 0
        var everSent = false
    }

    private val streams = ConcurrentSkipListMap<Int, Stream>()
    private val socket = DatagramSocket().apply { broadcast = true }
    private val cidBytes = Sacn.cidBytes(cid)
    private val syncPacket = ByteArray(Sacn.SYNC_LENGTH.coerceAtLeast(ArtNet.SYNC_LENGTH))
    private val discoveryPacket = ByteArray(120 + 2 * 512)
    private var syncSequence = 0
    private var lastDiscoveryNanos = 0L
    private val lock = Any()
    private val startNanos = System.nanoTime()
    private var executor: ScheduledExecutorService? = null

var isClosed

True once close has run. A closed output does not restart: start returns, tick sends nothing, and the socket is gone.

    @Volatile
    var isClosed = false
        private set

    private val sent = AtomicLong()

val packetsSent: Long get() = sent.get()

Data packets sent so far (sync and discovery packets not counted).

var lastError: Throwable?

The last send failure, or null. Sends never throw into Field code; look here instead.

    @Volatile
    var lastError: Throwable? = null
        private set

    private val hostAddress: InetAddress? = host?.let { InetAddress.getByName(it) }

init

    init {
        require(maxRateHz > 0 && maxRateHz <= 1000) { "maxRateHz must be in (0, 1000], got $maxRateHz" }
        if (protocol == DmxProtocol.ARTNET) require(hostAddress != null) { "Art-Net needs a destination host (a node IP or broadcast address)" }
        if (autoStart) start()
    }

var beforeTick: ((seconds: Double) -> Unit)? = null

Called with seconds since this output was created at the start of every sender tick, on the sender thread, before the universes are snapshotted — the place to compute the next look, so lights animate on the DMX clock whether or not anything else in Field is ticking. There is one slot, not a list: re-running a box replaces the animation instead of stacking a second one. Exceptions are caught into lastError. See animate and Patch.animate.

    @Volatile
    var beforeTick: ((seconds: Double) -> Unit)? = null

fun animate(fn: (seconds: Double) -> Unit) {

Set beforeTick: run fn once per sender tick, with seconds since the output started.

    fun animate(fn: (seconds: Double) -> Unit) {
        beforeTick = fn
    }

fun universe(number: Int): DmxUniverse

The frame buffer for number, created (all zeros) on first use.

    fun universe(number: Int): DmxUniverse {
        if (protocol == DmxProtocol.SACN) require(number in 1..63999) { "sACN universes are 1..63999, got $number" }
        else PortAddress(number) // validates 0..32767
        return streams.computeIfAbsent(number) { Stream(DmxUniverse(it)) }.universe
    }

val universes: List<Int> get() = streams.keys.toList()

The universes this output has been asked for, ascending.

fun <T> frame(block: DmxOutput.() -> T): T = synchronized(lock) { block() }

Run block holding the frame lock, so the sender never transmits a half-written frame. Every Patch.render into an output goes through here.

fun start()

Start the sender thread (the constructor does this unless autoStart = false).

    fun start() {
        if (executor != null || isClosed) return
        val ex = Executors.newSingleThreadScheduledExecutor { r ->
            Thread(r, "afield-dmx-${protocol.name.lowercase()}-out").apply { isDaemon = true }
        }
        val periodMicros = (1_000_000.0 / maxRateHz).toLong()
        ex.scheduleAtFixedRate({ runCatching { tick(System.nanoTime(), force = false) }.onFailure { lastError = it } },
            0, periodMicros, TimeUnit.MICROSECONDS)
        executor = ex
    }

fun flush() = tick(System.nanoTime(), force = true)

Send every universe now, regardless of change or rate; for tests and one-shot scripts.

fun tick(nowNanos: Long, force: Boolean = false): Int

One sender pass at time nowNanos; returns how many data packets it sent. Public so tests can drive the send-on-change / keep-alive policy with a synthetic clock and no thread.

body — 77 lines, from line 271 
    @Synchronized
    fun tick(nowNanos: Long, force: Boolean = false): Int {
        if (isClosed) return 0
        var count = 0
        val keepAliveNanos = keepAliveMs * 1_000_000
        beforeTick?.let { hook ->
            runCatching { hook((nowNanos - startNanos) / 1e9) }.onFailure { lastError = it }
        }
        synchronized(lock) {
            for (s in streams.values) {
                System.arraycopy(s.universe.data, 0, s.snapshot, 0, 512)
            }
        }
        for ((number, s) in streams) {
            val changed = !s.everSent || !java.util.Arrays.equals(s.snapshot, s.lastSent)
            val due = when {
                force || changed -> { s.repeatsLeft = 2; true }
                s.repeatsLeft > 0 -> { s.repeatsLeft--; true }
                else -> nowNanos - s.lastSendNanos >= keepAliveNanos
            }
            if (!due) continue
            if (send(number, s, options = 0)) {
                System.arraycopy(s.snapshot, 0, s.lastSent, 0, 512)
                s.lastSendNanos = nowNanos
                s.everSent = true
                count++
            }
        }
        if (count > 0 && sync) sendSync()
        if (discovery && protocol == DmxProtocol.SACN && streams.isNotEmpty() &&
            nowNanos - lastDiscoveryNanos >= Sacn.DISCOVERY_INTERVAL_MS * 1_000_000) {
            lastDiscoveryNanos = nowNanos
            sendDiscovery()
        }
        return count
    }

    private fun send(number: Int, s: Stream, options: Int): Boolean {
        val len = s.universe.length
        s.sequence = if (protocol == DmxProtocol.ARTNET) (s.sequence % 255) + 1 else (s.sequence + 1) and 0xff
        val (n, dest) = when (protocol) {
            DmxProtocol.ARTNET ->
                ArtNet.encodeDmx(s.packet, PortAddress(number), s.snapshot, len, sequence = s.sequence) to hostAddress!!
            DmxProtocol.SACN ->
                Sacn.encodeData(
                    s.packet, cidBytes, sourceName, number, s.snapshot, len,
                    sequence = s.sequence, priority = priority, options = options,
                    syncAddress = if (sync) syncUniverse else 0,
                ) to (hostAddress ?: Sacn.multicastAddress(number))
        }
        return transmit(s.packet, n, dest).also { if (it) sent.incrementAndGet() }
    }

    private fun sendSync() {
        when (protocol) {
            DmxProtocol.ARTNET -> transmit(syncPacket, ArtNet.encodeSync(syncPacket), hostAddress!!)
            DmxProtocol.SACN -> {
                syncSequence = (syncSequence + 1) and 0xff
                val n = Sacn.encodeSync(syncPacket, cidBytes, syncUniverse, syncSequence)
                transmit(syncPacket, n, hostAddress ?: Sacn.multicastAddress(syncUniverse))
            }
        }
    }

    private fun sendDiscovery() {
        val list = streams.keys.toIntArray().take(512).toIntArray()
        val n = Sacn.encodeDiscovery(discoveryPacket, cidBytes, sourceName, list)
        transmit(discoveryPacket, n, hostAddress ?: Sacn.multicastAddress(Sacn.DISCOVERY_UNIVERSE))
    }

    private fun transmit(buf: ByteArray, n: Int, dest: InetAddress): Boolean = try {
        socket.send(DatagramPacket(buf, n, InetSocketAddress(dest, port)))
        true
    } catch (t: Throwable) {
        lastError = t
        false
    }

override fun close()

Stop the sender and release the socket. An sACN output first sends three stream-terminated packets per universe. Idempotent.

    override fun close() {
        if (isClosed) return
        executor?.shutdownNow()
        runCatching { executor?.awaitTermination(250, TimeUnit.MILLISECONDS) }
        executor = null
        if (protocol == DmxProtocol.SACN) synchronized(this) {
            synchronized(lock) { for (s in streams.values) System.arraycopy(s.universe.data, 0, s.snapshot, 0, 512) }
            repeat(3) { for ((number, s) in streams) send(number, s, Sacn.OPTION_STREAM_TERMINATED) }
        }
        isClosed = true
        runCatching { socket.close() }
    }

override fun toString()

The protocol, where packets go, the universes this output has been asked for, packetsSent, and closed once it is. Enough to tell a live sender from one that never found a destination.

    override fun toString() =
        "DmxOutput(${protocol.name}, -> ${host ?: "multicast"}:$port, universes=${universes}, sent=$packetsSent${if (isClosed) ", closed" else ""})"
}