Merge pull request #1279 from Flowminder/blocking-init

Remove blocking call from DistanceCounterparts init

Author: mergify[bot]

flowmachine/flowmachine/features/subscriber/distance_counterparts.py

@@ -103,7 +103,7 @@ def __init__(
             subscriber_identifier="msisdn",
             hours=hours,
             subscriber_subset=subscriber_subset,
-        ).get_query()
+        )
 
         self.unioned_to_query = EventsTablesUnion(
             self.start,
@@ -113,7 +113,7 @@ def __init__(
             subscriber_identifier="msisdn_counterpart",
             hours=hours,
             subscriber_subset=subscriber_subset,
-        ).get_query()
+        )
 
         self.distance_matrix = DistanceMatrix()
 
@@ -141,8 +141,8 @@ def _make_query(self):
         FROM
             (
                 SELECT A.subscriber, A.location_id AS location_id_from, B.location_id AS location_id_to FROM
-                ({self.unioned_from_query}) AS A
-                JOIN ({self.unioned_to_query}) AS B
+                ({self.unioned_from_query.get_query()}) AS A
+                JOIN ({self.unioned_to_query.get_query()}) AS B
                 ON A.id = B.id AND A.outgoing != B.outgoing {on_filters}
             ) U
         JOIN

flowmachine/flowmachine/features/subscriber/interevent_period.py

@@ -110,7 +110,7 @@ def _make_query(self):
         sql = f"""
         SELECT
             subscriber,
-            EXTRACT(epoch FROM value/{self.time_divisor}) AS value
+            FLOOR(EXTRACT(epoch FROM value)/{self.time_divisor}) AS value
         FROM ({self.event_interval.get_query()}) AS U
         """
 

flowmachine/tests/test_subscriber_interevent_gap.py

@@ -35,7 +35,7 @@ def _intervent_period(*, start, stop, direction, subset, stat):
         agg = events.groupby("subscriber").agg(
             lambda x: getattr(x, postgres_stat_to_pandas_stat[stat])()
         )
-        agg = (pd.to_numeric(agg["duration"]) / 1000000000.0).to_dict()
+        agg = (agg["duration"].astype("timedelta64[s]")).to_dict()
 
         return agg
 
@@ -59,15 +59,19 @@ def test_interevent_period(
     """
 
     query = IntereventPeriod(
-        start=start, stop=stop, direction=direction, statistic=stat
+        start=start,
+        stop=stop,
+        direction=direction,
+        statistic=stat,
+        time_resolution="second",
     )
     df = get_dataframe(query).set_index("subscriber")
     sample = df.sample(n=5)
     want = intervent_period(
         start=start, stop=stop, direction=direction, stat=stat, subset=sample
     )
     assert query.column_names == ["subscriber", "value"]
-    assert (sample["value"] * query.time_divisor).to_dict() == pytest.approx(want)
+    assert (sample["value"]).to_dict() == pytest.approx(want)
 
 
 @pytest.mark.parametrize(
@@ -95,9 +99,7 @@ def test_interevent_interval(
         start=start, stop=stop, direction=direction, stat=stat, subset=sample
     )
     assert query.column_names == ["subscriber", "value"]
-    assert (pd.to_numeric(sample["value"]) / 1000000000.0).to_dict() == pytest.approx(
-        want
-    )
+    assert (sample["value"].astype("timedelta64[s]")).to_dict() == pytest.approx(want)
 
 
 @pytest.mark.parametrize("kwarg", ["direction", "statistic"])