updating pipelining multisession packing

Inflow/__init__.py

-__version__ = "1.1.28"
+__version__ = "1.1.29"
 
 def ask_for_session_label_gui():
     import tkinter as tk

Inflow/core/pipelining.py

@@ -13,7 +13,7 @@ import inspect
 from .logging import get_local_logger, LogSession, session_log_decorator
 from ..ios.load import preprocessed_data as load_preprocessed_data
 from ..ios.save import preprocessed_data as save_preprocessed_data
-from ..ios.helpers import get_preprocessed_filename, get_multi_session_df
+from ..ios.helpers import get_preprocessed_filename
 from .optimization import job_context_decorator
 
 class Pipelines:
@@ -264,37 +264,13 @@ class MultisessionGetter:
     @staticmethod
     def multisession_getter(function):
         def multisession_wrapper(sessions, *args, add_session_level=False, **kwargs):
-            if add_session_level:
-                multi_session_data = {}
-                for _, session in sessions.iterrows():
-                    with LogSession(session):
-                        multi_session_data[session.u_alias] = function(
-                            *args, session_details=session, **kwargs
-                        )
-
-                if isinstance(
-                    multi_session_data[list(multi_session_data.keys())[0]],
-                    pd.core.frame.DataFrame,
-                ):
-                    multi_session_data = get_multi_session_df(
-                        multi_session_data.values(), multi_session_data.keys()
-                    )
 
-            else:
-                multi_session_data = []
-                for _, session in sessions.iterrows():
-                    with LogSession(session):
-                        session_data = function(
-                            *args, session_details=session, **kwargs
-                        )
-                        session_data.index = MultisessionGetter.merge_index(
-                            session_data.index, session
-                        )
-                        session_data.loc[:, "session#"] = session.u_alias
-                        multi_session_data.append(session_data)
-                multi_session_data = pd.concat(multi_session_data)
-
-            return multi_session_data
+            multi_session_data = {}
+            for _, session in sessions.iterrows():
+                with LogSession(session):
+                    multi_session_data[session.u_alias] = function(session, *args, **kwargs)
+
+            return MultisessionGetter.aggregate_multisession_data(multi_session_data, add_session_level = add_session_level)
 
         return multisession_wrapper
 
@@ -314,8 +290,13 @@ class MultisessionGetter:
 
     @staticmethod
     def multisession_generater(function):
-        def multisession_wrapper(*args, **kwargs):
-            raise NotImplementedError
+        def multisession_wrapper(sessions, *args, add_session_level = False, **kwargs):
+            multi_session_data = {}
+            for _, session in sessions.iterrows():
+                with LogSession(session):
+                    multi_session_data[session.u_alias] = function(session, *args, **kwargs)
+
+            return MultisessionGetter.aggregate_multisession_data(multi_session_data, add_session_level = add_session_level)
 
         return multisession_wrapper
 
@@ -342,6 +323,59 @@ class MultisessionGetter:
             return pd.MultiIndex.from_tuples(new_index, names=index.names)
         return pd.Index(new_index, name=index.name)
 
+    @staticmethod
+    def aggregate_multisession_data(multisession_data_dict, add_session_level = False):
+        
+        are_dataframe = [isinstance(item, pd.core.frame.DataFrame) for item in multisession_data_dict.values()]
+
+        if not all(are_dataframe):
+            return multisession_data_dict
+
+        return MultisessionGetter.get_multi_session_df(multisession_data_dict , add_session_level = add_session_level)
+
+    @staticmethod
+    def get_multi_session_df(multisession_data_dict, add_session_level = False):
+        
+        dataframes = []
+        for session_name, dataframe  in multisession_data_dict.items() :
+
+            level_names = list(dataframe.index.names)
+
+            dataframe = dataframe.reset_index()
+
+            if add_session_level :
+
+                dataframe["session#"] = [session_name] * len(dataframe) 
+                dataframe = dataframe.set_index(["session#"] + level_names , inplace = False)
+
+            else :
+                level_0_copy = dataframe[ level_names[0] ].copy()
+                dataframe[ level_names[0].replace("#","") ] = level_0_copy
+                dataframe["session"] = [session_name] * len(dataframe) 
+
+                dataframe[ level_names[0] ] = dataframe[ level_names[0] ].apply(
+                    MultisessionGetter.merge_index_element, session_name=session_name
+                )
+                dataframe = dataframe.set_index(level_names)
+
+            dataframes.append( dataframe )
+    
+        multisession_dataframe = pd.concat( dataframes )
+        return multisession_dataframe
+
+    @staticmethod
+    def merge_index_element(values, session_name):
+        if not isinstance(values, tuple):
+            values = (values,)
+
+        new_values = []
+        for value in values:
+            value = str(value) + "_" + session_name
+            new_values.append(value)
+
+        if len(new_values) == 1:
+            return new_values[0]
+        return tuple(new_values)
 
 def load_preprocessing(
     func, getter, checker, block_name = "<pipeline_block>"

Inflow/ios/helpers.py

@@ -12,7 +12,6 @@ from ..core.logging import get_local_logger
 import os
 import pandas as pd
 
-#@session_to_path_argument(0,"session_details")
 def get_suite2p_path(session_details, collection = "suite2p", plane = 0, file = r".*\.npy$" ): 
     #collection allow to force finding only in a give subfolder of the session_path
     if not isinstance(collection,(tuple,list)):
@@ -21,11 +20,6 @@ def get_suite2p_path(session_details, collection = "suite2p", plane = 0, file =
         return findfiles(os.path.join(session_details.path,*collection),re_pattern = fr"plane{plane}\\{file}")    
     except FileNotFoundError: 
         return []
-    
-# def get_suite2p_binpath(session_details, collection = "", plane = 0):
-#     if not isinstance(collection,(list,tuple)):
-#         collection = [collection]
-#     return findfiles(os.path.join(session_details,*collection),re_pattern = fr"plane{plane}\\data(?:_chan\d+)?\.bin$")
 
 def get_preprocessed_filename(session_details , alf_identifier, extra = "", collection = "preprocessing_saves", makedirs = False ):
     if extra is None or extra == "":
@@ -53,25 +47,4 @@ def cast_preprocessed_file_to_revision(session_details, alf_identifier, extra =
         logger.debug(f"Newly revised file path : {new_file_name}")
         os.rename(filename,new_file_name)
     else :
-        raise OSError(f"File {filename} does not exist, cannot cast file to revision")
-
-# def get_endresults_filenames(session_details, makedirs = False):
-#     filenames = {}
-#     possible_end_results = {"rois_df":{"collections":"end_results"},
-#                             "trials_df":{"collections":"end_results"},
-#                             "trials_roi_df":{"collections":"end_results"}
-#                             }
-#     for end_result_name in possible_end_results.keys() :
-#         filenames[end_result_name] = get_preprocessed_filename(session_details, 
-#                     f"{end_result_name}_{session_details.u_alias}", 
-#                     collections = possible_end_results[end_result_name]["collections"], 
-#                     makedirs = makedirs)
-#     return filenames
-
-def get_multi_session_df(dataframe_list, session_short_names_list):
-    multindex_list = []
-    for dataframe, session_name in zip(dataframe_list,session_short_names_list) :
-        index_tuples = [ (session_name,) + item if hasattr(item,"index") else (session_name, item) for item in dataframe.index]
-        index_names = ["session#"] + list(dataframe.index.names) 
-        multindex_list.append( pd.MultiIndex.from_tuples(index_tuples , names = index_names))
-    return pd.concat( [ dataframe.set_index(multindex, inplace = False) for dataframe, multindex in zip(dataframe_list, multindex_list) ] )
\ No newline at end of file
+        raise OSError(f"File {filename} does not exist, cannot cast file to revision")
\ No newline at end of file

Inflow/ios/load.py

@@ -14,8 +14,6 @@ from ..core.decorators import hide_unused_kwargs
 from ..core.special_types import AttrDict
 from .helpers import (
     get_preprocessed_filename,
-    #get_endresults_filenames,
-    get_multi_session_df,
     get_suite2p_path,
 )
 import numpy as np, pandas as pd, pickle, os
@@ -234,43 +232,6 @@ def mat_file(filepath):
             _recursively_add_key(result, filehandle)
         return result
 
-
-# def pipeline_end_result(session_details):
-#     local_log = get_local_logger()
-
-#     if isinstance(session_details, pd.core.frame.DataFrame):
-#         multi_session_data = AttrDict()
-#         for _, session in session_details.iterrows():
-#             multi_session_data[session.u_alias] = pipeline_end_result(session)
-
-#         multi_session_data = pd.DataFrame.from_dict(multi_session_data, orient="index")
-#         results = AttrDict()
-#         for column_name in multi_session_data.columns:
-#             results[column_name] = get_multi_session_df(
-#                 multi_session_data[column_name].tolist(),
-#                 multi_session_data.index.tolist(),
-#             )
-#         return results
-
-#     else:
-#         filenames = get_endresults_filenames(session_details)
-#         kept_files = {
-#             key: value for key, value in filenames.items() if os.path.isfile(value)
-#         }
-
-#         local_log.load_info(
-#             f"About to load len(kept_files) pandas table files from {[file for file in kept_files.values()]}"
-#         )
-
-#         results = AttrDict()
-#         for itemname, filename in kept_files.items():
-#             results[itemname] = pd.read_pickle(filename)
-
-#         local_log.load_info("Loading successfull")
-
-#         return results
-
-
 def commented_json_file(json_file):
     with open(json_file, "r") as file:
         json_string = file.read()