From 273e60def1213f76f21e45d726ae10c9c08f64eb Mon Sep 17 00:00:00 2001 From: luiz Date: Thu, 12 Sep 2024 13:12:06 +0200 Subject: [PATCH] logger and csv_to_npy --- src/vame/logging/logger.py | 34 ++++++++------- src/vame/util/csv_to_npy.py | 85 ++++++++++++++++++++++++------------- 2 files changed, 74 insertions(+), 45 deletions(-) diff --git a/src/vame/logging/logger.py b/src/vame/logging/logger.py index 0d3aff15..4733c719 100644 --- a/src/vame/logging/logger.py +++ b/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) \ No newline at end of file + self.logger.log(self.level, self.buf) diff --git a/src/vame/util/csv_to_npy.py b/src/vame/util/csv_to_npy.py index 4ab83d79..e6cad9db 100644 --- a/src/vame/util/csv_to_npy.py +++ b/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