Skip to content

Alarm handling with AMSG #52

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pheest wants to merge 8 commits into
base: master
Choose a base branch
from
Open

Alarm handling with AMSG #52

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
f70f1e6
Copied from old fork.
pheest Jul 30, 2026
1b2fec0
Accepted changes to ci-scripts-build.yml made by Windows support pull…
pheest Aug 12, 2026
1dec995
Merge remote-tracking branch 'community/master' into Alarm_handling_w…
tynanford Aug 18, 2026
e4de9a8
Correct mistakes moted by tynanford.
pheest Aug 19, 2026
1f47d20
Fix more than one instance
pheest Aug 20, 2026
ba69d4c
Correct possible editing mistake
pheest Aug 20, 2026
e8220b1
Set (local copy of) stat to COMM_ALARM if self.stat has not been set.
pheest Aug 27, 2026
cbfdca7
Reversed the the order of @onproc and stat variable assignment.
pheest Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 31 additions & 27 deletions devsupApp/src/devsup/ptable.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
_tables = {}

from .db import IOScanListThread
from . import INVALID_ALARM, UDF_ALARM
from . import INVALID_ALARM, UDF_ALARM, COMM_ALARM

__all__ = [
'Parameter',
Expand Down Expand Up @@ -159,20 +159,16 @@ class _ParamInstance(object):
"""
def __init__(self, table, name, scan):
self.name = name
self.table, self.scan, self._value = table, scan, None
self.table, self.scan, self.value = table, scan, None
self.alarm, self.actions = 0, []
self.stat = None
self.amsg = None
self._groups = set()
def _get_value(self):
return self._value
def _set_value(self, val):
self._value = val
self.alarm = 3 if val is None else 0
value = property(_get_value, _set_value, doc="The current parameter value")
@property
def isvalid(self):
"""Is the parameter value valid (not None and no INVALID_ALARM)
"""
return self.alarm < INVALID_ALARM and self._value is not None
return self.alarm < INVALID_ALARM and self.value is not None
def notify(self):
"""Notify attached records of parameter value change.
A no-op unless Parameter(iointr=True)
Expand Down Expand Up @@ -215,12 +211,16 @@ class _ParamSupBase(object):
def __init__(self, inst, rec, info):
self.inst, self.info = inst, info
# Determine which field to use to store the value
fname = rec.info('pyfield','VAL')
self.raw = fname!='RVAL'
self.vfld = rec.field(fname)
self.raw = False
try:
self.vfld = rec.field("RVAL")
self.raw = True
except KeyError:
self.vfld = rec.field("VAL")
self.vdata = None
if len(self.vfld)>1:
self.vdata = self.vfld.getarray()

def detach(self, rec):
pass
def allowScan(self, rec):
Expand All @@ -232,19 +232,20 @@ def process(self, rec, reason=None):
"""Read a value from the table into the record
"""
with self.inst.table.lock:
nval, alrm = self.inst.value, self.inst.alarm
self.inst.table.log.debug('%s -> %s (%s)', self.inst.name, rec.NAME, nval)
value, alarm, stat, amsg = self.inst.value, self.inst.alarm, self.inst.stat, self.inst.amsg
self.inst.table.log.debug('%s -> %s (%s)', self.inst.name, rec.NAME, value)
if stat is None:
stat = COMM_ALARM

if nval is not None:
if value is not None:
if self.vdata is None:
self.vfld.putval(nval)
self.vfld.putval(value)
else:
if len(nval)>len(self.vdata):
nval = nval[:len(self.vdata)]
self.vdata[:len(nval)] = nval
self.vfld.putarraylen(len(nval))
if alrm:
rec.setSevr(alrm)
if len(value)>len(self.vdata):
value = value[:len(self.vdata)]
self.vdata[:len(value)] = value
self.vfld.putarraylen(len(value))
rec.setSevr(alarm, stat, amsg)
else:
# undefined value
rec.setSevr(INVALID_ALARM, UDF_ALARM)
Expand All @@ -261,19 +262,22 @@ def process(self, rec, reason=None):
# sync record to table
self.inst.table.log.debug('%s <- %s (%s)', self.inst.name, rec.NAME, rec.VAL)
if self.vdata is None:
nval = self.vfld.getval()
value = self.vfld.getval()
else:
# A copy is made which can be used without locking the record
nval = self.vdata[:self.vfld.getarraylen()].copy()
value = self.vdata[:self.vfld.getarraylen()].copy()

with self.inst.table.lock:
oval, self.inst.value = self.inst.value, nval

oval, self.inst.value = self.inst.value, value
# Execute actions
self.inst._exec(oval)
stat = self.inst.stat
if stat is None:
stat = COMM_ALARM
rec.setSevr(self.inst.alarm, stat, self.inst.amsg)
for G in self.inst._groups:
G._exec()

class TableBase(object):
"""Base class for all parameter tables.

Expand Down
63 changes: 63 additions & 0 deletions devsupApp/src/devsup/test/test_db.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,69 @@ def test_increment(self):
self.assertEqual(rec.VAL, 1)
self.assertEqual(rec.UDF, 0)

class TestAlarmScan(IOCHelper):
db = """
record(longin, "rec:inalarm:amsg") {
field(PINI, "YES")
}
record(longin, "rec:inalarm:plain") {
field(PINI, "YES")
}
record(longout, "rec:outalarm:amsg") {
field(PINI, "YES")
field(VAL, "0")
}
record(longout, "rec:outalarm:plain") {
field(PINI, "YES")
field(VAL, "0")
}
"""
def test_setin_severity_message(self):
rec = getRecord("rec:inalarm:amsg")
with rec:
self.assertEqual(rec.SEVR, _dbapi.NO_ALARM)
self.assertEqual(rec.STAT, _dbapi.NO_ALARM)
if _dbapi.epicsver[:4] >= (7, 0, 6, 0):
self.assertEqual(rec.AMSG, "")
self.assertEqual(rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.HIHI_ALARM, amsg="Meaningful input alarm message"), None)
self.assertEqual(rec.scan(sync=True), 0)
self.assertEqual(rec.STAT, _dbapi.HIHI_ALARM)
if _dbapi.epicsver[:4] >= (7, 0, 6, 0):
self.assertEqual(rec.AMSG, "Meaningful input alarm message")

def test_setin_severity_without_message(self):
rec = getRecord("rec:inalarm:plain")
with rec:
self.assertEqual(rec.SEVR, _dbapi.NO_ALARM)
self.assertEqual(rec.STAT, _dbapi.NO_ALARM)
self.assertEqual(rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.COMM_ALARM), None)
self.assertEqual(rec.scan(sync=True), 0)
self.assertEqual(rec.SEVR, _dbapi.MAJOR_ALARM)
self.assertEqual(rec.STAT, _dbapi.COMM_ALARM)

def test_setout_severity_message(self):
rec = getRecord("rec:outalarm:amsg")
with rec:
self.assertEqual(rec.SEVR, _dbapi.NO_ALARM)
self.assertEqual(rec.STAT, _dbapi.NO_ALARM)
if _dbapi.epicsver[:4] >= (7, 0, 6, 0):
self.assertEqual(rec.AMSG, "")
self.assertEqual(rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.HIHI_ALARM, amsg="Meaningful output alarm message"), None)
self.assertEqual(rec.scan(sync=True), 0)
self.assertEqual(rec.STAT, _dbapi.HIHI_ALARM)
if _dbapi.epicsver[:4] >= (7, 0, 6, 0):
self.assertEqual(rec.AMSG, "Meaningful output alarm message")

def test_setout_severity_without_message(self):
rec = getRecord("rec:outalarm:plain")
with rec:
self.assertEqual(rec.SEVR, _dbapi.NO_ALARM)
self.assertEqual(rec.STAT, _dbapi.NO_ALARM)
self.assertEqual(rec.setSevr(_dbapi.MAJOR_ALARM, _dbapi.COMM_ALARM), None)
self.assertEqual(rec.scan(sync=True), 0)
self.assertEqual(rec.SEVR, _dbapi.MAJOR_ALARM)
self.assertEqual(rec.STAT, _dbapi.COMM_ALARM)

class TestLongStringField(IOCHelper):
db = """
record(lsi, "rec:lsi") {
Expand Down
Loading