Add min event filter to chip-repl ReadEvent (#23657)

* Add min event filter to chip-repl ReadEvent

* Add test to chip-repl

* Minor fix while pulling in master branch

src/controller/python/chip/ChipDeviceCtrl.py

@@ -1016,7 +1016,7 @@ async def Read(self, nodeid: int, attributes: typing.List[typing.Union[
         typing.Tuple[int,
                      typing.Type[ClusterObjects.ClusterEvent], int]
     ]] = None,
-            returnClusterObject: bool = False, reportInterval: typing.Tuple[int, int] = None, fabricFiltered: bool = True, keepSubscriptions: bool = False):
+            eventNumberFilter: typing.Optional[int] = None, returnClusterObject: bool = False, reportInterval: typing.Tuple[int, int] = None, fabricFiltered: bool = True, keepSubscriptions: bool = False):
         '''
         Read a list of attributes and/or events from a target node
 
@@ -1046,6 +1046,8 @@ async def Read(self, nodeid: int, attributes: typing.List[typing.Union[
             Clusters.ClusterA:                          Endpoint = *,           Cluster = specific,   Event = *, Urgent = True/False
             '*' or ():                                  Endpoint = *,           Cluster = *,          Event = *, Urgent = True/False
 
+        eventNumberFilter: Optional minimum event number filter.
+
         returnClusterObject: This returns the data as consolidated cluster objects, with all attributes for a cluster inside
                              a single cluster-wide cluster object.
 
@@ -1065,7 +1067,7 @@ async def Read(self, nodeid: int, attributes: typing.List[typing.Union[
         eventPaths = [self._parseEventPathTuple(
             v) for v in events] if events else None
 
-        ClusterAttribute.Read(future=future, eventLoop=eventLoop, device=device.deviceProxy, devCtrl=self, attributes=attributePaths, dataVersionFilters=clusterDataVersionFilters, events=eventPaths, returnClusterObject=returnClusterObject,
+        ClusterAttribute.Read(future=future, eventLoop=eventLoop, device=device.deviceProxy, devCtrl=self, attributes=attributePaths, dataVersionFilters=clusterDataVersionFilters, events=eventPaths, eventNumberFilter=eventNumberFilter, returnClusterObject=returnClusterObject,
                               subscriptionParameters=ClusterAttribute.SubscriptionParameters(reportInterval[0], reportInterval[1]) if reportInterval else None, fabricFiltered=fabricFiltered, keepSubscriptions=keepSubscriptions).raise_on_error()
         return await future
 
@@ -1124,7 +1126,7 @@ async def ReadEvent(self, nodeid: int, events: typing.List[typing.Union[
         typing.Tuple[int, typing.Type[ClusterObjects.Cluster], int],
         # Concrete path
         typing.Tuple[int, typing.Type[ClusterObjects.ClusterEvent], int]
-    ]], reportInterval: typing.Tuple[int, int] = None, keepSubscriptions: bool = False):
+    ]], eventNumberFilter: typing.Optional[int] = None, reportInterval: typing.Tuple[int, int] = None, keepSubscriptions: bool = False):
         '''
         Read a list of events from a target node, this is a wrapper of DeviceController.Read()
 
@@ -1144,10 +1146,11 @@ async def ReadEvent(self, nodeid: int, events: typing.List[typing.Union[
             ReadEvent(1, [ Clusters.Basic ] ) -- case 5 above.
             ReadEvent(1, [ (1, Clusters.Basic.Events.Location ] ) -- case 1 above.
 
+        eventNumberFilter: Optional minimum event number filter.
         reportInterval: A tuple of two int-s for (MinIntervalFloor, MaxIntervalCeiling). Used by establishing subscriptions.
             When not provided, a read request will be sent.
         '''
-        res = await self.Read(nodeid=nodeid, events=events, reportInterval=reportInterval, keepSubscriptions=keepSubscriptions)
+        res = await self.Read(nodeid=nodeid, events=events, eventNumberFilter=eventNumberFilter, reportInterval=reportInterval, keepSubscriptions=keepSubscriptions)
         if isinstance(res, ClusterAttribute.SubscriptionTransaction):
             return res
         else:

src/controller/python/chip/clusters/Attribute.py

@@ -22,7 +22,7 @@
 from asyncio.futures import Future
 import ctypes
 from dataclasses import dataclass, field
-from typing import Tuple, Type, Union, List, Any, Callable, Dict, Set
+from typing import Tuple, Type, Union, List, Any, Callable, Dict, Set, Optional
 from ctypes import CFUNCTYPE, c_char_p, c_size_t, c_void_p, c_uint64, c_uint32,  c_uint16, c_uint8, py_object, c_uint64
 import construct
 from rich.pretty import pprint
@@ -952,7 +952,7 @@ def WriteAttributes(future: Future, eventLoop, device, attributes: List[Attribut
 )
 
 
-def Read(future: Future, eventLoop, device, devCtrl, attributes: List[AttributePath] = None, dataVersionFilters: List[DataVersionFilter] = None, events: List[EventPath] = None, returnClusterObject: bool = True, subscriptionParameters: SubscriptionParameters = None, fabricFiltered: bool = True, keepSubscriptions: bool = False) -> PyChipError:
+def Read(future: Future, eventLoop, device, devCtrl, attributes: List[AttributePath] = None, dataVersionFilters: List[DataVersionFilter] = None, events: List[EventPath] = None, eventNumberFilter: Optional[int] = None, returnClusterObject: bool = True, subscriptionParameters: SubscriptionParameters = None, fabricFiltered: bool = True, keepSubscriptions: bool = False) -> PyChipError:
     if (not attributes) and dataVersionFilters:
         raise ValueError(
             "Must provide valid attribute list when data version filters is not null")
@@ -1032,6 +1032,9 @@ def Read(future: Future, eventLoop, device, devCtrl, attributes: List[AttributeP
         params.KeepSubscriptions = keepSubscriptions
     params.IsFabricFiltered = fabricFiltered
     params = _ReadParams.build(params)
+    eventNumberFilterPtr = ctypes.POINTER(ctypes.c_ulonglong)()
+    if eventNumberFilter is not None:
+        eventNumberFilterPtr = ctypes.POINTER(ctypes.c_ulonglong)(ctypes.c_ulonglong(eventNumberFilter))
 
     res = builtins.chipStack.Call(
         lambda: handle.pychip_ReadClient_Read(
@@ -1044,6 +1047,7 @@ def Read(future: Future, eventLoop, device, devCtrl, attributes: List[AttributeP
             ctypes.c_size_t(
                 0 if dataVersionFilters is None else len(dataVersionFilters)),
             ctypes.c_size_t(0 if events is None else len(events)),
+            eventNumberFilterPtr,
             *readargs))
 
     transaction.SetClientObjPointers(readClientObj, readCallbackObj)
@@ -1057,8 +1061,8 @@ def ReadAttributes(future: Future, eventLoop, device, devCtrl, attributes: List[
     return Read(future=future, eventLoop=eventLoop, device=device, devCtrl=devCtrl, attributes=attributes, dataVersionFilters=dataVersionFilters, events=None, returnClusterObject=returnClusterObject, subscriptionParameters=subscriptionParameters, fabricFiltered=fabricFiltered)
 
 
-def ReadEvents(future: Future, eventLoop, device, devCtrl, events: List[EventPath], returnClusterObject: bool = True, subscriptionParameters: SubscriptionParameters = None, fabricFiltered: bool = True) -> int:
-    return Read(future=future, eventLoop=eventLoop, device=device, devCtrl=devCtrl, attributes=None, dataVersionFilters=None, events=events, returnClusterObject=returnClusterObject, subscriptionParameters=subscriptionParameters, fabricFiltered=fabricFiltered)
+def ReadEvents(future: Future, eventLoop, device, devCtrl, events: List[EventPath], eventNumberFilter=None, returnClusterObject: bool = True, subscriptionParameters: SubscriptionParameters = None, fabricFiltered: bool = True) -> int:
+    return Read(future=future, eventLoop=eventLoop, device=device, devCtrl=devCtrl, attributes=None, dataVersionFilters=None, events=events, eventNumberFilter=eventNumberFilter, returnClusterObject=returnClusterObject, subscriptionParameters=subscriptionParameters, fabricFiltered=fabricFiltered)
 
 
 def Init():

src/controller/python/chip/clusters/attribute.cpp

@@ -391,7 +391,7 @@ void pychip_ReadClient_OverrideLivenessTimeout(ReadClient * pReadClient, uint32_
 
 PyChipError pychip_ReadClient_Read(void * appContext, ReadClient ** pReadClient, ReadClientCallback ** pCallback,
                                    DeviceProxy * device, uint8_t * readParamsBuf, size_t numAttributePaths,
-                                   size_t numDataversionFilters, size_t numEventPaths, ...)
+                                   size_t numDataversionFilters, size_t numEventPaths, uint64_t * eventNumberFilter, ...)
 {
     CHIP_ERROR err                 = CHIP_NO_ERROR;
     PyReadAttributeParams pyParams = {};
@@ -401,7 +401,7 @@ PyChipError pychip_ReadClient_Read(void * appContext, ReadClient ** pReadClient,
     std::unique_ptr<ReadClientCallback> callback = std::make_unique<ReadClientCallback>(appContext);
 
     va_list args;
-    va_start(args, numEventPaths);
+    va_start(args, eventNumberFilter);
 
     std::unique_ptr<AttributePathParams[]> attributePaths(new AttributePathParams[numAttributePaths]);
     std::unique_ptr<chip::app::DataVersionFilter[]> dataVersionFilters(new chip::app::DataVersionFilter[numDataversionFilters]);
@@ -462,6 +462,14 @@ PyChipError pychip_ReadClient_Read(void * appContext, ReadClient ** pReadClient,
             params.mpEventPathParamsList    = eventPaths.get();
             params.mEventPathParamsListSize = numEventPaths;
         }
+        if (eventNumberFilter != nullptr)
+        {
+            static_assert(sizeof(chip::EventNumber) == sizeof(*eventNumberFilter) &&
+                              std::is_unsigned<chip::EventNumber>::value ==
+                                  std::is_unsigned<std::remove_pointer<decltype(eventNumberFilter)>::type>::value,
+                          "EventNumber type mismatch");
+            params.mEventNumber = MakeOptional(EventNumber(*eventNumberFilter));
+        }
 
         params.mIsFabricFiltered = pyParams.isFabricFiltered;
 

src/controller/python/test/test_scripts/cluster_objects.py

@@ -320,7 +320,7 @@ async def _TriggerEvent(cls, devCtrl):
         # We trigger sending an event a couple of times just to be safe.
         await devCtrl.SendCommand(nodeid=NODE_ID, endpoint=1, payload=Clusters.UnitTesting.Commands.TestEmitTestEventRequest())
         await devCtrl.SendCommand(nodeid=NODE_ID, endpoint=1, payload=Clusters.UnitTesting.Commands.TestEmitTestEventRequest())
-        await devCtrl.SendCommand(nodeid=NODE_ID, endpoint=1, payload=Clusters.UnitTesting.Commands.TestEmitTestEventRequest())
+        return await devCtrl.SendCommand(nodeid=NODE_ID, endpoint=1, payload=Clusters.UnitTesting.Commands.TestEmitTestEventRequest())
 
     @ classmethod
     async def _RetryForContent(cls, request, until, retryCount=10, intervalSeconds=1):
@@ -337,6 +337,28 @@ async def TriggerAndWaitForEvents(cls, devCtrl, req):
         await cls._TriggerEvent(devCtrl)
         await cls._RetryForContent(request=lambda: devCtrl.ReadEvent(nodeid=NODE_ID, events=req), until=lambda res: res != 0)
 
+    @ classmethod
+    async def TriggerAndWaitForEventsWithFilter(cls, devCtrl, req):
+        response = await cls._TriggerEvent(devCtrl)
+        current_event_filter = response.value
+
+        def validate_got_expected_event(events):
+            number_of_events = len(events)
+            if number_of_events != 1:
+                return False
+
+            parsed_event_number = events[0].Header.EventNumber
+            if parsed_event_number != current_event_filter:
+                return False
+            return True
+
+        await cls._RetryForContent(request=lambda: devCtrl.ReadEvent(nodeid=NODE_ID, events=req, eventNumberFilter=current_event_filter), until=validate_got_expected_event)
+
+        def validate_got_no_event(events):
+            return len(events) == 0
+
+        await cls._RetryForContent(request=lambda: devCtrl.ReadEvent(nodeid=NODE_ID, events=req, eventNumberFilter=(current_event_filter + 1)), until=validate_got_no_event)
+
     @ classmethod
     @ base.test_case
     async def TestGenerateUndefinedFabricScopedEventRequests(cls, devCtrl):
@@ -393,6 +415,12 @@ async def TestReadEventRequests(cls, devCtrl, expectEventsNum):
 
         await cls.TriggerAndWaitForEvents(devCtrl, req)
 
+        logger.info("6: Reading Ex Cx Ex, with filter")
+        req = [
+            (1, Clusters.UnitTesting.Events.TestEvent, 0),
+        ]
+        await cls.TriggerAndWaitForEventsWithFilter(devCtrl, req)
+
         # TODO: Add more wildcard test for IM events.
 
     @ classmethod