Merge pull request #19 from TogetherCrew/feat/4-analytics-batch-save

feat: updated heatmap analytics to be generator!

setup.py

@@ -6,7 +6,7 @@
 
 setup(
     name="tc-analyzer-lib",
-    version="1.4.3",
+    version="1.4.4",
     author="Mohammad Amin Dadgar, TogetherCrew",
     maintainer="Mohammad Amin Dadgar",
     maintainer_email="dadgaramin96@gmail.com",

tc_analyzer_lib/metrics/heatmaps/heatmaps.py

@@ -39,7 +39,11 @@ def __init__(
         self.analyzer_config = analyzer_config
         self.utils = HeatmapsUtils(platform_id)
 
-    async def start(self, from_start: bool = False) -> list[dict]:
+    async def start(
+        self,
+        from_start: bool = False,
+        batch_return: int = 5,
+    ):
         """
         Based on the rawdata creates and stores the heatmap data
 
@@ -75,13 +79,14 @@ async def start(self, from_start: bool = False) -> list[dict]:
         async for bot in cursor:
             bot_ids.append(bot["id"])
 
+        index = 0
         while analytics_date.date() < datetime.now().date():
             start_day = analytics_date.replace(
                 hour=0, minute=0, second=0, microsecond=0
             )
             end_day = start_day + timedelta(days=1)
             logging.info(
-                f"{log_prefix} ANALYZING HEATMAPS {start_day.date()} - {end_day.date()}!"
+                f"{log_prefix} ANALYZING HEATMAPS {start_day.date()} - {end_day.date()}! | index: {index}"
             )
 
             # getting the active resource_ids (activities being done there by users)
@@ -150,10 +155,18 @@ async def start(self, from_start: bool = False) -> list[dict]:
 
                 heatmaps_results.extend(day_results)
 
+            if index % batch_return == 0:
+                yield heatmaps_results
+                # emptying it
+                heatmaps_results = []
+
+            index += 1
+
             # analyze next day
             analytics_date += timedelta(days=1)
 
-        return heatmaps_results
+        # returning any other values
+        yield heatmaps_results
 
     async def _prepare_heatmaps_document(
         self, date: datetime, resource_id: str, author_id: str

tc_analyzer_lib/tc_analyzer.py

@@ -85,20 +85,19 @@ async def run_once(self):
             resources=self.resources,
             analyzer_config=self.analyzer_config,
         )
-        heatmaps_data = await heatmaps_analysis.start(from_start=False)
-
-        # storing heatmaps since memberactivities use them
-        analytics_data = {}
-        analytics_data["heatmaps"] = heatmaps_data
-        analytics_data["memberactivities"] = (None, None)
-
-        self.DB_connections.store_analytics_data(
-            analytics_data=analytics_data,
-            platform_id=self.platform_id,
-            graph_schema=self.graph_schema,
-            remove_memberactivities=False,
-            remove_heatmaps=False,
-        )
+        async for heatmaps_data in heatmaps_analysis.start(from_start=False):
+            # storing heatmaps since memberactivities use them
+            analytics_data = {}
+            analytics_data["heatmaps"] = heatmaps_data
+            analytics_data["memberactivities"] = (None, None)
+
+            self.DB_connections.store_analytics_data(
+                analytics_data=analytics_data,
+                platform_id=self.platform_id,
+                graph_schema=self.graph_schema,
+                remove_memberactivities=False,
+                remove_heatmaps=False,
+            )
 
         memberactivity_analysis = MemberActivities(
             platform_id=self.platform_id,
@@ -149,13 +148,12 @@ async def recompute(self):
             resources=self.resources,
             analyzer_config=self.analyzer_config,
         )
-        heatmaps_data = await heatmaps_analysis.start(from_start=True)
 
-        # storing heatmaps since memberactivities use them
+        # This is to remove heatmaps data
+        # TODO: in future remove heatmaps better instead of using the lines below
         analytics_data = {}
-        analytics_data["heatmaps"] = heatmaps_data
+        analytics_data["heatmaps"] = []
         analytics_data["memberactivities"] = (None, None)
-
         self.DB_connections.store_analytics_data(
             analytics_data=analytics_data,
             platform_id=self.platform_id,
@@ -164,6 +162,20 @@ async def recompute(self):
             remove_heatmaps=True,
         )
 
+        async for heatmaps_data in heatmaps_analysis.start(from_start=True):
+            # storing heatmaps since memberactivities use them
+            analytics_data = {}
+            analytics_data["heatmaps"] = heatmaps_data
+            analytics_data["memberactivities"] = (None, None)
+
+            self.DB_connections.store_analytics_data(
+                analytics_data=analytics_data,
+                platform_id=self.platform_id,
+                graph_schema=self.graph_schema,
+                remove_memberactivities=False,
+                remove_heatmaps=False,
+            )
+
         # run the member_activity analyze
         logging.info(
             f"Analyzing the MemberActivities data for platform: {self.platform_id}!"

tests/integration/test_analyzer_period_week_run_once_empty_analytics.py

@@ -33,7 +33,8 @@ async def test_analyzer_week_period_run_once_empty_analytics(self):
         rawinfo_samples = []
 
         # generating random rawinfo data
-        for i in range(160):
+        # for past 7 days (168 / 24 = 7)
+        for i in range(168):
             author = np.random.choice(acc_id)
             replied_user = np.random.choice(acc_id)
             # not producing any self-interactions

tests/integration/test_assess_engagement_mention.py

@@ -39,7 +39,9 @@ async def heatmaps_analytics(self):
         """
         heatmaps are the input for assess_engagement's interaction matrix
         """
-        heatmaps_data = await self.heatmaps.start(from_start=True)
+        heatmaps_data = []
+        async for results in self.heatmaps.start(from_start=True):
+            heatmaps_data.extend(results)
 
         analytics_data = {}
         analytics_data["heatmaps"] = heatmaps_data

tests/integration/test_assess_engagement_reactions.py

@@ -39,7 +39,9 @@ async def heatmaps_analytics(self):
         """
         heatmaps are the input for assess_engagement's interaction matrix
         """
-        heatmaps_data = await self.heatmaps.start(from_start=True)
+        heatmaps_data = []
+        async for results in self.heatmaps.start(from_start=True):
+            heatmaps_data.extend(results)
 
         analytics_data = {}
         analytics_data["heatmaps"] = heatmaps_data

tests/integration/test_assess_engagement_replies.py

@@ -39,7 +39,9 @@ async def heatmaps_analytics(self):
         """
         heatmaps are the input for assess_engagement's interaction matrix
         """
-        heatmaps_data = await self.heatmaps.start(from_start=True)
+        heatmaps_data = []
+        async for results in self.heatmaps.start(from_start=True):
+            heatmaps_data.extend(results)
 
         analytics_data = {}
         analytics_data["heatmaps"] = heatmaps_data

tests/integration/test_heatmaps_analytics.py

@@ -156,7 +156,9 @@ async def test_heatmaps_single_day_from_start(self):
             sample_raw_data
         )
 
-        analytics = await self.heatmaps.start(from_start=True)
+        analytics = []
+        async for heatmaps_data in self.heatmaps.start(from_start=True):
+            analytics.extend(heatmaps_data)
 
         self.assertIsInstance(analytics, list)
 

tests/integration/test_heatmaps_analytics_different_source.py

@@ -150,7 +150,9 @@ async def test_heatmaps_single_day_from_start(self):
             sample_raw_data
         )
 
-        analytics = await self.heatmaps.start(from_start=True)
+        analytics = []
+        async for heatmaps_data in self.heatmaps.start(from_start=True):
+            analytics.extend(heatmaps_data)
 
         self.assertIsInstance(analytics, list)
 
@@ -181,7 +183,10 @@ async def test_heatmaps_analytics_pre_filled(self):
             }
         )
 
-        analytics = await self.heatmaps.start(from_start=False)
+        analytics = []
+        async for heatmaps_data in self.heatmaps.start(from_start=False):
+            analytics.extend(heatmaps_data)
+
         # the day was pre-filled before
         # and the period was exactly yesterday
         self.assertEqual(analytics, [])

tests/integration/test_heatmaps_hourly_lone_message.py

@@ -75,7 +75,9 @@ async def test_lone_messages(self):
             resources=list(channelIds),
             analyzer_config=DiscordAnalyzerConfig(),
         )
-        results = await analyzer_heatmaps.start(from_start=True)
+        results = []
+        async for heatmaps_data in analyzer_heatmaps.start(from_start=True):
+            results.extend(heatmaps_data)
 
         assert len(results) == len(acc_names) * DAY_COUNT * len(channelIds)
         for document in results:

tests/integration/test_heatmaps_hourly_mentions.py

@@ -104,7 +104,9 @@ async def test_mentioned_messages(self):
             resources=list(channelIds),
             analyzer_config=DiscordAnalyzerConfig(),
         )
-        results = await analyzer_heatmaps.start(from_start=True)
+        results = []
+        async for heatmaps_data in analyzer_heatmaps.start(from_start=True):
+            results.extend(heatmaps_data)
 
         assert len(results) == len(acc_names) * DAY_COUNT * len(channelIds)
         for document in results:

tests/integration/test_heatmaps_reactions.py

@@ -142,7 +142,9 @@ async def test_reacted_messages(self):
             resources=list(channelIds),
             analyzer_config=DiscordAnalyzerConfig(),
         )
-        results = await analyzer_heatmaps.start(from_start=True)
+        results = []
+        async for heatmaps_data in analyzer_heatmaps.start(from_start=True):
+            results.extend(heatmaps_data)
 
         assert len(results) == len(acc_names) * DAY_COUNT * len(channelIds)
         for document in results:

tests/integration/test_heatmaps_replier.py

@@ -101,7 +101,9 @@ async def test_reply_messages(self):
             resources=list(channelIds),
             analyzer_config=DiscordAnalyzerConfig(),
         )
-        results = await analyzer_heatmaps.start(from_start=True)
+        results = []
+        async for heatmaps_data in analyzer_heatmaps.start(from_start=True):
+            results.extend(heatmaps_data)
 
         assert len(results) == len(acc_names) * DAY_COUNT * len(channelIds)
         for document in results: