Replace _IntFlag class with simple int constants

Author: mMerlin

.pylintrc

@@ -19,7 +19,7 @@ ignore-patterns=
 
 # Python code to execute, usually for sys.path manipulation such as
 # pygtk.require().
-init-hook='import sys;sys.path.insert(0,"./lib")'
+init-hook=
 
 # Use multiple processes to speed up Pylint.
 jobs=1
@@ -159,7 +159,7 @@ ignored-classes=optparse.Values,thread._local,_thread._local
 # (useful for modules/projects where namespaces are manipulated during runtime
 # and thus existing member attributes cannot be deduced by static analysis. It
 # supports qualified module names, as well as Unix pattern matching.
-ignored-modules=board,socketpool
+ignored-modules=
 
 # Show a hint with possible names when a member name was not found. The aspect
 # of finding the hint is based on edit distance.

adafruit_ntp.py

@@ -37,7 +37,7 @@
 except ImportError:
     pass
 try:
-    from typing import Callable, Optional, Union, Tuple, Dict
+    from typing import Callable, Optional, Tuple, Dict
 except ImportError:
     pass
 try:
@@ -73,79 +73,14 @@ def recv_into(
 PACKET_SIZE: int = 48
 
 
-# No enum module in CircuitPython, so implement simple version.
-class _IntFlag:  # pylint:disable=too-few-public-methods
-    """Simple private implementation of an IntFlag enum.
-
-    This contains hard-coded limits specific to the EventType class."""
-
-    def __init__(self, value=0):
-        if isinstance(value, _IntFlag):
-            self.value = value.value
-        else:
-            self.value = value
-
-    def __and__(self, other):
-        if isinstance(other, _IntFlag):
-            return _IntFlag(self.value & other.value)
-        return self.value & other
-
-    def __or__(self, other):
-        if isinstance(other, _IntFlag):
-            return _IntFlag(self.value | other.value)
-        return self.value | other
-
-    def __xor__(self, other):
-        if isinstance(other, _IntFlag):
-            return _IntFlag(self.value ^ other.value)
-        return self.value ^ other
-
-    def __add__(self, other):
-        return self.__or__(other)
-
-    def __gt__(self, other):
-        if isinstance(other, _IntFlag):
-            return self.value > other.value
-        return self.value > other
-
-    def __lt__(self, other):
-        if isinstance(other, _IntFlag):
-            return self.value < other.value
-        return self.value < other
-
-    def __invert__(self):
-        # This relies on ALL_EVENTS existing, and being set correctly.
-        all_flags = EventType.ALL_EVENTS.value
-        return _IntFlag(~self.value & all_flags)
-
-    def __repr__(self):
-        return f"IntFlag({self.value})"
-
-    def __eq__(self, other):
-        if isinstance(other, _IntFlag):
-            return self.value == other.value
-        return self.value == other
-
-    def __bool__(self):
-        return bool(self.value)
-
-    @classmethod
-    def from_value(cls, value):
-        """enum emulation"""
-        return cls(value)
-
-
-class EventType(_IntFlag):  # pylint:disable=too-few-public-methods
+class EventType:  # pylint:disable=too-few-public-methods
     """NTP callback notification Event Types."""
 
-    NO_EVENT = _IntFlag(0b000)
-    SYNC_COMPLETE = _IntFlag(0b001)
-    SYNC_FAILED = _IntFlag(0b010)  # get packet failed
-    LOOKUP_FAILED = _IntFlag(0b100)  # getaddrinfo failed (not timedout?)
-    # The custom _IntFlat class used uses the ALL_EVENTS attribute here to get a bit mask to use
-    # when inverting flag values. Make sure to update this if changing how many event types are
-    # being used.
-    ALL_EVENTS = _IntFlag(0b111)
+    NO_EVENT = const(0b000)
+    SYNC_COMPLETE = const(0b001)
+    SYNC_FAILED = const(0b010)  # get packet failed
+    LOOKUP_FAILED = const(0b100)  # getaddrinfo failed (not timedout?)
+    ALL_EVENTS = const(0b111)
 
 
 class NTPIncompleteError(TimeoutError):
@@ -247,7 +182,7 @@ def __init__(  # pylint:disable=too-many-arguments
         self._next_sync: int = 0
         self._state: int = self.USING_CACHED_REFERENCE
         self._last_sync_time: int = 0  # Track the last successful sync time
-        self._callbacks: Dict[Callable[[_IntFlag, int], None], _IntFlag] = {}
+        self._callbacks: Dict[Callable[[int, int], None], int] = {}
         # The variables _next_rate_limit_end and _next_rate_limit_delay are intentionally
         # not initialized here because they are only used after calling
         # _initialize_backoff_timing(). Accessing them before initialization will raise an
@@ -317,7 +252,6 @@ def _update_time_sync(self) -> None:
         - GETTING_SOCKET: Perform DNS lookup for the NTP server.
         - GETTING_PACKET: Send NTP request and await response.
         """
-        self._notify_ntp_event_callbacks(EventType.NO_EVENT, -1)
         if self._state == self.USING_CACHED_REFERENCE:
             # Cached offset value expired, reinitialize backoff timing and proceed to DNS lookup.
             self._initialize_backoff_timing()
@@ -372,8 +306,8 @@ def _server_dns_lookup(self) -> None:
 
     def register_ntp_event_callback(
         self,
-        callback: Callable[[_IntFlag, int], None],
-        event_types: Union[_IntFlag, int] = EventType.SYNC_COMPLETE,
+        callback: Callable[[int, int], None],
+        event_types: int = EventType.SYNC_COMPLETE,
     ) -> None:
         """
         Register a callback to be notified for specific NTP events.
@@ -424,23 +358,20 @@ def on_ntp_event(event_type: IntFlag, next_time: int) -> None:
             ntp.register_ntp_event_callback(on_ntp_event,
                 EventType.SYNC_COMPLETE | EventType.SYNC_FAILED | EventType.LOOKUP_FAILED)
         """
-        if not isinstance(event_types, (_IntFlag, int)):
+        if not isinstance(event_types, int):
             raise TypeError(f"{type(event_types)} is not compatible with event types")
-        if not (
-            (EventType.ALL_EVENTS | event_types) == EventType.ALL_EVENTS
-            or event_types == EventType.NO_EVENT
+        if (
+            EventType.ALL_EVENTS | event_types != EventType.ALL_EVENTS
+            or event_types == 0
         ):
-            events_value = (
-                event_types if isinstance(event_types, int) else event_types.value
-            )
             raise TypeError(
-                f"Invalid event type mask 0b{events_value:b}. "
+                f"Invalid event type mask 0b{event_types:b}. "
                 "Only known events can receive notifications."
             )
         self._callbacks[callback] = event_types
 
     def _notify_ntp_event_callbacks(
-        self, event_type: _IntFlag, next_operation_time: int
+        self, event_type: int, next_operation_time: int
     ) -> None:
         """
         Call all registered callbacks that are interested in the given event type.

examples/ntp_connection_manager.py

@@ -10,7 +10,7 @@
 
 # determine which radio is available
 try:
-    import wifi
+    import wifi  # type:ignore
     import os
 
     # adjust method to get credentials as necessary...
@@ -22,7 +22,7 @@
 except ImportError:
     import board
     from digitalio import DigitalInOut
-    from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K
+    from adafruit_wiznet5k.adafruit_wiznet5k import WIZNET5K  # type:ignore
 
     # adjust with busio.SPI() as necessary...
     spi = board.SPI()

examples/ntp_powersave.py

@@ -6,8 +6,8 @@
 import os
 import time
 
-import socketpool
-import wifi
+import socketpool  # type:ignore
+import wifi  # type:ignore
 
 from adafruit_ntp import NTP, EventType, NTPIncompleteError
 

examples/ntp_set_rtc.py

@@ -6,9 +6,9 @@
 import os
 import time
 
-import rtc
-import socketpool
-import wifi
+import rtc  # type:ignore
+import socketpool  # type:ignore
+import wifi  # type:ignore
 
 import adafruit_ntp
 

examples/ntp_simpletest.py

@@ -6,8 +6,8 @@
 import os
 import time
 
-import socketpool
-import wifi
+import socketpool  # type:ignore
+import wifi  # type:ignore
 
 import adafruit_ntp
 

tests/mocks/mock_pool.py

@@ -228,7 +228,7 @@ def socket(
         return MockSocket(family, type, proto)
 
     def set_getaddrinfo_responses(
-        self, responses: List[Union[Exception, GetAddressInfoT]]
+        self, responses: List[Union[Exception, GetAddressInfoT]]  # type:ignore
     ) -> None:  # type:ignore
         """Set a sequence of responses for the getaddrinfo method."""
         self.mock_getaddrinfo_attempts = responses
@@ -256,8 +256,7 @@ def __init__(self, context: str):
 
     def mock_callback(self, event: int, delay: int) -> None:
         """Log which callback was called, and the triggering event details"""
-        event_num = event if isinstance(event, int) else event.value
-        if event_num == 0:
+        if event == 0:
             _logger.debug(MOCKED_CALLBACK_MSG % (self.context, event, delay))
             return
         _logger.info(MOCKED_CALLBACK_MSG % (self.context, event, delay))

tests/ntp_testing_support.py

@@ -66,7 +66,6 @@
 from adafruit_ntp import (  # pylint:disable=wrong-import-position
     NTP,
     EventType,
-    _IntFlag,
     NTPIncompleteError,
 )  # pylint:disable=wrong-import-position
 
@@ -120,7 +119,7 @@ def __init__(
         monotonic_start_ns: int = 0,
         next_sync: int = 0,
         last_sync_time: int = 0,
-        callbacks: Dict[Callable[[_IntFlag, int], None], _IntFlag] = None,
+        callbacks: Dict[Callable[[int, int], None], int] = None,
         state: int = 0,
         blocking: bool = True,
         monotonic_ns: int = 0,  # mocked functionality internal state information
@@ -138,7 +137,7 @@ def __init__(
         self.monotonic_start_ns: int = monotonic_start_ns
         self.next_sync: int = next_sync
         self.last_sync_time: int = last_sync_time
-        self.callbacks: Dict[Callable[[_IntFlag, int], None], _IntFlag] = (
+        self.callbacks: Dict[Callable[[int, int], None], int] = (
             {} if callbacks is None else dict(callbacks)
         )
         self.state: int = state
@@ -865,19 +864,11 @@ def _changed_state_fields(
         # in memory.
         # All attempts to find a data structure and comparison code to match expected and actual
         # state dictionary callable instance have failed. An expected entry created from exactly
-        # the same method as was registered still compares as not equal. The best can apparently
-        # be done, is to make sure that the registered event masks are the same.
+        # the same method as was registered still compares as not equal. The best that can
+        # apparently be done, is to make sure that the registered event masks are the same.
         callback1 = getattr(state1, field)
         callback2 = getattr(state2, field)
-        # Forcing all values to IntFlag instances seems easier than extracting the .value when the
-        # dict value is an IntFlag. Need to get them to be consistent for the sorting (on
-        # circuitpython) to work. Otherwise can get.
-        #     TypeError: unsupported types for __lt__: 'int', 'IntFlag'
-        # Either cpython automatically reverse the comparison, or it happens to only compare IntFlag
-        # to int, and not int to IntFlag.
-        if sorted([_IntFlag(v) for v in callback1.values()]) != sorted(
-            [_IntFlag(v) for v in callback2.values()]
-        ):
+        if sorted(callback1.values()) != sorted(callback2.values()):
             changed_fields.add(field)
 
     return changed_fields

tests/test_eventtype.py

@@ -9,83 +9,63 @@
 
 import unittest
 from tests.shared_for_testing import mock_cleanup
-from adafruit_ntp import EventType, _IntFlag
+from adafruit_ntp import EventType
 
 
 class TestEventType(unittest.TestCase):
     """Unittests for the EventType (emulated enum) class based on the custom IntFlag."""
 
     def test_basic_event_types(self):
         """Test basic event types and their values."""
-        self.assertEqual(EventType.NO_EVENT.value, _IntFlag(0b000))  # 0
-        self.assertEqual(EventType.SYNC_COMPLETE.value, _IntFlag(0b001))  # 1
-        self.assertEqual(EventType.SYNC_FAILED.value, _IntFlag(0b010))  # 2
-        self.assertEqual(EventType.LOOKUP_FAILED.value, _IntFlag(0b100))  # 4
-        self.assertEqual(EventType.ALL_EVENTS.value, _IntFlag(0b111))  # 7
+        self.assertEqual(EventType.NO_EVENT, 0b000)  # 0
+        self.assertEqual(EventType.SYNC_COMPLETE, 0b001)  # 1
+        self.assertEqual(EventType.SYNC_FAILED, 0b010)  # 2
+        self.assertEqual(EventType.LOOKUP_FAILED, 0b100)  # 4
+        self.assertEqual(EventType.ALL_EVENTS, 0b111)  # 7
 
     def test_event_combination(self):
         """Test bitwise OR combination of event types."""
         combined_event = EventType.SYNC_COMPLETE | EventType.LOOKUP_FAILED
-        self.assertEqual(combined_event, _IntFlag(0b101))  # 1 | 4 = 5
-        self.assertEqual(combined_event.value, 0b101)  # 1 | 4 = 5
+        self.assertEqual(combined_event, 0b101)  # 1 | 4 = 5
 
         combined_event = EventType.SYNC_COMPLETE | EventType.SYNC_FAILED
-        self.assertEqual(combined_event, _IntFlag(0b011))  # 1 | 2 = 3
-        self.assertEqual(combined_event.value, 0b011)  # 1 | 2 = 3
+        self.assertEqual(combined_event, 0b011)  # 1 | 2 = 3
 
         combined_event = EventType.SYNC_FAILED | EventType.LOOKUP_FAILED
-        self.assertEqual(combined_event, _IntFlag(0b110))  # 2 | 4 = 6
-        self.assertEqual(combined_event.value, 0b110)  # 2 | 4 = 6
+        self.assertEqual(combined_event, 0b110)  # 2 | 4 = 6
 
     def test_event_intersection(self):
         """Test bitwise AND intersection of event types."""
         combined_event = EventType.SYNC_COMPLETE | EventType.LOOKUP_FAILED
         intersect_event = combined_event & EventType.SYNC_COMPLETE
-        self.assertEqual(intersect_event.value, 0b001)  # 5 & 1 = 1
+        self.assertEqual(intersect_event, 0b001)  # 5 & 1 = 1
 
         intersect_event = combined_event & EventType.SYNC_FAILED
-        self.assertEqual(intersect_event.value, 0b000)  # 5 & 2 = 0
+        self.assertEqual(intersect_event, 0b000)  # 5 & 2 = 0
 
     def test_event_complement(self):
         """Test bitwise NOT complement of event types."""
-        complement_event = ~EventType.SYNC_COMPLETE
+        complement_event = (~EventType.SYNC_COMPLETE) & 0b111
         # Expected to be the bitwise complement of SYNC_COMPLETE within the range of all flags
-        expected_value = 2 | EventType.SYNC_FAILED.value | EventType.LOOKUP_FAILED.value
-        self.assertEqual(complement_event.value, expected_value)
-        self.assertNotEqual(complement_event.value, EventType.SYNC_COMPLETE.value)
+        expected_value = 2 | EventType.SYNC_FAILED | EventType.LOOKUP_FAILED
+        self.assertEqual(complement_event, expected_value)
+        self.assertNotEqual(complement_event, EventType.SYNC_COMPLETE)
         self.assertEqual(complement_event & EventType.SYNC_COMPLETE, 0b000)
         self.assertEqual(complement_event & EventType.SYNC_COMPLETE, EventType.NO_EVENT)
-        self.assertEqual(~EventType.NO_EVENT, 0b111)
-        self.assertEqual(~EventType.NO_EVENT, EventType.ALL_EVENTS)
 
     def test_event_equality(self):
         """Test equality between event types."""
         event1 = EventType.SYNC_COMPLETE
         event2 = EventType.SYNC_COMPLETE
         self.assertEqual(event1, event2)
 
-        combined_event = EventType.SYNC_COMPLETE | EventType.LOOKUP_FAILED
-        self.assertNotEqual(event1, combined_event)
-
-        self.assertEqual(EventType.SYNC_COMPLETE, EventType.SYNC_COMPLETE.value)
-        self.assertEqual(EventType.SYNC_COMPLETE, _IntFlag(0b001))
-        self.assertEqual(EventType.SYNC_COMPLETE, 0b001)
-        self.assertNotEqual(EventType.SYNC_COMPLETE, EventType.SYNC_FAILED)
-        self.assertNotEqual(EventType.SYNC_COMPLETE, EventType.NO_EVENT)
-        self.assertNotEqual(EventType.SYNC_COMPLETE, EventType.LOOKUP_FAILED)
-        self.assertNotEqual(EventType.SYNC_COMPLETE, EventType.ALL_EVENTS)
-        self.assertNotEqual(EventType.SYNC_COMPLETE, _IntFlag(0b000))
-        self.assertNotEqual(EventType.SYNC_COMPLETE, 0b000)
-        self.assertNotEqual(EventType.SYNC_COMPLETE, _IntFlag(0b010))
-        self.assertNotEqual(EventType.SYNC_COMPLETE, 0b010)
-
     def test_event_xor(self):
         """Test bitwise XOR operation of event types."""
         xor_event = EventType.SYNC_COMPLETE ^ EventType.SYNC_FAILED
-        self.assertEqual(xor_event.value, 0b011)  # 1 ^ 2 = 3
+        self.assertEqual(xor_event, 0b011)  # 1 ^ 2 = 3
 
         xor_event = EventType.SYNC_COMPLETE ^ EventType.SYNC_COMPLETE
-        self.assertEqual(xor_event.value, 0b000)  # 1 ^ 1 = 0
+        self.assertEqual(xor_event, 0b000)  # 1 ^ 1 = 0
 
     def test_event_to_bool(self):
         """Test conversion of event types to bool."""