logger and csv_to_npy

src/vame/logging/logger.py

@@ -9,17 +9,20 @@ class VameLogger:
         "%(asctime)-15s.%(msecs)d %(levelname)-5s --- [%(threadName)s]"
         " %(name)-15s : %(lineno)d : %(message)s"
     )
-    LOG_DATE_FORMAT = '%Y-%m-%d %H:%M:%S'
-
-    def __init__(self, base_name: str, file_path: str | None = None, log_level: int = logging.INFO):
+    LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
+
+    def __init__(
+        self,
+        base_name: str,
+        file_path: str | None = None,
+        log_level: int = logging.INFO,
+    ):
         self.log_level = log_level
         self.file_handler = None
         logging.basicConfig(
-            level=log_level,
-            format=self.LOG_FORMAT,
-            datefmt=self.LOG_DATE_FORMAT
+            level=log_level, format=self.LOG_FORMAT, datefmt=self.LOG_DATE_FORMAT
         )
-        self.logger = logging.getLogger(f'{base_name}')
+        self.logger = logging.getLogger(f"{base_name}")
         if self.logger.hasHandlers():
             self.logger.handlers.clear()
 
@@ -37,12 +40,11 @@ def __init__(self, base_name: str, file_path: str | None = None, log_level: int
 
     def add_file_handler(self, file_path: str):
         # File handler for logging to a file
-        #if not Path(file_path).parent.exists():
-
-        line_break = '\n'
+        # if not Path(file_path).parent.exists():
+        line_break = "\n"
         if not Path(file_path).exists():
             Path(file_path).parent.mkdir(parents=True, exist_ok=True)
-            line_break = ''
+            line_break = ""
 
         if self.file_handler is not None:
             self.file_handler.close()
@@ -53,8 +55,10 @@ def add_file_handler(self, file_path: str):
         with open(file_path, "a") as f:
             f.write(f"{line_break}[LOGGING STARTED AT: {handler_datetime}]")
 
-        self.file_handler = logging.FileHandler(file_path, mode='a')
-        self.file_handler.setFormatter(logging.Formatter(self.LOG_FORMAT, self.LOG_DATE_FORMAT))
+        self.file_handler = logging.FileHandler(file_path, mode="a")
+        self.file_handler.setFormatter(
+            logging.Formatter(self.LOG_FORMAT, self.LOG_DATE_FORMAT)
+        )
         self.logger.addHandler(self.file_handler)
 
     def remove_file_handler(self):
@@ -64,13 +68,11 @@ def remove_file_handler(self):
             self.file_handler = None
 
 
-
 class TqdmToLogger(io.StringIO):
     """
     Output stream for TQDM which will output to logger module instead of
     the StdOut.
     """
-
     logger = None
     level = None
     buf = ""
@@ -84,4 +86,4 @@ def write(self, buf):
         self.buf = buf.strip("\r\n\t ")
 
     def flush(self):
-        self.logger.log(self.level, self.buf)
+        self.logger.log(self.level, self.buf)

src/vame/util/csv_to_npy.py

@@ -12,83 +12,110 @@
 import os
 import numpy as np
 import pandas as pd
-
 from pathlib import Path
 from vame.util.auxiliary import read_config
 from vame.schemas.states import PoseToNumpyFunctionSchema, save_state
 from vame.logging.logger import VameLogger
-from vame.util.data_manipulation import interpol_first_rows_nans, read_pose_estimation_file
+from vame.util.data_manipulation import (
+    interpol_first_rows_nans,
+    read_pose_estimation_file,
+)
 
 
 logger_config = VameLogger(__name__)
 logger = logger_config.logger
 
 
-
 @save_state(model=PoseToNumpyFunctionSchema)
-def pose_to_numpy(config: str, save_logs=False) -> None:
-    """Converts a pose-estimation.csv file to a numpy array. Note that this code is only useful for data which is a priori egocentric, i.e. head-fixed
+def pose_to_numpy(
+    config: str,
+    save_logs=False,
+) -> None:
+    """
+    Converts a pose-estimation.csv file to a numpy array.
+    Note that this code is only useful for data which is a priori egocentric, i.e. head-fixed
     or otherwise restrained animals.
 
-    Raises:
-        ValueError: If the config.yaml file indicates that the data is not egocentric.
+    Parameters
+    ----------
+    config : str
+        Path to the config.yaml file.
+    save_logs : bool, optional
+        If True, the logs will be saved to a file, by default False.
+
+    Raises
+    ------
+    ValueError
+        If the config.yaml file indicates that the data is not egocentric.
     """
     try:
         config_file = Path(config).resolve()
-        cfg = read_config(config_file)
+        cfg = read_config(str(config_file))
 
         if save_logs:
-            log_path = Path(cfg['project_path']) / 'logs' / 'pose_to_numpy.log'
-            logger_config.add_file_handler(log_path)
-
-
-        path_to_file = cfg['project_path']
-        filename = cfg['video_sets']
-        confidence = cfg['pose_confidence']
-        if not cfg['egocentric_data']:
-            raise ValueError("The config.yaml indicates that the data is not egocentric. Please check the parameter egocentric_data")
+            log_path = Path(cfg["project_path"]) / "logs" / "pose_to_numpy.log"
+            logger_config.add_file_handler(str(log_path))
+
+        path_to_file = cfg["project_path"]
+        filename = cfg["video_sets"]
+        confidence = cfg["pose_confidence"]
+        if not cfg["egocentric_data"]:
+            raise ValueError(
+                "The config.yaml indicates that the data is not egocentric. Please check the parameter egocentric_data"
+            )
 
-        folder_path = os.path.join(path_to_file,'videos','pose_estimation')
-        paths_to_pose_nwb_series_data = cfg['paths_to_pose_nwb_series_data']
+        folder_path = os.path.join(path_to_file, "videos", "pose_estimation")
+        paths_to_pose_nwb_series_data = cfg["paths_to_pose_nwb_series_data"]
         for i, file in enumerate(filename):
             data, data_mat = read_pose_estimation_file(
                 folder_path=folder_path,
                 filename=file,
-                filetype=cfg['pose_estimation_filetype'],
-                path_to_pose_nwb_series_data=paths_to_pose_nwb_series_data if not paths_to_pose_nwb_series_data else paths_to_pose_nwb_series_data[i],
+                filetype=cfg["pose_estimation_filetype"],
+                path_to_pose_nwb_series_data=(
+                    paths_to_pose_nwb_series_data
+                    if not paths_to_pose_nwb_series_data
+                    else paths_to_pose_nwb_series_data[i]
+                ),
             )
 
             pose_list = []
 
             # get the number of bodyparts, their x,y-position and the confidence from DeepLabCut
-            for i in range(int(data_mat.shape[1]/3)):
-                pose_list.append(data_mat[:,i*3:(i+1)*3])
+            for i in range(int(data_mat.shape[1] / 3)):
+                pose_list.append(data_mat[:, i * 3 : (i + 1) * 3])
 
             # find low confidence and set them to NaN
             for i in pose_list:
                 for j in i:
                     if j[2] <= confidence:
-                        j[0],j[1] = np.nan, np.nan
+                        j[0], j[1] = np.nan, np.nan
 
             # interpolate NaNs
             for i in pose_list:
                 i = interpol_first_rows_nans(i)
 
             positions = np.concatenate(pose_list, axis=1)
-            final_positions = np.zeros((data_mat.shape[0], int(data_mat.shape[1]/3)*2))
+            final_positions = np.zeros(
+                (data_mat.shape[0], int(data_mat.shape[1] / 3) * 2)
+            )
 
             jdx = 0
             idx = 0
-            for i in range(int(data_mat.shape[1]/3)):
-                final_positions[:,idx:idx+2] = positions[:,jdx:jdx+2]
+            for i in range(int(data_mat.shape[1] / 3)):
+                final_positions[:, idx : idx + 2] = positions[:, jdx : jdx + 2]
                 jdx += 3
                 idx += 2
 
             # save the final_positions array with np.save()
-            np.save(os.path.join(path_to_file,'data',file,file+"-PE-seq.npy"), final_positions.T)
+            np.save(
+                os.path.join(path_to_file, "data", file, file + "-PE-seq.npy"),
+                final_positions.T,
+            )
             logger.info("conversion from DeepLabCut csv to numpy complete...")
 
-        logger.info("Your data is now in right format and you can call vame.create_trainset()")
+        logger.info(
+            "Your data is now in right format and you can call vame.create_trainset()"
+        )
     except Exception as e:
         logger.exception(f"{e}")
         raise e