diff --git a/main.py b/main.py new file mode 100644 index 0000000..0eb6c2a --- /dev/null +++ b/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() diff --git a/raw_data_pipeline_tools/__init__.py b/raw_data_pipeline_tools/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/raw_data_pipeline_tools/average_signal_fftsize.py b/raw_data_pipeline_tools/average_signal_fftsize.py new file mode 100644 index 0000000..492804e --- /dev/null +++ b/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) diff --git a/raw_data_pipeline_tools/convert_to_numpy.py b/raw_data_pipeline_tools/convert_to_numpy.py new file mode 100644 index 0000000..855976c --- /dev/null +++ b/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 diff --git a/raw_data_pipeline_tools/preprocessing_pipeline.py b/raw_data_pipeline_tools/preprocessing_pipeline.py new file mode 100644 index 0000000..94018ba --- /dev/null +++ b/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 diff --git a/raw_data_pipeline_tools/preprocessing_plots.py b/raw_data_pipeline_tools/preprocessing_plots.py new file mode 100644 index 0000000..63bed9c --- /dev/null +++ b/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()