Xignal and YAOPCUA: Soft PLC software in Elixir

It will look very foreign to someone who is used to ladder logic or function block.

A module with a bunch of function calls for defining tags or defining them at any point in a program is going to be a lot messier and harder to maintain than a table with sorting and filtering.

Did it have a front end, i.e What is the live debugging story? Can you display trends? Can you see tag values?

Does it have a testing framework?

Can you have separate routines, programs, etc? Can you have periodic ones?

You can’t really do live editing and testing right? It’s just upload a new program? How does it handle that? Do you swap everything out at the end of a scan?

Also, what does sr mean? Not seen that one before.

Many questions there.

It will look foreign. Though for AI it will be more at home. The structure is usually similar, so I guess some users could adapt. Its not for everyone. I tried to make it simple, but its still Elixir.

The think I like is that its text programming, not tied into some graphical system generating XML files. Sometimes you can get eg ST code in PLC software, but its not a complete program that you could put into git and run CI easily. So that part is a lot easier with Xignal

Hmmm. I dont think many problems could be expressed as a table. So I dont really know what you mean here. But Xignal features a nice state machine. For me its a lot easier than eg. dealing with Siemens software.

It can show “online view”, a web page containing the code, then the live values next to each statement. I think this should be almost on par with many PLC software I have used. Though very few bells and whistles at this point. You can see IO and state machine states etc.

Yes. It uses Elixir testing and then there is a test runner that does not use wall time. I think this is a huge improvement to PLS software I have seen, and also for AI it can iterate better with testing.

You could have any structure you want. Each “PLC” can be a genserver. Run many in one process or just separate them on OS level. the “store” allows easy transfer of values between them over erlang messaging.

Hot code reload works well and was foundational to the design. So you can reload the code and timers, integrals, etc run uninterrupted. There is also retentive values for storing between restarts.

If the new code doesn’t match the old code or doesnt run in a sandbox, the hot code reload is blocked.

Its a different way then traditional PLC, but the functionality should let you work pretty easily with running systems.

SR is the set reset flip flop. Basically a boolean state holder. Xignal takes care of finding a memory location for the SR so that if you add a second SR, it wont get mixed up.

I’m super excited about this I think its a novel thing both for Elixir and for the PLC world…

Contact me for more access… I dont want to go public just yet.

Eg today I was working on realtime functionality using the nanosleep package on hex.pm. Real time performance of Erlang on RT Linux is quite decent, on my Beaglebone Blue I get the following results. Short story 500 Hz scan time feasible on beaglebone for a small system, with little jitter:

  period error and scan time
  runner         schedule  interval  timer      ticks  p50    p90     p99     p99.9   max     scan p50  scan p99  scan max  overruns  devices
  :benchmark_1   hard      10ms      nanosleep  5999   6 µs   16 µs   227 µs  321 µs  491 µs  712 µs    1025 µs   1913 µs   0         :benchmark_plant
  :benchmark_10  hard      10ms      nanosleep  5999   16 µs  299 µs  370 µs  775 µs  812 µs  3312 µs   4826 µs   6346 µs   0         :benchmark_plant

  :benchmark_1  garbage collection: 1721 minor, 6 major, 315.5 ms in all, 0.53% of the time, 5 KB freed per scan; longest 1.2 ms, minor, heap 144 KB -> 166 KB
  :benchmark_10  garbage collection: 5620 minor, 84 major, 1735.3 ms in all, 2.89% of the time, 39 KB freed per scan; longest 2.8 ms, minor, heap 289 KB -> 92 KB

  :benchmark_10: scans of 5 ms or more, or that far off the grid
    +34.957s  scan 6.3 ms  period 10.0 ms
        2.5 ms  gc major, heap 537 KB -> 89 KB
        0.1 ms  gc minor, heap 537 KB -> 537 KB
        3.5 ms  #9 Xignal.Jitter.Benchmark.tank
        1.6 ms  write store :benchmark_plant
        0.3 ms  in the runner itself, between the steps
        0.2 ms  #0 Xignal.Jitter.Benchmark.tank
    ...
    10 more, not shown

Hey @tallakt,

first thank you for sharing your ideas with the community. I am currently working with Siemens TIA Portal and the software is really frustrating. So I’m glad to see people in this community who are also struggling with it and looking for better solutions

We are not using OPC-UA at work, so I can’t help in that regard.
But I wrote a Nerves System for the EC100 and would like to use it to control my heat pump based on energy production of the solar system at my home. It’s a really simple project, but maybe a good start to test out the Xignal library.
Also I’m waiting for the EM300 (which is a PLC based on linux and very similar to the EC100) to arrive and maybe Xignal can be extended to directly write/read from the IO cards.

This looks very promising

I would be happy to write that with you if you want to try…

I realize nerves may be a good option as well, even considering it is much more mature. But Xignal is pretty darn cool too.

It would also be a chance to compare Nerves vs Xignal…

The only thing that would worry me a bit, if the heater is mostly protocol work its not a good match maybe

Hmm I thought Xignal is a library which just runs on the beam? Then nerves would be just the “operating system” which delivers the basis and Xignal would be the application code.

There is no Nerves in Xignal. The OS is Linux, with the Erlang virtual machine running the bytecode (BEAM).

The only common package I think is circuits for serial port communication. To compare the code bases, in Xignal (repeated):

# A sump pump station: a level transmitter and three contacts on a Modbus TCP I/O module, a pump tha
t
# starts at a high level and stops at a low one, three alarms, and an OPC UA server for the HMI with
 the
# alarms as conditions.
#
#     elixir pump_station.exs
#
# then browse opc.tcp://localhost:4840 with an OPC UA client such as UaExpert.
Mix.install([{:xignal, github: "tallakt/xignal"}, {:yaopcua, "~> 0.2"}])

defmodule Station do
  use Xignal

  xignal pump(raw, high_switch, overload, running, auto, start_at, stop_at, ack, alarms) do
    level = raw / 100
    run = auto and not overload and sr(level > start_at, level < stop_at)
    {_q, starts} = ctu(run, false, 1_000_000)

    high = Xignal.Alarm.check(alarms.high, ton(high_switch or level > 95.0, ~t(5s)), ack)
    trip = Xignal.Alarm.check(alarms.overload, overload, ack)
    stuck = Xignal.Alarm.check(alarms.no_start, ton(run and not running, ~t(5s)), ack)

    [level: level, run: run, starts: starts, alarm: high[:on] or trip[:on] or stuck[:on]]
  end
end

io = Xignal.Modbus.new("127.0.0.1", port: 1502, scan: Xignal.t("100ms"))
level = Xignal.Modbus.input(io, input_register: 0, desc: "Level, 0.01 %")
high_switch = Xignal.Modbus.input(io, discrete_input: 0, desc: "High level switch")
overload = Xignal.Modbus.input(io, discrete_input: 1, desc: "Pump overload")
running = Xignal.Modbus.input(io, discrete_input: 2, desc: "Pump running")

alarms = Xignal.Alarm.new(:station)

points = %{
  high: Xignal.Alarm.define(alarms, :level_high, desc: "Sump level high", severity: :high),
  overload: Xignal.Alarm.define(alarms, :pump_overload, desc: "Pump overload", severity: :critical),
  no_start: Xignal.Alarm.define(alar"Pump did not start")
}

hmi = Xignal.OPCUA.Server.new(port: 4840, alarms: :station)
percent = [range: {0.0, 100.0}, unit
auto = Xignal.OPCUA.Server.input(hmi, "Station/Auto", default: true)
start_at = Xignal.OPCUA.Server.input, [default: 80.0] ++ percent)
stop_at = Xignal.OPCUA.Server.input(hmi, "Station/StopLevel", [default: 20.0] ++ percent)
ack = Xignal.OPCUA.Server.input(hmi,fault: false)

Station.pump(level, high_switch, ovet_at, stop_at, ack, points)
|> Xignal.Modbus.sink(:run, Xignal.Modbus.output(io, coil: 0, desc: "Pump run"))
|> Xignal.OPCUA.Server.sink(:level, (hmi, "Station/Level", [default: 0.0]
++ percent))
|> Xignal.OPCUA.Server.sink(:run, hmi, "Station/PumpRunning", default: fa
lse))
|> Xignal.OPCUA.Server.sink(:starts,t(hmi, "Station/PumpStarts", default:
0))
|> Xignal.OPCUA.Server.sink(:alarm, (hmi, "Station/Alarm", default: false)
)

Xignal.Modbus.print_io_list(io)
Xignal.Runner.run_forever([io, alarm)

An ported to Nerves by an ai chatbot:

# A sump pump station in plain Elixir, the way a Nerves application is written: a supervision tree
# with a Modbus client, an OPC UA server, and one GenServer that scans the logic every 100 ms.
#
# A level transmitter and three contacts are on a Modbus TCP I/O module. The pump starts at a high
# level and stops at a low one, there are three alarms, and the HMI gets an OPC UA server with the
# alarms as conditions.
#
#     elixir pump_station.exs
#
# then browse opc.tcp://localhost:4840 with an OPC UA client such as UaExpert.
Mix.install([{:yamodbus, "~> 0.1"}, {:yaopcua, "~> 0.3"}])

defmodule PumpStation do
  @moduledoc "The supervision tree: the I/O, the HMI, and the control that uses both."
  use Application

  @impl Application
  def start(_type, _args) do
    children = [
      {Modbus.Client, tcp: "127.0.0.1", port: 1502, name: PumpStation.IO},
      {OPCUA.Server, port: 4840, name: PumpStation.HMI},
      PumpStation.Control
    ]

    # The control holds what it has read from the two before it, so it restarts with either.
    Supervisor.start_link(children, strategy: :rest_for_one, name: PumpStation.Supervisor)
  end
end

defmodule PumpStation.Alarm do
  @moduledoc """
  One alarm, as a value: a trigger that must hold for `delay` ms, and the four states of ISA 18.2.

      normal -> active_unacked -> active_acked -> normal          acknowledged while it stands
      normal -> active_unacked -> rtn_unacked  -> normal          cleared before anyone looked
  """

  defstruct [:key, :text, :severity, delay: 0, held: 0, state: :normal]

  @doc "The alarm after a scan `dt` ms long, in which its condition was `condition` and `ack` asked."
  def step(%__MODULE__{} = alarm, condition, ack, dt) do
    held = if condition, do: alarm.held + dt, else: 0
    trigger = condition and held >= alarm.delay
    %{alarm | held: held, state: next(alarm.state, trigger, ack)}
  end

  defp next(:normal, true, _ack), do: :active_unacked
  defp next(:active_unacked, true, true), do: :active_acked
  defp next(:active_unacked, false, true), do: :normal
  defp next(:active_unacked, false, false), do: :rtn_unacked
  defp next(:active_acked, false, _ack), do: :normal
  defp next(:rtn_unacked, true, _ack), do: :active_unacked
  defp next(:rtn_unacked, false, true), do: :normal
  defp next(state, _trigger, _ack), do: state

  def on?(%__MODULE__{state: state}), do: state != :normal
  def active?(%__MODULE__{state: state}), do: state in [:active_unacked, :active_acked]
  def acked?(%__MODULE__{state: state}), do: state in [:normal, :active_acked]
end

defmodule PumpStation.HMI.Nodes do
  @moduledoc "The address space the HMI sees: settings it writes, values it reads, and the alarms."

  alias OPCUA.Server

  @percent [units: {"P1", "%"}, range: {0.0, 100.0}]

  @settings [
    auto: {"Auto", :boolean, true, []},
    start_at: {"StartLevel", :double, 80.0, @percent},
    stop_at: {"StopLevel", :double, 20.0, @percent},
    ack: {"Ack", :boolean, false, []}
  ]

  @values [
    level: {"Level", :double, 0.0, @percent},
    run: {"PumpRunning", :boolean, false, []},
    starts: {"PumpStarts", :int32, 0, []},
    alarm: {"Alarm", :boolean, false, []}
  ]

  @doc "Adds the station to the server. `acknowledged` is told the key of an alarm a client acknowledges."
  def add(server, alarms, acknowledged) do
    ns = Server.namespace(server, "urn:pump-station")
    station = "ns=#{ns};s=Station"
    Server.add_object(server, station, "Station")

    for {table, writable} <- [{@settings, true}, {@values, false}],
        {_key, {name, type, value, opts}} <- table do
      opts = [parent: station, type: type, value: value, writable: writable] ++ opts
      Server.add_variable(server, "#{station}.#{name}", name, within_range(opts))
    end

    for alarm <- alarms do
      Server.add_condition(server, "#{station}.#{alarm.key}", "#{alarm.key}",
        source: station,
        message: alarm.text,
        severity: alarm.severity,
        acknowledge: fn _comment -> acknowledged.(alarm.key) end
      )
    end

    station
  end

  # A setting with a range refuses a value outside it.
  defp within_range(opts) do
    case {opts[:writable], opts[:range]} do
      {true, {low, high}} ->
        check = fn value -> if value >= low and value <= high, do: :ok, else: {:error, :bad_out_of_range} end
        Keyword.put(opts, :write, check)

      _ ->
        opts
    end
  end

  @doc "What the HMI has set, as a map."
  def settings(space, station) do
    Map.new(@settings, fn {key, {name, _type, _value, _opts}} ->
      {key, Server.get(space, "#{station}.#{name}").value.value}
    end)
  end

  @doc "Shows the HMI the values that differ from those last shown, and returns what is shown now."
  def show(space, station, shown, values) do
    for {key, {name, _type, _value, _opts}} <- @values, values[key] != shown[key] do
      Server.set(space, "#{station}.#{name}", values[key])
    end

    Map.take(values, Keyword.keys(@values))
  end
end

defmodule PumpStation.Control do
  @moduledoc """
  The scan: every 100 ms it asks the I/O module for its inputs, works out the pump and the alarms
  from what it last heard and what the HMI has set, and writes the pump and shows the HMI.

  The Modbus requests are sent without waiting, and their answers come as messages, so a slow or
  missing module never holds up a scan: the logic goes on with the last values it read.
  """
  use GenServer

  alias PumpStation.{Alarm, HMI, IO}
  alias PumpStation.HMI.Nodes

  @scan 100
  @unit 1

  defstruct [
    :station,
    :space,
    :due,
    :last,
    # What was last read from the I/O module, and the requests not yet answered.
    inputs: %{raw: 0, high_switch: false, overload: false, running: false},
    waiting: %{},
    # What the logic remembers from one scan to the next.
    wanted: false,
    run: false,
    starts: 0,
    acked: MapSet.new(),
    shown: %{},
    alarms: [
      %Alarm{key: :level_high, text: "Sump level high", severity: 700, delay: 5_000},
      %Alarm{key: :pump_overload, text: "Pump overload", severity: 900},
      %Alarm{key: :pump_no_start, text: "Pump did not start", severity: 700, delay: 5_000}
    ]
  ]

  def start_link(_opts), do: GenServer.start_link(__MODULE__, nil, name: __MODULE__)

  @impl GenServer
  def init(nil) do
    s = %__MODULE__{}
    control = self()
    station = Nodes.add(HMI, s.alarms, fn key -> send(control, {:acknowledged, key}) && :ok end)
    now = System.monotonic_time(:millisecond)
    send(self(), :scan)
    {:ok, %{s | station: station, space: OPCUA.Server.space(HMI), due: now, last: now}}
  end

  @impl GenServer
  def handle_info(:scan, s) do
    now = System.monotonic_time(:millisecond)

    s =
      %{s | last: now}
      |> control(Nodes.settings(s.space, s.station), now - s.last)
      |> write()
      |> read()

    # On a grid, so that a long scan doesn't push the next ones back.
    due = max(s.due + @scan, now)
    Process.send_after(self(), :scan, due, abs: true)
    {:noreply, %{s | due: due}}
  end

  # An answer from the I/O module: what was read goes into the inputs; an error leaves them as they are.
  def handle_info({Modbus.Client, ref, result}, s) do
    {what, waiting} = Map.pop(s.waiting, ref)

    inputs =
      case {what, result} do
        {:contacts, {:ok, [high_switch, overload, running]}} ->
          %{s.inputs | high_switch: high_switch, overload: overload, running: running}

        {:level, {:ok, [raw]}} ->
          %{s.inputs | raw: raw}

        _written_or_failed ->
          s.inputs
      end

    {:noreply, %{s | inputs: inputs, waiting: waiting}}
  end

  # A client acknowledged an alarm through the server; the next scan takes it.
  def handle_info({:acknowledged, key}, s), do: {:noreply, %{s | acked: MapSet.put(s.acked, key)}}

  ## The logic

  defp control(s, hmi, dt) do
    %{raw: raw, high_switch: high_switch, overload: overload, running: running} = s.inputs
    level = raw / 100

    # A latch, set above the start level and reset below the stop level; set wins.
    wanted = level > hmi.start_at or (s.wanted and not (level < hmi.stop_at))
    run = hmi.auto and not overload and wanted
    starts = if run and not s.run, do: s.starts + 1, else: s.starts

    conditions = %{
      level_high: high_switch or level > 95.0,
      pump_overload: overload,
      pump_no_start: run and not running
    }

    alarms =
      for alarm <- s.alarms do
        ack = hmi.ack or MapSet.member?(s.acked, alarm.key)
        alarm |> Alarm.step(conditions[alarm.key], ack, dt) |> tell(alarm, s)
      end

    values = %{level: level, run: run, starts: starts, alarm: Enum.any?(alarms, &Alarm.on?/1)}
    shown = Nodes.show(s.space, s.station, s.shown, values)
    %{s | wanted: wanted, run: run, starts: starts, alarms: alarms, acked: MapSet.new(), shown: shown}
  end

  # The server's condition follows the alarm, when it changed.
  defp tell(%Alarm{state: state} = alarm, %Alarm{state: state}, _s), do: alarm

  defp tell(alarm, _before, s) do
    OPCUA.Server.condition(HMI, "#{s.station}.#{alarm.key}",
      active: Alarm.active?(alarm),
      acked: Alarm.acked?(alarm)
    )

    alarm
  end

  ## The I/O module

  defp write(s), do: ask(s, :pump, {:write_single_coil, 0, s.run})

  defp read(s) do
    s
    |> ask(:contacts, {:read_discrete_inputs, 0, 3})
    |> ask(:level, {:read_input_registers, 0, 1})
  end

  # One request of a kind at a time: an answer that is late is waited for, not asked for again.
  defp ask(s, what, request) do
    if what in Map.values(s.waiting) do
      s
    else
      ref = Modbus.Client.send_request(IO, @unit, request)
      %{s | waiting: Map.put(s.waiting, ref, what)}
    end
  end
end

{:ok, _supervisor} = PumpStation.start(:normal, [])
Process.sleep(:infinity)

I guess you could use Nerves to install a Xignal project on eg raspberry pi… I didnt look into that

I think we are talking past each other. Nerves is just an OS based on Linux which directly boots into the BEAM. So you can add Xignal to your dependencies, write your Station module in lib and flash it to a RPi or similar. Would love to try that out if you could give me access to the library

I think perhaps you are right. Nerves + Xignal can be a good match.

I was thinking about how Nerves can interface IO on the board pins. This is maybe something I would like to integrate into Xignal at one point, so far it didnt get priority. I probably should also review this in more detail to get an overview.

Also sorry for the AI generated code I posted without reading it through. It seems my AI had done a very thorough interpretation of the Xignal code which is probably not what someone who did not use Xignal would have done. They would have made a more barebones implementation without all the bells and whistles of Xignal (eg. the alarm xignal is comprehensive). So I asked it to make a new code saying: “what would the above Xignal code look like for someone not using Xignal”, then a few rounds of simplifications, leaving out OPC-UA and then getting this (which can probably still be improved upon, but is ok as an illustration):

added around it:

Mix.install([{:modbuzz, "~> 0.3"}])

defmodule Station do
  use GenServer
  require Logger
  alias Modbuzz.PDU.{ReadDiscreteInputs, ReadInputRegisters, WriteSingleCoil}

  @scan_ms 100
  @alarm_delay_ms 5_000
  @io :station_io
  @unit 1

  # Everything the station remembers from one scan to the next.
  defstruct auto: true,
            start_at: 80.0,
            stop_at: 20.0,
            ack: false,
            level: 0.0,
            latch: false,
            run: false,
            starts: 0,
            high_ms: 0,
            stuck_ms: 0,
            alarms: %{level_high: false, pump_overload: false, pump_no_start: false},
            io_ok: true,
            scanned_at: nil

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  def status, do: GenServer.call(__MODULE__, :status)
  def ack, do: GenServer.call(__MODULE__, {:set, :ack, true})

  def set(key, value) when key in [:auto, :start_at, :stop_at],
    do: GenServer.call(__MODULE__, {:set, key, value})

  # The logic, with no I/O in it: the state, this scan's inputs and the time since the last scan.
  def step(%__MODULE__{} = s, inputs, dt) do
    %{raw: raw, high_switch: high_switch, overload: overload, running: running} = inputs
    level = raw / 100

    # Start above start_at, stop below stop_at, hold in between.
    latch = level > s.start_at or (s.latch and not (level < s.stop_at))
    run = s.auto and not overload and latch
    starts = if run and not s.run, do: s.starts + 1, else: s.starts

    # On-delays: count up while the condition holds, start over when it drops.
    high_ms = if high_switch or level > 95.0, do: s.high_ms + dt, else: 0
    stuck_ms = if run and not running, do: s.stuck_ms + dt, else: 0

    triggers = %{
      level_high: high_ms >= @alarm_delay_ms,
      pump_overload: overload,
      pump_no_start: stuck_ms >= @alarm_delay_ms
    }

    # An alarm is set by its condition and stays until someone acknowledges it.
    alarms = Map.new(triggers, fn {key, on} -> {key, on or (s.alarms[key] and not s.ack)} end)

    %{
      s
      | level: level,
        latch: latch,
        run: run,
        starts: starts,
        high_ms: high_ms,
        stuck_ms: stuck_ms,
        alarms: alarms,
        ack: false,
        io_ok: true
    }
  end

  @impl true
  def init(opts) do
    host = Keyword.fetch!(opts, :host)
    port = Keyword.get(opts, :port, 502)

    # The client lives under Modbuzz's supervisor, so it is still there when this process restarts.
    with {:error, :already_started} <- Modbuzz.start_tcp_client(@io, host, port), do: :ok

    send(self(), :scan)
    {:ok, %__MODULE__{scanned_at: System.monotonic_time(:millisecond)}}
  end

  @impl true
  def handle_info(:scan, s) do
    now = System.monotonic_time(:millisecond)

    next =
      case read() do
        {:ok, inputs} -> step(s, inputs, now - s.scanned_at)
        {:error, reason} -> io_lost(s, reason)
      end

    write(next.run)

    # An absolute deadline, so that the time a scan takes does not add to the period.
    Process.send_after(self(), :scan, now + @scan_ms, abs: true)
    {:noreply, %{next | scanned_at: now}}
  end

  @impl true
  def handle_call(:status, _from, s) do
    alarm = Enum.any?(Map.values(s.alarms))
    {:reply, Map.put(Map.take(s, [:level, :run, :starts, :alarms, :io_ok]), :alarm, alarm), s}
  end

  def handle_call({:set, key, value}, _from, s), do: {:reply, :ok, Map.replace!(s, key, value)}

  defp read do
    with {:ok, %{input_registers: [raw]}} <-
           request(%ReadInputRegisters.Req{starting_address: 0, quantity_of_input_registers: 1}),
         {:ok, %{input_status: [high_switch, overload, running | _]}} <-
           request(%ReadDiscreteInputs.Req{starting_address: 0, quantity_of_inputs: 3}) do
      {:ok, %{raw: raw, high_switch: high_switch, overload: overload, running: running}}
    end
  end

  defp write(run), do: request(%WriteSingleCoil.Req{output_address: 0, output_value: run})

  defp request(req), do: Modbuzz.request(@io, @unit, req, 500)

  # Without inputs there is nothing to decide on: stop the pump and keep the rest as it was.
  defp io_lost(s, reason) do
    if s.io_ok, do: Logger.error("I/O module lost, stopping the pump: #{inspect(reason)}")
    %{s | run: false, io_ok: false}
  end
end

{:ok, _} = Station.start_link(host: {127, 0, 0, 1}, port: 1502)
Process.sleep(:infinity)

Ok I see. So Xignal is basically reimplementing the tick based approach from traditional plc IEC 61131-3. Any reason to rely on that? I understand that protocols like Modbus need to poll, but from there wouldn’t it make more sense to use Elixirs event driven approach? There is also a newer norm (IEC 61499) which is closer to the BEAM event model, have you looked into that?