diff --git a/devsupApp/src/devsup/ptable.py b/devsupApp/src/devsup/ptable.py index 365cb91..db3adc6 100644 --- a/devsupApp/src/devsup/ptable.py +++ b/devsupApp/src/devsup/ptable.py @@ -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', @@ -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 @@ -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): @@ -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) @@ -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() diff --git a/devsupApp/src/devsup/test/test_db.py b/devsupApp/src/devsup/test/test_db.py index 07d0fbd..e31ecb0 100644 --- a/devsupApp/src/devsup/test/test_db.py +++ b/devsupApp/src/devsup/test/test_db.py @@ -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") {