Add support for Rule restriction by NT Type, apply to alarmNTEnum rule

Author: Monarda

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__
@@ -85,11 +89,6 @@ def __init__(
                     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
 
@@ -240,7 +239,13 @@ def __setup_registered_rule(
             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:

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"]

tests/unit/thread/test_pvrecipe.py

@@ -370,7 +370,7 @@ def test_ntenum_create_pv(mock_time):
         PVTypes.ENUM,
         description="test enum",
         initial_value={"index": 0, "choices": ["OFF", "ON"]},
-        handler_constructors={"alarmNTEnum": {}},
+        alarmNTEnum={},
     )
 
     pv = recipe.create_pv("TEST:PV:ENUM")

tests/unit/thread/test_sharednt.py

@@ -27,9 +27,7 @@ def testntscalar_thread_create(pvtype, expected_handlername):
 
 
 def testntenum_thread_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 len(testpv.handler) == 3
     assert list(testpv.handler.keys()) == ["alarm", "alarmNTEnum", "timestamp"]