Raw data processing notebook to Python files

main.py

@@ -0,0 +1,17 @@
+import pandas as pd
+import numpy as np
+import matplotlib.pyplot as plt
+
+# Import functions from folder
+from raw_data_pipeline_tools.preprocessing_pipeline import preprocessing_pipeline
+
+def main():
+
+    on_filename = "/home/dimitrios-pakakis/Desktop/Astro/data-analysis/2502202_Hot202020.dat"
+    off_filename = "/home/dimitrios-pakakis/Desktop/Astro/data-analysis/2502202_Cold202020.dat"
+    fft_size = 2048
+    _ = preprocessing_pipeline(on_signal_filename=on_filename, off_signal_filename=off_filename,fft_size=fft_size,calibration_method="on/off",plot_analysis=True)
+
+
+if __name__ == "__main__":
+    main()

raw_data_pipeline_tools/average_signal_fftsize.py

@@ -0,0 +1,26 @@
+import numpy as np
+
+
+def get_avg_signal(time_series: np.ndarray, fft_size: int) -> np.ndarray:
+    """
+    Create the average spectrum from the time series.
+
+    Args:
+       time_series (np.ndarray): The time series to be averaged
+       fft_size (int): The size that was used for the fast fourier transformation
+    Returns:
+       np.ndarray: The average spectrum
+    """
+    if fft_size <= 0:
+        msg = "fft_size must be a positive integer"
+        raise ValueError(msg)
+
+    usable_size = (time_series.size // fft_size) * fft_size
+
+    if usable_size == 0:
+        msg = "time_series must contain at least one complete fft_size block"
+        raise ValueError(msg)
+
+    trimmed = time_series[:usable_size]
+    reshaped = trimmed.reshape(-1, fft_size)
+    return reshaped.mean(axis=0)

raw_data_pipeline_tools/convert_to_numpy.py

@@ -0,0 +1,24 @@
+import numpy as np
+
+
+def convert_dat_to_numpy(on_file_path: str, off_file_path: str) -> tuple[np.ndarray, np.ndarray]:
+    """
+    Convert .dat files to numpy arrays.
+
+    Args:
+        on_file_path (str): The file path of the on observation .dat file.
+        off_file_path (str): The file path of the off observation .dat file.
+
+    Returns:
+        tuple[np.ndarray, np.ndarray]: A tuple containing the on and off signal as numpy arrays.
+    """
+    try:
+        on_signal_numpy = np.fromfile(on_file_path, dtype=np.float32)
+        off_signal_numpy = np.fromfile(off_file_path, dtype=np.float32)
+
+    except FileNotFoundError:
+        print("File not found")  # noqa: T201
+    except Exception as e:  # noqa: BLE001
+        print(f"Convert Error: {e}")  # noqa: T201
+    else:
+        return on_signal_numpy, off_signal_numpy

raw_data_pipeline_tools/preprocessing_pipeline.py

@@ -0,0 +1,41 @@
+import numpy as np
+
+from raw_data_pipeline_tools.convert_to_numpy import convert_dat_to_numpy
+from raw_data_pipeline_tools.average_signal_fftsize import get_avg_signal
+from raw_data_pipeline_tools.preprocessing_plots import create_preprocessing_plots
+
+
+
+def preprocessing_pipeline(
+    on_signal_filename: str, off_signal_filename: str, fft_size: int, calibration_method: str = "on/off", plot_analysis: bool = True
+) -> np.ndarray:
+    """
+    Take the raw on and off observations, average the signals to create the spectrum and calibrate.
+
+    Args:
+        on_signal_filename (str): The on observation filename
+        off_signal_filename (str): The off observation filename
+        fft_size (int): The size that was used for the fast fourier transformation
+        calibration_method (str, optional): The calibration method to be used. (on/off or on-off). Defaults to "on/off".
+        plot_analysis (bool, optional): If True then plot the off,on and calibrated spectrum. Defaults to True.
+    """
+    # Convert the files to numpy arrays
+    on_spectrum, off_spectrum = convert_dat_to_numpy(on_signal_filename, off_signal_filename)
+
+    # Average the time series using the fft size
+    on_spectrum_avg: np.ndarray = get_avg_signal(on_spectrum, fft_size=fft_size)
+    off_spectrum_avg: np.ndarray = get_avg_signal(off_spectrum, fft_size=fft_size)
+
+    # Calibration
+    if calibration_method == "on/off":
+        calibrated_signal: np.ndarray = on_spectrum_avg / off_spectrum_avg
+    elif calibration_method == "on-off":
+        calibrated_signal: np.ndarray = on_spectrum_avg - off_spectrum_avg
+    else:
+        raise ValueError(f"Calibration Method does not exists. : {calibration_method}")
+
+    # If we want plots
+    if plot_analysis:
+        create_preprocessing_plots(on_spectrum_avg, off_spectrum_avg, calibrated_signal, fft_size)
+
+    return calibrated_signal

raw_data_pipeline_tools/preprocessing_plots.py

@@ -0,0 +1,32 @@
+import matplotlib.pyplot as plt
+import numpy as np
+
+
+def create_preprocessing_plots(on_spectrum_avg: np.ndarray, off_spectrum_avg: np.ndarray, calibrated_signal: np.ndarray, fft_size: int) -> None:
+    """
+    Create on_spectrum , off_spectrum , calibrated_spectrum plots in frequencies axes.
+
+    Args:
+        on_spectrum_avg (np.ndarray): the on spectrum
+        off_spectrum_avg (np.ndarray): the off spectrum
+        calibrated_signal (np.ndarray): the calibrated spectrum
+        fft_size (int): the fft size used in the observation
+    """
+    frequencies = np.linspace(1.4205 - 0.003840 / 2, 1.4205000 + 0.003840 / 2, fft_size)
+
+    _, (ax1, ax2, ax3) = plt.subplots(nrows=3, ncols=1, figsize=(8, 10))
+
+    ax1.plot(frequencies, off_spectrum_avg, color="blue")
+    ax1.set_title("Avg Cold/Off")
+    ax1.set_ylabel("Relative Power")
+
+    ax2.plot(frequencies, on_spectrum_avg, color="red")
+    ax2.set_title("Avg Hot/On")
+    ax2.set_ylabel("Relative Power")
+
+    ax3.plot(frequencies, calibrated_signal, color="green")
+    ax3.set_title("On/Off calibration")
+    ax3.set_ylabel("Relative Power")
+    ax3.set_xlabel("Frequencies")
+    plt.tight_layout()
+    plt.show()