first commit

measprocess/__init__.py

+from .preprocess import synchronize_df

measprocess/preprocess.py

+import pandas as pd
+
+def synchronize_df(majority_df, minority_df, column, fk_entry, merge_columns):
+    '''
+    Synchronize Dataframes along a column
+
+    Example:
+    Synchronize Measurements and GPS - Append merge columns to the final dataframe
+    synchronize_df(meas, gps, 'Datetime', 'fk_gps', ['Lon.', 'Lat.', 'Quality'])
+    '''
+
+    majority[fk_entry] = 0
+
+    for col in merge_columns:
+        majority[col] = 0
+
+    c = 0
+    for i, meas in majority.iterrows():
+        c += 1
+        print(f"{c} / {len(majority)}", end='\r')
+
+        sol = (meas[column] - minority[column]).abs().argmin()
+        majority.loc[i, fk_entry] = minority.iloc[sol].name
+        for col in merge_columns:
+            majority.loc[i, col] = minority.iloc[sol][col]
+
+    return majority

requirements.txt

+#Needs to be adapted
+
+pandas
+pytest

tests/preprocess_test.py

+import unittest
+from context import measprocess as mpc
+
+import os
+import pickle
+
+def extract_scan(path):
+    filepath = os.path.join(path, 'ofdmscan.p0')
+    with open(filepath, "rb") as fh:
+        data = pickle.load(fh)
+        return data[0][7][1]['rec'][0], data[0][7][1]['ot']['Band'], data[0][7]['ot']['Ch']
+
+def extract_gps(path):
+    filepath = os.path.join(path, 'gps.p0')
+    with open(filepath, "rb") as fh:
+        data = pickle.load(fh)
+        return data[0]['ot']
+
+class TestSum(unittest.TestCase):
+    def test_synchronize(self):
+        path = [
+            f"tests/example_files/",
+            'scanner'
+        ]
+
+        meas, band, channel  = extract_scan(
+            os.path.join(*path)
+        )
+
+        meas['band'] = band
+        meas['channel'] = channel
+
+        gps = extract_gps(
+            os.path.join(*path)
+        )
+
+        merged = mpc.preprocess.synchronize_df(meas, gps, 'Datetime', 'fk_gps', ['Lon.', 'Lat.', 'Quality'])
+
+        print(merged)
+
+        #self.assertEqual(sum([1, 2, 3]), 6, "Should be 6")
+
+    #def test_sum_tuple(self):
+    #    self.assertEqual(sum((1, 2, 2)), 6, "Should be 6")
+
+if __name__ == '__main__':
+    unittest.main()