implemented set group filter and aggregator functions

backend/experiment/experiment.py

@@ -13,6 +13,11 @@
 from hub.exceptions import ErrorDictException
 from session.data.session import SessionData
 
+from group_filters.group_filter_aggregator import GroupFilterAggregator
+from filters.filter_dict import FilterDict
+from group_filters import group_filter_aggregator_factory
+import asyncio
+
 if TYPE_CHECKING:
     from users import Experimenter, Participant
 
@@ -30,6 +35,8 @@ class Experiment(AsyncIOEventEmitter):
     session: SessionData
     _experimenters: list[Experimenter]
     _participants: dict[str, Participant]
+    _audio_group_filter_aggregators: dict[str, GroupFilterAggregator]
+    _video_group_filter_aggregators: dict[str, GroupFilterAggregator]
 
     def __init__(self, session: SessionData):
         """Start a new Experiment.
@@ -48,6 +55,8 @@ def __init__(self, session: SessionData):
         self._participants = {}
         self._logger.info(f"Experiment created: {self}")
         self.session.creation_time = timestamp()
+        self._audio_group_filter_aggregators = {}
+        self._video_group_filter_aggregators = {}
 
     def __str__(self) -> str:
         """Get string representation of this Experiment."""
@@ -422,3 +431,97 @@ def _set_state(self, state: ExperimentState) -> None:
 
         self._state = state
         self.emit("state", self._state)
+
+    async def set_video_group_filter_aggregators(
+        self, group_filter_configs: list[FilterDict], ports: list[int]
+    ) -> None:
+        old_group_filter_aggregators = self._video_group_filter_aggregators
+
+        self._video_group_filter_aggregators = {}
+        coroutines = []
+        for config, port in zip(group_filter_configs, ports):
+            filter_id = config["id"]
+            # Reuse existing filter for matching id and type.
+            if (
+                filter_id in old_group_filter_aggregators
+                and old_group_filter_aggregators[filter_id]._group_filter.config["type"]
+                == config["type"]
+            ):
+                self._video_group_filter_aggregators[
+                    filter_id
+                ] = old_group_filter_aggregators[filter_id]
+
+                self._video_group_filter_aggregators[
+                    filter_id
+                ]._group_filter.set_config(config)
+                self._video_group_filter_aggregators[filter_id].delete_data()
+            else:
+                # Create a new filter for configs with empty id.
+                self._video_group_filter_aggregators[
+                    filter_id
+                ] = group_filter_aggregator_factory.create_group_filter_aggregator(
+                    "video", config, port
+                )
+
+        # Cleanup old group filter aggregators
+        for (
+            filter_id,
+            old_group_filter_aggregator,
+        ) in old_group_filter_aggregators.items():
+            if filter_id not in self._video_group_filter_aggregators:
+                coroutines.append(old_group_filter_aggregator.cleanup())
+
+        # Run new group filter aggregators
+        for (
+            new_group_filter_aggregator
+        ) in self._video_group_filter_aggregators.values():
+            coroutines.append(new_group_filter_aggregator.run())
+
+        await asyncio.gather(*coroutines)
+
+    async def set_audio_group_filter_aggregators(
+        self, group_filter_configs: list[FilterDict], ports: list[int]
+    ) -> None:
+        old_group_filter_aggregators = self._audio_group_filter_aggregators
+
+        self._audio_group_filter_aggregators = {}
+        coroutines = []
+        for config, port in zip(group_filter_configs, ports):
+            filter_id = config["id"]
+            # Reuse existing filter for matching id and type.
+            if (
+                filter_id in old_group_filter_aggregators
+                and old_group_filter_aggregators[filter_id]._group_filter.config["type"]
+                == config["type"]
+            ):
+                self._audio_group_filter_aggregators[
+                    filter_id
+                ] = old_group_filter_aggregators[filter_id]
+
+                self._audio_group_filter_aggregators[
+                    filter_id
+                ]._group_filter.set_config(config)
+                self._audio_group_filter_aggregators[filter_id].delete_data()
+            else:
+                # Create a new filter for configs with empty id.
+                self._audio_group_filter_aggregators[
+                    filter_id
+                ] = group_filter_aggregator_factory.create_group_filter_aggregator(
+                    "audio", config, port
+                )
+
+        # Cleanup old group filter aggregators
+        for (
+            filter_id,
+            old_group_filter_aggregator,
+        ) in old_group_filter_aggregators.items():
+            if filter_id not in self._audio_group_filter_aggregators:
+                coroutines.append(old_group_filter_aggregator.cleanup())
+
+        # Run new group filter aggregators
+        for (
+            new_group_filter_aggregator
+        ) in self._audio_group_filter_aggregators.values():
+            coroutines.append(new_group_filter_aggregator.run())
+
+        await asyncio.gather(*coroutines)

backend/users/experimenter.py

@@ -12,6 +12,7 @@
 from typing import Any, Coroutine
 
 from filters import filter_utils
+from group_filters import group_filter_utils
 from session.data.session import is_valid_session
 from custom_types.chat_message import is_valid_chatmessage
 from custom_types.kick import is_valid_kickrequest
@@ -79,7 +80,7 @@ def __init__(self, experimenter_id: str, hub: _h.Hub) -> None:
         self.on_message("BAN_PARTICIPANT", self._handle_ban)
         self.on_message("MUTE", self._handle_mute)
         self.on_message("SET_FILTERS", self._handle_set_filters)
-        self.on_message("GROUP_FILTER", self._handle_group_filters)
+        self.on_message("SET_GROUP_FILTERS", self._handle_set_group_filters)
 
     def __str__(self) -> str:
         """Get string representation of this experimenter.
@@ -745,36 +746,65 @@ async def _handle_set_filters(self, data: Any) -> MessageDict:
         )
         return MessageDict(type="SUCCESS", data=success)
 
-    async def _handle_group_filters(self, data: Any) -> None:
-        self._logger.debug(f"Data received: {data}")
-        participant = data["participant"]
-
-        self.participants[participant] = data
-
-        if len(self.participants) > 1:
-            for p1 in self.participants.keys():
-                for p2 in self.participants.keys():
-                    if p1 != p2:
-                        p1_frame = int(list(self.participants[p1].keys())[0])
-                        p2_frame = int(list(self.participants[p2].keys())[0])
-
-                        if p1_frame > p2_frame:
-                            p1_data_aligned, p2_data_aligned = self.align(
-                                list(self.participants[p1].values())[0],
-                                list(self.participants[p2].values())[0],
-                            )
-                        else:
-                            p2_data_aligned, p1_data_aligned = self.align(
-                                list(self.participants[p2].values())[0],
-                                list(self.participants[p1].values())[0],
-                            )
-
-                        self.apply_oasis()
-
-    def align(self, longer_data, shorter_data):
-        for l, s in zip(longer_data, shorter_data):
-            if len(l) != len(s):
-                for _ in range(len(l) - len(s)):
-                    l.pop(0)
-
-        return longer_data, shorter_data
+    async def _handle_set_group_filters(self, data: Any) -> MessageDict:
+        if not group_filter_utils.is_valid_set_group_filters_request(data):
+            raise ErrorDictException(
+                code=400,
+                type="INVALID_DATATYPE",
+                description="Message data is not a valid SetGroupFiltersRequest.",
+            )
+
+        video_group_filters = data["video_group_filters"]
+        video_group_filter_ports = []
+        for _ in video_group_filters:
+            video_group_filter_ports.append(group_filter_utils.find_an_available_port())
+
+        audio_group_filters = data["audio_group_filters"]
+        audio_group_filter_ports = []
+        for _ in audio_group_filters:
+            audio_group_filter_ports.append(group_filter_utils.find_an_available_port())
+
+        experiment = self.get_experiment_or_raise("Failed to set filters.")
+        coroutines = []
+
+        # Update experiment with group filter aggregator
+        coroutines.append(
+            experiment.set_video_group_filter_aggregators(
+                video_group_filters, video_group_filter_ports
+            )
+        )
+        coroutines.append(
+            experiment.set_audio_group_filter_aggregators(
+                audio_group_filters, audio_group_filter_ports
+            )
+        )
+
+        # Update participant data
+        for p_data in experiment.session.participants.values():
+            p_data.video_group_filters = video_group_filters
+            p_data.audio_group_filters = audio_group_filters
+
+        # Update connected Participants
+        for p in experiment.participants.values():
+            if p.connection is not None:
+                coroutines.append(
+                    p.set_video_group_filters(
+                        video_group_filters, video_group_filter_ports
+                    )
+                )
+                coroutines.append(
+                    p.set_audio_group_filters(
+                        audio_group_filters, audio_group_filter_ports
+                    )
+                )
+        await asyncio.gather(*coroutines)
+
+        # Notify Experimenters connected to the hub about the data change
+        message = MessageDict(type="SESSION_CHANGE", data=experiment.session.asdict())
+        await self._hub.send_to_experimenters(message)
+
+        # Respond with success message
+        success = SuccessDict(
+            type="SET_GROUP_FILTERS", description="Successfully changed group filters."
+        )
+        return MessageDict(type="SUCCESS", data=success)

backend/users/user.py

@@ -488,7 +488,7 @@ async def set_video_filters(self, filters: list[FilterDict]) -> None:
             async def _set_video_filters_later(_):
                 if self._connection is None:
                     self._logger.error(
-                        "__set_video_filters_later callback failed, _connection is "
+                        "_set_video_filters_later callback failed, _connection is "
                         "None."
                     )
                     return
@@ -513,12 +513,46 @@ async def set_audio_filters(self, filters: list[FilterDict]) -> None:
             async def _set_audio_filters_later(_):
                 if self._connection is None:
                     self._logger.error(
-                        "__set_audio_filters_later callback failed, _connection is "
+                        "_set_audio_filters_later callback failed, _connection is "
                         "None."
                     )
                     return
                 await self._connection.set_audio_filters(filters)
 
+    async def set_video_group_filters(
+        self, group_filters: list[FilterDict], ports: list[int]
+    ) -> None:
+        if self._connection is not None:
+            await self._connection.set_video_group_filters(group_filters, ports)
+        else:
+
+            @self.once("connection_set")
+            async def _set_video_group_filters_later(_):
+                if self._connection is None:
+                    self._logger.error(
+                        "_set_video_group_filters_later callback failed, _connection is "
+                        "None."
+                    )
+                    return
+                await self._connection.set_video_group_filters(group_filters, ports)
+
+    async def set_audio_group_filters(
+        self, group_filters: list[FilterDict], ports: list[int]
+    ) -> None:
+        if self._connection is not None:
+            await self._connection.set_audio_group_filters(group_filters, ports)
+        else:
+
+            @self.once("connection_set")
+            async def _set_audio_group_filters_later(_):
+                if self._connection is None:
+                    self._logger.error(
+                        "_set_audio_group_filters_later callback failed, _connection is "
+                        "None."
+                    )
+                    return
+                await self._connection.set_audio_group_filters(group_filters, ports)
+
     async def start_recording(self) -> None:
         """Start recording for this user."""
         if self._connection is not None: