add new files to file info and collect window summary

Author: danielparthier

ephys/GUI/datacontainer.py

@@ -0,0 +1,32 @@
+import numpy as np
+from ephys.classes.experiment_objects import ExpData
+
+
+class DataContainer:
+    def __init__(self, data: ExpData):
+        self.data = data
+
+    def get_data(self) -> ExpData:
+        return self.data
+
+    def set_data(self, new_data: ExpData) -> None:
+        self.data = new_data
+
+    def new_files_check(self, file_path: str | list[str]) -> list[str]:
+        # Implement duplicate checking logic here
+        existing_file_path = self.data.meta_data.get_file_path()
+        new_files = np.array(file_path)[
+            np.invert(np.isin(file_path, existing_file_path))
+        ]
+        if new_files.size == 0:
+            print("No new files to add.")
+        else:
+            print(f"New files detected: {new_files}")
+        return new_files.tolist()
+
+    def add_file(self, file_path: str | list[str]) -> None:
+        new_files = self.new_files_check(file_path)
+        self.data.add_file(new_files)
+
+    def __repr__(self) -> str:
+        return f"DataContainer(data={self.data})"

ephys/classes/experiment_objects.py

@@ -4,6 +4,7 @@
 
 import os
 import struct
+from typing import Any
 
 from datetime import datetime, timedelta
 import matplotlib.style as mplstyle
@@ -366,9 +367,16 @@ def add_file(
         """
         if isinstance(file_path, str):
             file_path = [file_path]
-        for file in file_path:
+
+        # Check for duplicates
+        existing_file_path = self.meta_data.get_file_path()
+        new_files = np.array(file_path)[
+            np.invert(np.isin(file_path, existing_file_path))
+        ].tolist()
+
+        for file in new_files:
             self.protocols.append(Trace(file))
-        self.meta_data.add_file_info(file_path, experimenter)
+        self.meta_data.add_file_info(new_files, experimenter)
         if sort:
             self.sort_by_date()
 
@@ -425,5 +433,91 @@ def meta_data_summary(self, to_dataframe: bool = True) -> dict | pd.DataFrame:
             return pd.DataFrame(summary_dict)
         return summary_dict
 
+    def get_window_summary(self, remove_duplicates: bool = True) -> pd.DataFrame:
+        """
+        Returns a summary of the window information for each protocol.
+        """
+        pd_collection = []
+        meta_df = self.meta_data_summary()
+        exp_start = np.array(
+            [
+                protocol.rec_datetime.timestamp()
+                for protocol in self.protocols
+                if protocol.rec_datetime is not None
+            ]
+        ).min()
+        for i, protocol in enumerate(self.protocols):
+            protocol: Trace
+            df = protocol.window_summary.to_dataframe()
+            df["file_name"] = meta_df["file_name"][i]
+            if protocol.rec_datetime is not None:
+                df["exp_time"] = (
+                    df["time"] + protocol.rec_datetime.timestamp() - exp_start
+                )
+            df["protocol"] = protocol
+            pd_collection.append(df)
+        df_out = pd.concat(pd_collection, ignore_index=True)
+        if remove_duplicates:
+            df_out.drop_duplicates(inplace=True)
+        return df_out
+
+    def window_function(
+        self,
+        window: list | tuple | None = None,
+        channels: Any = None,
+        signal_type: Any = None,
+        rec_type: str = "",
+        function: str = "mean",
+        label: str = "",
+        sweep_subset: Any = None,
+        return_output: bool = False,
+        plot_individual: bool = False,
+    ) -> None:
+        for protocol in self.protocols:
+            protocol: Trace
+            protocol.window_function(
+                window=window,
+                channels=channels,
+                signal_type=signal_type,
+                rec_type=rec_type,
+                function=function,
+                label=label,
+                sweep_subset=sweep_subset,
+                return_output=return_output,
+                plot=plot_individual,
+            )
+
+    def label_diff(
+        self, labels: list | None = None, new_name: str = "", time_label: str = ""
+    ) -> None:
+        for protocol in self.protocols:
+            protocol: Trace
+            protocol.window_summary.label_diff(
+                labels=labels, new_name=new_name, time_label=time_label
+            )
+
+    def label_ratio(
+        self, labels: list | None = None, new_name: str = "", time_label: str = ""
+    ) -> None:
+        for protocol in self.protocols:
+            protocol: Trace
+            protocol.window_summary.label_ratio(
+                labels=labels, new_name=new_name, time_label=time_label
+            )
+
+    def plot_summary(
+        self,
+        align_onset: bool = False,
+        **kwargs,
+    ) -> None:
+        from ephys.classes.window_functions import FunctionOutput
+
+        df_complete: pd.DataFrame = self.get_window_summary()
+        summary_output = FunctionOutput()
+        df_complete = df_complete.drop(columns=["time"], errors="ignore")
+        df_complete = df_complete.rename(columns={"exp_time": "time"})
+        summary_output.from_dataframe(df=df_complete)
+        summary_output.plot(plot_trace=False, align_onset=align_onset, **kwargs)
+
     # TODO: get summary outputs for the experiment
     # TODO: get summary plots for the experiment