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
to those PLCs). Live edits are essential in my eyes.
