1+ """nidaqmx feature toggles."""
2+
3+ from __future__ import annotations
4+
5+ import functools
6+ import sys
7+ from decouple import AutoConfig , Undefined , undefined
8+ from enum import Enum
9+ from typing import TYPE_CHECKING , Callable , TypeVar
10+ from nidaqmx ._dotenvpath import get_dotenv_search_path
11+ from nidaqmx .errors import FeatureNotSupportedError
12+
13+ if TYPE_CHECKING :
14+ if sys .version_info >= (3 , 10 ):
15+ from typing import ParamSpec
16+ else :
17+ from typing_extensions import ParamSpec
18+
19+ if sys .version_info >= (3 , 11 ):
20+ from typing import Self
21+ else :
22+ from typing_extensions import Self
23+
24+ _P = ParamSpec ("_P" )
25+ _T = TypeVar ("_T" )
26+
27+ _PREFIX = "NIDAQMX"
28+
29+ _config = AutoConfig (str (get_dotenv_search_path ()))
30+
31+ if TYPE_CHECKING :
32+ # Work around decouple's lack of type hints.
33+ def _config (
34+ option : str ,
35+ default : _T | Undefined = undefined ,
36+ cast : Callable [[str ], _T ] | Undefined = undefined ,
37+ ) -> _T : ...
38+
39+ # Based on the recipe at https://docs.python.org/3/howto/enum.html
40+ class _OrderedEnum (Enum ):
41+ def __ge__ (self , other : Self ) -> bool :
42+ if self .__class__ is other .__class__ :
43+ return self .value >= other .value
44+ return NotImplemented
45+
46+ def __gt__ (self , other : Self ) -> bool :
47+ if self .__class__ is other .__class__ :
48+ return self .value > other .value
49+ return NotImplemented
50+
51+ def __le__ (self , other : Self ) -> bool :
52+ if self .__class__ is other .__class__ :
53+ return self .value <= other .value
54+ return NotImplemented
55+
56+ def __lt__ (self , other : Self ) -> bool :
57+ if self .__class__ is other .__class__ :
58+ return self .value < other .value
59+ return NotImplemented
60+
61+
62+ class CodeReadiness (_OrderedEnum ):
63+ """Indicates whether code is ready to be supported."""
64+
65+ RELEASE = 0
66+ NEXT_RELEASE = 1
67+ INCOMPLETE = 2
68+ PROTOTYPE = 3
69+
70+
71+ def _init_code_readiness_level () -> CodeReadiness :
72+ if _config (f"{ _PREFIX } , default = False , cast = bool ):
73+ return CodeReadiness .INCOMPLETE
74+ elif _config (f"{ _PREFIX } , default = False , cast = bool ):
75+ return CodeReadiness .NEXT_RELEASE
76+ else :
77+ return CodeReadiness .RELEASE
78+
79+
80+ # This is not public because `from _feature_toggles import CODE_READINESS_LEVEL`
81+ # is incompatible with the patching performed by the use_code_readiness mark.
82+ _CODE_READINESS_LEVEL = _init_code_readiness_level ()
83+
84+
85+ def get_code_readiness_level () -> CodeReadiness :
86+ """Get the current code readiness level.
87+
88+ You can override this in tests by specifying the ``use_code_readiness``
89+ mark.
90+ """
91+ return _CODE_READINESS_LEVEL
92+
93+
94+ class FeatureToggle :
95+ """A run-time feature toggle."""
96+
97+ name : str
98+ """The name of the feature."""
99+
100+ readiness : CodeReadiness
101+ """The code readiness at which this feature is enabled."""
102+
103+ def __init__ (self , name : str , readiness : CodeReadiness ) -> None :
104+ """Initialize the feature toggle."""
105+ assert name == name .upper ()
106+ self .name = name
107+ self .readiness = readiness
108+ self ._is_enabled_override = None
109+ # Only read the env var at initialization time.
110+ if _config (f"{ _PREFIX } { name } , default = False , cast = bool ):
111+ self ._is_enabled_override = True
112+
113+ @property
114+ def is_enabled (self ) -> bool :
115+ """Indicates whether the feature is currently enabled.
116+
117+ You can enable/disable features in tests by specifying the
118+ ``enable_feature_toggle`` or ``disable_feature_toggle`` marks.
119+ """
120+ if self ._is_enabled_override is not None :
121+ return self ._is_enabled_override
122+ return self .readiness <= get_code_readiness_level ()
123+
124+ def _raise_if_disabled (self ) -> None :
125+ if self .is_enabled :
126+ return
127+
128+ env_vars = f"{ _PREFIX } { self .name }
129+ if self .readiness in [CodeReadiness .NEXT_RELEASE , CodeReadiness .INCOMPLETE ]:
130+ env_vars += f" or { _PREFIX } { self .readiness .name }
131+ message = (
132+ f"The { self .name }
133+ f" To enable it, set { env_vars }
134+ )
135+ raise FeatureNotSupportedError (message )
136+
137+
138+ def requires_feature (
139+ feature_toggle : FeatureToggle ,
140+ ) -> Callable [[Callable [_P , _T ]], Callable [_P , _T ]]:
141+ """Decorator specifying that the function requires the specified feature toggle."""
142+
143+ def decorator (func : Callable [_P , _T ]) -> Callable [_P , _T ]:
144+ @functools .wraps (func )
145+ def wrapper (* args : _P .args , ** kwargs : _P .kwargs ) -> _T :
146+ feature_toggle ._raise_if_disabled ()
147+ return func (* args , ** kwargs )
148+
149+ return wrapper
150+
151+ return decorator
152+
153+
154+ # --------------------------------------
155+ # Define feature toggle constants here:
156+ # --------------------------------------
157+
158+ WAVEFORM_SUPPORT = FeatureToggle ("WAVEFORM_SUPPORT" , CodeReadiness .INCOMPLETE )
0 commit comments