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 6 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
6 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
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
35 changes: 21 additions & 14 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, NO_ALARM

__all__ = [
'Parameter',
Expand Down Expand Up @@ -161,6 +161,8 @@ def __init__(self, table, name, scan):
self.name = name
self.table, self.scan, self._value = table, scan, None
self.alarm, self.actions = 0, []
self.stat = UDF_ALARM
self.amsg = None
self._groups = set()
def _get_value(self):
return self._value
Expand Down Expand Up @@ -221,6 +223,10 @@ def __init__(self, inst, rec, info):
self.vdata = None
if len(self.vfld)>1:
self.vdata = self.vfld.getarray()
if rec.PINI == "YES":
# record processing will have updated the field.
self.inst.stat = NO_ALARM

def detach(self, rec):
pass
def allowScan(self, rec):
Expand All @@ -232,19 +238,19 @@ 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 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))
if alarm:
rec.setSevr(alarm, stat, amsg)
else:
# undefined value
rec.setSevr(INVALID_ALARM, UDF_ALARM)
Expand All @@ -261,16 +267,17 @@ 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)
rec.setSevr(self.inst.alarm, self.inst.stat, self.inst.amsg)
for G in self.inst._groups:
G._exec()

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