Cleaned up tests. Fixed df repair header assumptions

Author: fnunnari

PostProcessing/PostProcessVideos.py

@@ -1,11 +1,8 @@
-import os
-import csv
 import argparse
 from pathlib import Path
-from typing import List, Dict, Tuple
+from typing import List, Tuple
 
 import pandas as pd
-import cv2
 import re
 
 from dataframes import compute_time_range, trim_into_interval, repair_dropped_frames

PostProcessing/dataframes.py

@@ -3,7 +3,8 @@
 
 from typing import Tuple
 
-THRESHOLD_NS = 10 * 1000 * 1000 #10 milisecond
+THRESHOLD_NS = 10 * 1000 * 1000  # 10 milisecond
+
 
 def compute_time_step(video_timestamps: pd.DataFrame) -> float:
     """
@@ -16,15 +17,18 @@ def compute_time_step(video_timestamps: pd.DataFrame) -> float:
     float: The time step of the video time stamps.
     """
 
-    time_step = (video_timestamps['timestamp'].diff()).dropna().value_counts().index[0]
-
+    first_col_name = video_timestamps.columns[0]
+    time_step = (video_timestamps[first_col_name].diff()).dropna().value_counts().index[0]
 
     return time_step
 
 
 def repair_dropped_frames(df: pd.DataFrame,  time_step: float) -> pd.DataFrame:
-    df['timestamp'] = pd.to_datetime(df['timestamp']).astype(np.int64)
-    timestamps = df['timestamp']
+
+    first_col_name = df.columns[0]
+
+    df[first_col_name] = pd.to_datetime(df[first_col_name]).astype(np.int64)
+    timestamps = df[first_col_name]
     repaired_rows = []
 
     # Check for missing timestamps and generate them
@@ -47,7 +51,7 @@ def repair_dropped_frames(df: pd.DataFrame,  time_step: float) -> pd.DataFrame:
 
     print(len(repaired_rows))
     # Create a new DataFrame with repaired rows
-    columns = ['timestamp', 'Generated']
+    columns = ['timestamp', 'generated']
     output_df = pd.DataFrame(repaired_rows, columns=columns)
     output_df['timestamp'] = pd.to_datetime(output_df['timestamp']).astype(np.int64)
 

PostProcessing/test_functions.py

@@ -2,19 +2,15 @@
 
 import os
 from pathlib import Path
-
 from typing import List, Tuple
 
-
-from dataframes import repair_dropped_frames, compute_time_range, trim_into_interval
-from PostProcessVideos import scan_session_dir
-
-
 import pandas as pd
 
-import pandas.testing as pd_testing
+from dataframes import compute_time_step
+from dataframes import repair_dropped_frames, compute_time_range
+
+from PostProcessVideos import scan_session_dir
 
-import numpy as np
 
 RECSYNCH_SESSION_DIR_VAR = "RECSYNCH_SESSION_DIR"
 
@@ -56,61 +52,35 @@ def session_data_list() -> List[Tuple[str, pd.DataFrame, str]]:
     clienIDs, dataframes, video_paths = scan_session_dir(Path(RECSYNCH_SESSION_DIR))
 
     for clientID, df, video_path in zip(clienIDs, dataframes, video_paths):
-        yield (clientID, df, video_path)
-
-
-def client_IDs() -> List[Path]:
-
-    out = []
-
-    for p in Path(RECSYNCH_SESSION_DIR).iterdir():
-        print("-->", p.stem)
-        out.append(p)
-
-    return out
-
+        yield clientID, df, video_path
 
-def CSVs() -> List[str]:
 
-    out = []
-
-    clients = client_IDs()
-    for c in clients:
-        client_dir = Path(RECSYNCH_SESSION_DIR) / c
-        for csv_file in client_dir.glob("*.csv"):
-            print("==>", csv_file)
-            rel_filepath = str(csv_file.relative_to(RECSYNCH_SESSION_DIR))
-            print("++>", rel_filepath)
-            out.append(rel_filepath)
-
-    return out
-
-
-@pytest.mark.parametrize("csv_file", CSVs())
-def test_df_reparation(csv_file):
-
-    # Load the test dataframes
-    csv_path = Path(RECSYNCH_SESSION_DIR) / csv_file
-    df = pd.read_csv(csv_path)
+@pytest.mark.parametrize("client_data", session_data_list())
+def test_df_reparation(client_data):
 
+    _, df, _ = client_data
     assert len(df) >= 2
 
-    repaired_df = repair_dropped_frames(df)
+    first_col_name = df.columns[0]
 
-    assert len(repaired_df) >= len(df)
+    time_step = compute_time_step(df)
+    assert time_step > 0
 
-@pytest.mark.parametrize("client_data", session_data_list())
-def test_df_reparation(client_data):
+    repaired_df = repair_dropped_frames(df, time_step)
 
-    _, df, _ = client_data
+    assert len(repaired_df) > 2
 
-    assert len(df) >= 2
+    assert len(repaired_df) >= len(df), "Dataframe size was reduced after reparation."
 
-    repaired_df = repair_dropped_frames(df)
+    repaired_first_col_name = repaired_df.columns[0]
+    repaired_second_col_name = repaired_df.columns[1]
+    assert repaired_first_col_name == "timestamp"
+    assert repaired_second_col_name == "generated"
 
-    assert len(repaired_df) >= len(df)
-    assert df[0].iloc[0] == repaired_df[0].iloc[0]
-    assert df[0].iloc[-1] == repaired_df[0].iloc[-1]
+    assert df[first_col_name].iloc[0] == repaired_df[repaired_first_col_name].iloc[0],\
+        "The first element changed after reparation"
+    assert df[first_col_name].iloc[-1] == repaired_df[repaired_first_col_name].iloc[-1],\
+        "The last element changed after reparation"
 
 
 def test_df_trimming(session_data):