36 fields determine sharednt handler initialisation

examples/custom_rule/README.md

@@ -0,0 +1,19 @@
+# Intro
+This folder shows two different ways a developer may implement a custom `Rule`. 
+
+## Custom Rule - Imatch
+The imatch Rule is a simple rule that triggers an alarm of MAJOR severity when the value of an NTScalar PV matches a specified value. Although this rule can apply for any type, as the name suggests it makes most sense in the case of an integer. We do not specify what should happen in the case of an NTScalarArray.
+
+## Implementations
+### Public
+The NTScalar is a Normative Type and its fields are defined by the specification of that type. However, in practice tools are permissive and will understand a type which is a superset of a Normative Type, ignoring those fields they do not expect or understand. As such it is possible to add additional fields to an NTScalar. 
+
+In this case we add an `imatch` field which is a structure with a boolean `active` and an integer `imatch` field. If the `imatch` is active then the value of `imatch.imatch` is tested against the PV's value. If equal then a the severity is set to MAJOR otherwise it is cleared. 
+
+The advantage of this method is that the imatch field is publicly available and may be manipulated through the standard EPICS pvAccess tools. This includes through the standard interfaces that are a part of p4p and p4pillon.
+
+## Hidden 
+An alternative way to implement the rule without modifying the Normative Type. The value of the imatch variable is held in the handler / Rule instead of the p4p.Value. This has the advantage that the Normative Type is unmodified. It has the possible disadvantage that the imatch value cannot be easily manipuluated by external pvAccess tools (hence describing it as hidden). As the code illustrates, it also makes it harder to reliably trigger the rule in some circumstances.
+
+## Notes
+This is a simplified example and does not consider some important cases. For example, what behaviour is correct in the case that the severity indicates invalidAlarm?

examples/custom_rule/hidden/imatch_alarm.py

@@ -0,0 +1,83 @@
+"""
+Demonstration of implementing a custom Rule
+"""
+
+import logging
+
+from p4p import Value
+from p4p.server import Server
+
+from p4pillon.nt import NTScalar
+from p4pillon.rules import (
+    AlarmRule,
+    BaseRule,
+    RulesFlow,
+    TimestampRule,
+)
+from p4pillon.rules.rules import check_applicable_init
+from p4pillon.thread.sharednt import SharedNT
+
+logger = logging.getLogger(__name__)
+
+
+class IMatchRule(BaseRule):
+    name = "imatch"
+    fields = ["alarm"]
+
+    def __init__(self, imatch: int | None = None):
+        self._imatch = imatch
+
+    @property
+    def imatch(self):
+        return self._imatch
+
+    @imatch.setter
+    def imatch(self, newval: int | None):
+        self._imatch = newval
+
+    @check_applicable_init
+    def init_rule(self, newpvstate: Value) -> RulesFlow:
+        # Check if imatch alarms are present and active!
+        if not self._imatch:
+            logger.debug("imatch not active")
+            return RulesFlow.CONTINUE
+
+        if self._imatch == newpvstate["value"]:
+            newpvstate["alarm.severity"] = 2
+            newpvstate["alarm.message"] = "IMATCH"
+        else:
+            newpvstate["alarm.severity"] = 0
+            newpvstate["alarm.message"] = ""
+
+        return RulesFlow.CONTINUE
+
+
+def main():
+    registered_handlers = [
+        AlarmRule,
+        IMatchRule,  # <-- This is the new rule
+        TimestampRule,
+    ]
+
+    pv1 = SharedNT(nt=NTScalar("i"), initial=5, imatch={"imatch": 5}, registered_handlers=registered_handlers)
+
+    pv2 = SharedNT(nt=NTScalar("i"), initial=5, imatch={"imatch": 3}, registered_handlers=registered_handlers)
+    pv2.handler["imatch"].rule.imatch = 5
+
+    pv3 = SharedNT(nt=NTScalar("i"), initial=5, imatch={"imatch": 3}, registered_handlers=registered_handlers)
+    pv3.handler["imatch"].rule.imatch = 5
+    pv3.post(5)
+
+    Server.forever(
+        providers=[
+            {
+                "demo:pv:name1": pv1,
+                "demo:pv:name2": pv2,
+                "demo:pv:name3": pv3,
+            }
+        ]
+    )  # runs until KeyboardInterrupt
+
+
+if __name__ == "__main__":
+    main()

examples/custom_rule/public/imatch_alarm.py

@@ -0,0 +1,105 @@
+"""
+Demonstration of implementing a custom Rule
+"""
+
+import logging
+
+import p4p.nt.common as nt_common
+from p4p import Type, Value
+from p4p.nt.scalar import _metaHelper
+from p4p.server import Server
+
+from p4pillon.nt import NTScalar
+from p4pillon.rules import (
+    AlarmRule,
+    BaseRule,
+    ControlRule,
+    RulesFlow,
+    TimestampRule,
+    ValueAlarmRule,
+)
+from p4pillon.rules.rules import check_applicable_init
+from p4pillon.thread.sharednt import SharedNT
+
+logger = logging.getLogger(__name__)
+
+
+class NTScalarMatch(NTScalar):
+    """
+    Extend NTScalar with imatch fields
+    """
+
+    @staticmethod
+    def buildType(valtype, extra=[], *args, **kws):
+        """Build a Type
+
+        :param str valtype: A type code to be used with the 'value' field.  See :ref:`valuecodes`
+        :param list extra: A list of tuples describing additional non-standard fields
+        :param bool display: Include optional fields for display meta-data
+        :param bool control: Include optional fields for control meta-data
+        :param bool valueAlarm: Include optional fields for alarm level meta-data
+        :param bool form: Include ``display.form`` instead of the deprecated ``display.format``.
+        :returns: A :py:class:`Type`
+        """
+        isarray = valtype[:1] == "a"
+        fields = [
+            ("value", valtype),
+            ("alarm", nt_common.alarm),
+            ("timeStamp", nt_common.timeStamp),
+            ("imatch", Type(id="imatch_t", spec=[("active", "?"), ("imatch", "i")])),
+        ]
+        _metaHelper(fields, valtype, *args, **kws)
+        fields.extend(extra)
+        return Type(id="epics:nt/NTScalarArray:1.0" if isarray else "epics:nt/NTScalar:1.0", spec=fields)
+
+    def __init__(self, valtype="d", **kws):
+        self.type = self.buildType(valtype, **kws)
+
+
+class IMatchRule(BaseRule):
+    name = "imatch"
+    fields = ["alarm", "imatch"]
+
+    @check_applicable_init
+    def init_rule(self, newpvstate: Value) -> RulesFlow:
+        # Check if imatch alarms are present and active!
+        if not newpvstate["imatch.active"]:
+            # TODO: This is wrong! If valueAlarm was active and then made inactive
+            #       the alarm will not be cleared
+            logger.debug("imatch not active")
+            return RulesFlow.CONTINUE
+
+        if newpvstate["imatch.imatch"] == newpvstate["value"]:
+            newpvstate["alarm.severity"] = 2
+            newpvstate["alarm.message"] = "IMATCH"
+        else:
+            newpvstate["alarm.severity"] = 0
+            newpvstate["alarm.message"] = ""
+
+        return RulesFlow.CONTINUE
+
+
+# Register the new rule
+SharedNT.registered_handlers = [
+    AlarmRule,
+    ControlRule,
+    ValueAlarmRule,
+    IMatchRule,  # <-- This is the new rule
+    TimestampRule,
+]
+
+
+def main():
+    pv = SharedNT(nt=NTScalarMatch("i"), initial={"value": 5, "imatch.active": True, "imatch.imatch": 5})
+
+    Server.forever(
+        providers=[
+            {
+                "demo:pv:name": pv,  # PV name only appears here
+            }
+        ]
+    )  # runs until KeyboardInterrupt
+
+
+if __name__ == "__main__":
+    main()

examples/thread/mailbox_sharedntenum.py

@@ -11,12 +11,8 @@
         "value.choices": ["STOP", "START", "STANDBY"],
         "display.description": "Pump on/off control word.",
     },
-    handler_constructors={
-        "alarmNTEnum": {
-            "STOP": AlarmDict(
-                severity=AlarmSeverity.MAJOR_ALARM, status=AlarmStatus.NO_STATUS, message="Shouldn't be off"
-            )
-        }
+    alarmNTEnum={
+        "STOP": AlarmDict(severity=AlarmSeverity.MAJOR_ALARM, status=AlarmStatus.NO_STATUS, message="Shouldn't be off")
     },
 )  # setting initial value also open()'s
 

p4pillon/nt/identify.py

@@ -3,7 +3,7 @@
 """
 
 import re
-from enum import Enum, auto
+from enum import IntEnum, auto
 
 from p4p import Type, Value
 
@@ -17,7 +17,7 @@
 )
 
 
-class NTType(Enum):
+class NTType(IntEnum):
     """
     Normative Types in the Specification Document, excluding those in Appendix A
     """

p4pillon/rules/alarm_ntenum_rule.py

@@ -16,7 +16,7 @@ class AlarmNTEnumRule(AlarmRule):
     Uses a dictionary to map NTEnum values to severity, status, and message.
     """
 
-    name = "alarmntenum"
+    name = "alarmNTEnum"
     nttypes = [SupportedNTTypes.NTENUM]
     fields = ["alarm"]
 

p4pillon/rules/rules.py

@@ -19,6 +19,7 @@
 from p4p.server import ServerOperation
 from p4p.server.raw import ServOpWrap
 
+from p4pillon.nt.identify import NTType
 from p4pillon.utils import overwrite_marked
 
 logger = logging.getLogger(__name__)
@@ -27,11 +28,11 @@
 class SupportedNTTypes(IntEnum):
     """Supported Normative Types."""
 
-    NTSCALAR = auto()
-    NTSCALARARRAY = auto()
-    NTENUM = auto()
-    NTTABLE = auto()  # basic support only
-    NTNDARRAY = auto()  # basic support only
+    NTSCALAR = NTType.NTSCALAR
+    NTSCALARARRAY = NTType.NTSCALARARRAY
+    NTENUM = NTType.NTENUM
+    NTTABLE = NTType.NTTABLE  # basic support only
+    NTNDARRAY = NTType.NTNDARRAY  # basic support only
     ALL = auto()  # signal that this is essentially type independent
 
 

p4pillon/sharednt.py

@@ -12,7 +12,7 @@
 from p4p import Type, Value
 
 from p4pillon.composite_handler import CompositeHandler
-from p4pillon.nt.identify import is_scalararray
+from p4pillon.nt.identify import id_nttype, is_scalararray
 from p4pillon.nthandlers import ComposeableRulesHandler
 from p4pillon.rules import (
     AlarmNTEnumRule,
@@ -56,6 +56,7 @@ class SharedNT(SharedPV, ABC):
     registered_handlers: list[type[BaseRule]] = [
         AlarmRule,
         ControlRule,
+        AlarmNTEnumRule,
         ValueAlarmRule,
         TimestampRule,
         CalcRule,
@@ -66,9 +67,12 @@ def __init__(
         *,
         auth_handlers: OrderedDict[str, Handler] | None = None,
         user_handlers: OrderedDict[str, Handler] | None = None,
-        handler_constructors: dict[str, Any] | None = None,
+        registered_handlers: list[type[BaseRule]] | None = None,
         **kwargs,
     ):
+        if registered_handlers:
+            self.registered_handlers = registered_handlers
+
         # Create a CompositeHandler. If there is no user supplied handler, and this is not
         # an NT type then it won't do anything. Unfortunately, an empty CompositeHandler
         # will be discarded and won't be passed to the super().__init__
@@ -81,15 +85,10 @@ def __init__(
 
             if nttype:
                 for registered_handler in self.registered_handlers:
-                    name, component_handler = self.__setup_registered_rule(registered_handler, nttype, **kwargs)
+                    name, component_handler, kwargs = self.__setup_registered_rule(registered_handler, nttype, **kwargs)
                     if name and component_handler:
                         handler[name] = component_handler
 
-                if handler_constructors and "alarmNTEnum" in handler_constructors:
-                    handler["alarmNTEnum"] = ComposeableRulesHandler(
-                        AlarmNTEnumRule(handler_constructors["alarmNTEnum"])
-                    )
-
         if user_handlers:
             handler = handler | user_handlers
 
@@ -221,7 +220,7 @@ def on_close(self):
 
     def __setup_registered_rule(
         self, class_to_instantiate: type[BaseRule], nttype, **kwargs
-    ) -> tuple[str | None, ComposeableRulesHandler | None]:
+    ) -> tuple[str | None, ComposeableRulesHandler | None, dict[str, Any]]:
         """The existence of a single function that does everything suggests this is the wrong approach!"""
 
         # Examine the class member variables to determine how/whether to setup this Rule
@@ -233,19 +232,25 @@ def __setup_registered_rule(
 
         # If we're not relying on the rule to provide enough information to configure itself then
         if not auto_add and name not in kwargs:
-            return (name, None)
+            return (name, None, kwargs)
 
         # Perform tests on whether the rule is applicable to the nttype and/or the fields
         if supported_nttypes:
             if len(supported_nttypes) == 1 and supported_nttypes == [SupportedNTTypes.ALL]:
                 pass
             else:
-                raise NotImplementedError("We're not yet testing for NT type!")
+                matchfound = False
+                type_id = id_nttype(nttype)
+                for supported_nttype in supported_nttypes:
+                    if supported_nttype == type_id:
+                        matchfound = True
+                if not matchfound:
+                    return (name, None, kwargs)
 
         if required_fields:
             for required_field in required_fields:
                 if required_field not in nttype:
-                    return (name, None)
+                    return (name, None, kwargs)
 
         # See if there's an attempt to pass arguments to the constructor of this Rule
         args = {}
@@ -262,4 +267,4 @@ def __setup_registered_rule(
         else:
             composed_instance = ComposeableRulesHandler(instance)
 
-        return (name, composed_instance)
+        return (name, composed_instance, kwargs)

tests/unit/test_sharednt.py

@@ -121,9 +121,7 @@ def testntscalar_create_with_handlers(pvtype, expected_handlername):
 
 
 def testntenum_create():
-    testpv = SharedNT(
-        nt=NTEnum(), initial={"index": 0, "choices": ["OFF", "ON"]}, handler_constructors={"alarmNTEnum": {}}
-    )
+    testpv = SharedNT(nt=NTEnum(), initial={"index": 0, "choices": ["OFF", "ON"]}, alarmNTEnum={})
 
     assert set(testpv.handler.keys()) == set(["alarm", "alarmNTEnum", "timestamp"])
     assert list(testpv.handler.keys())[-1] == "timestamp"
@@ -133,9 +131,9 @@ def testntenum_create_with_handlers():
     testpv = SharedNT(
         nt=NTEnum(),
         initial={"index": 0, "choices": ["OFF", "ON"]},
+        alarmNTEnum={},
         auth_handlers=OrderedDict({"pre1": Handler(), "pre2": Handler()}),
         user_handlers=OrderedDict({"post1": Handler(), "post2": Handler()}),
-        handler_constructors={"alarmNTEnum": {}},
     )
 
     assert list(testpv.handler.keys()) == ["pre1", "pre2", "alarm", "alarmNTEnum", "post1", "post2", "timestamp"]