Extract file names

.gitignore

@@ -81,3 +81,7 @@ venv/
 
 # written by setuptools_scm
 **/_version.py
+
+# config specific
+**/config.yml
+.env

load_suite2p/load/load_data.py

@@ -1,12 +1,12 @@
 import logging
-from pathlib import Path
 from typing import Tuple
 
-from read_config import read
+from decouple import config
 
 from ..objects.specifications import Specifications
+from .read_config import read
 
-config_path = Path(__file__).parent / "config/config.yml"
+CONFIG_PATH = config("CONFIG_PATH")
 
 
 def load_data(folder_name: str) -> Tuple[list, Specifications]:
@@ -51,7 +51,7 @@ def get_specifications(folder_name: str) -> Specifications:
  return specs
 
 
-def load(config: Specifications) -> list:
+def load(specs: Specifications) -> list:
  raise NotImplementedError("TODO")
 
 
@@ -65,7 +65,7 @@ def read_configurations() -> dict:
  """
 
  logging.debug("Reading configurations")
- config = read(config_path)
+ config = read(CONFIG_PATH)
  logging.debug(f"Configurations read: {config}")
 
  return config

load_suite2p/main.py

@@ -1,9 +1,9 @@
-from analysis.spatial_freq_temporal_freq import SF_TF
-from load.load_data import load_data
-from objects.photon_data import PhotonData
-from plots.plotter import Plotter
 from rich.prompt import Prompt
 
+from .analysis.spatial_freq_temporal_freq import SF_TF
+from .load.load_data import load_data
+from .objects.photon_data import PhotonData
+from .plots.plotter import Plotter
 from .utils import exception_handler, start_logging
 
 

load_suite2p/objects/enums.py

@@ -0,0 +1,17 @@
+from enum import Enum
+
+
+class DataType(Enum):
+ SIGNAL = 1
+ STIMULUS_INFO = 2
+ TRIGGER_INFO = 3
+ REGISTERS2P = 4
+ ALLEN_DFF = 5
+ NOT_FOUND = 6
+
+
+class AnalysisType(Enum):
+ SF_TF = 1
+ SPARSE_NOISE = 2
+ RETINOTOPY = 3
+ UNCLASSIFIED = 4

load_suite2p/objects/file.py

@@ -0,0 +1,55 @@
+from pathlib import Path
+
+from .enums import AnalysisType, DataType
+
+
+class File:
+ """Class containing the name of the file, its path and its
+ extension.
+
+ Attributes
+ ----------
+ name: str
+ file name
+
+ path: Path
+ complete file path
+
+ extension: str
+ file extension
+ """
+
+ def __init__(self, name: str, path: Path):
+ self.name = name
+ self.path: Path = path
+ self._path_str = str(path)
+ self.datatype: DataType = self._get_data_type()
+ self.analysistype: AnalysisType = self._get_analysis_type()
+
+ def _get_data_type(self) -> DataType:
+ if (
+ "suite2p" in self._path_str
+ or "plane0" in self._path_str
+ or "Fall.mat" in self._path_str
+ ):
+ return DataType.SIGNAL
+ elif "stimulus_info.mat" in self._path_str:
+ return DataType.STIMULUS_INFO
+ elif "trigger_info.mat" in self._path_str:
+ return DataType.TRIGGER_INFO
+ elif "rocro_reg.mat" in self._path_str:
+ return DataType.REGISTERS2P
+ elif "allen_dff.mat" in self._path_str:
+ return DataType.ALLEN_DFF
+ else:
+ return DataType.NOT_FOUND
+
+ def _get_analysis_type(self) -> AnalysisType:
+ if "sf_tf" in str(self._path_str):
+ return AnalysisType.SF_TF
+ elif "sparse_noise" in str(self._path_str):
+ return AnalysisType.SPARSE_NOISE
+ elif "retinotopy" in str(self._path_str):
+ return AnalysisType.RETINOTOPY
+ else:
+ return AnalysisType.UNCLASSIFIED

load_suite2p/objects/folder_naming_specs.py

@@ -1,6 +1,8 @@
 import logging
 from pathlib import Path
 
+from .enums import DataType
+from .file import File
 from .parsers2p.parser2pRSP import Parser2pRSP
 
 
@@ -48,7 +50,7 @@ def __init__(
 
  self.folder_name = folder_name
 
- logging.info("Parsing folder name")
+ logging.info(f"Parsing folder name: {folder_name}")
  self.parse_name()
  self.mouse_line = self._parser.info["mouse_line"]
  self.mouse_id = self._parser.info["mouse_id"]
@@ -66,12 +68,12 @@ def __init__(
  except KeyError:
  self.cre = None
 
- if not self.check_if_folder_exists():
- logging.error(f"File {self.get_path()} does not exist")
- raise FileNotFoundError(
-  f"File {self.folder_name} not found. "
-  + "Please check the file name and try again."
-  )
+ self.paths = [
+ self._parser.get_path_to_experimental_folder(),
+ self._parser.get_path_to_stimulus_analog_input_schedule_files(),
+ self._parser.get_path_to_serial2p(),
+ ]
+ self.allen_dff_file_path = self._parser.get_path_to_allen_dff_file()
 
  def parse_name(self) -> None:
  """Parses the folder name and evaluates the parameters `mouse_line`,
@@ -98,32 +100,70 @@ def parse_name(self) -> None:
  not supported"
  )
 
- def get_path(self) -> Path:
- """Returns the path to the folder containing the experimental data.
- Reads the server location from the config file and appends the
- parent folder and the given folder name.
+ def extract_all_file_names(self) -> None:
+ """Recursively searches files in the given folder.
+ It also locates the allen_dff file and the serial2p files.
+
+ Raises:
+ FileNotFoundError: if the allen_dff is not present
+ FileNotFoundError: if the serial2p folder is not present
 
- Returns
- -------
- Path
- path to the folder containing the experimental data
+ Returns:
+ list: of :class:`File` containing all read files with
+ their path and extension.
  """
- return self._parser.get_path()
+ logging.info("Extracting all file names")
+
+ self.all_files = []
+
+ if self.original_config["use-allen-dff"]:
+ logging.info("Using allen dff file")
+
+ if self.allen_dff_file_path.exists():
+ self.all_files.append(
+ File(
+ name=self.allen_dff_file_path.name,
+ path=self.allen_dff_file_path,
+ )
+ )
+
+ else:
+ logging.info("No allen dff file found")
+ raise FileNotFoundError(
+ "No allen dff file found. Is this path correct: "
+ + f"{self.allen_dff_file_path}?"
+ )
+ else:
+ logging.info("Not using allen dff file")
+
+ for path in self.paths:
+ if path.exists():
+ self.search_file_paths(path)
+ else:
+ logging.info(f"No files found in {path}")
+ raise FileNotFoundError(
+ f"Folder not found: {path}. Is it correct?"
+ )
+
+ for file in self.all_files:
+ logging.info(
+ f"Filename found and stored: {file.name}, "
+ + f"its path is {file.path}"
+ )
 
- def check_if_folder_exists(self) -> bool:
- """Checks if the folder containing the experimental data exists.
- The folder path is obtained by calling the method :meth:`get_path`.
+ def search_file_paths(self, path: Path) -> None:
+ """Recursively searches files in the given folder.
+ Saves file path in self.all_files only if the DataType is found.
 
- Returns
- -------
- bool
- True if folder exists, False otherwise
+ Parameters
+ ----------
+ path : Path
+ Path to the folder to be searched
  """
- return self.get_path().exists()
-
- def extract_all_file_names(self) -> list:
- # get filenames by day
- # search for files called 'suite2p', 'plane0', 'Fall.mat'
- # get session names to get name of stim files
- # corrects for exceptions
- raise NotImplementedError("This method is not implemented yet")
+ for i in path.glob("**/*"):
+ file = File(i.name, i)
+ if not file.datatype == DataType.NOT_FOUND:
+ logging.info(f"File path stored: {file}")
+ self.all_files.append(file)
+ else:
+ logging.info(f"File path NOT stored: {file}")

load_suite2p/objects/parsers2p/parser2p.py

@@ -51,13 +51,37 @@ def _parse(self) -> dict:
  pass
 
  @abstractmethod
- def get_path(self) -> Path:
+ def get_path_to_experimental_folder(self) -> Path:
  """Returns the path to the file containing the suite2p output. To be
  implemented by the child classes taking into account the folder
  structure of each project.
  """
  pass
 
+ @abstractmethod
+ def get_path_to_allen_dff_file(self) -> Path:
+ """Returns the path to the file containing the allen dff. To be
+ implemented by the child classes taking into account the folder
+ structure of each project.
+ """
+ pass
+
+ @abstractmethod
+ def get_path_to_serial2p(self) -> Path:
+ """Returns the path to the file containing the serial2p output. To be
+ implemented by the child classes taking into account the folder
+ structure of each project.
+ """
+ pass
+
+ @abstractmethod
+ def get_path_to_stimulus_analog_input_schedule_files(self) -> Path:
+ """Returns the path to the file containing the stimulus AI schedule
+ files. To be implemented by the child classes taking into account the
+ folder structure of each project.
+ """
+ pass
+
  def _minimum_params_required(self) -> bool:
  """Checks if the minimum parameters have been evaluated by the parser.
 

load_suite2p/objects/parsers2p/parser2pRSP.py

@@ -91,13 +91,13 @@ def _parse(self) -> dict:
  info["cre"] = item
  elif "monitor" == item:
  info["monitor_position"] = item
- if hasattr(self, "monitor_position"):
- info["monitor_position"] += item
+ if "monitor_position" in info and item != "monitor":
+ info["monitor_position"] += "_" + item
 
  if "monitor" not in info["monitor_position"]:
  logging.debug(
  "Monitor position not found in folder name",
- extra={"ChryssanthiParser": self},
+ extra={"Parser2pRSP": self},
  )
  logging.debug(info["monitor_position"])
  raise RuntimeError("Monitor position not found in folder name")
@@ -117,7 +117,7 @@ def _get_parent_folder_name(self) -> str:
  """
  return f'{self.info["mouse_line"]}_{self.info["mouse_id"]}'
 
- def get_path(self) -> Path:
+ def get_path_to_experimental_folder(self) -> Path:
  """Returns the path to the folder containing the experimental data.
  Reads the server location from the config file and appends the
  parent folder and the given folder name.
@@ -128,3 +128,33 @@ def get_path(self) -> Path:
  / Path(self._get_parent_folder_name())
  / Path(self._folder_name)
  )
+
+ def get_path_to_allen_dff_file(self) -> Path:
+ """Returns the path to the folder containing the allen dff files.
+ Reads the server location from the config file and appends the
+ parent folder and the given folder name.
+ """
+ filename = self._folder_name + "_sf_tf_allen_dff.mat"
+
+ return Path(self._config["paths"]["allen-dff"]) / Path(filename)
+
+ def get_path_to_serial2p(self) -> Path:
+ """Returns the path to the folder containing the serial2p files.
+ Reads the server location from the config file and appends the
+ parent folder and the given folder name.
+ """
+
+ return Path(self._config["paths"]["serial2p"]) / Path(
+ "CT_" + self._get_parent_folder_name()
+ )
+
+ def get_path_to_stimulus_analog_input_schedule_files(self) -> Path:
+ """Returns the path to the folder containing the stimulus
+ AI schedule files.
+ Reads the server location from the config file and appends the
+ parent folder and the given folder name.
+ """
+
+ return Path(self._config["paths"]["stimulus-ai-schedule"]) / Path(
+ self._folder_name
+ )

load_suite2p/objects/specifications.py

@@ -11,5 +11,5 @@ def __init__(self, config: dict, folder_name: str):
  self.base_paths: dict = config["paths"]
  self.folder_name = folder_name
  self.folder_naming_specs = FolderNamingSpecs(folder_name, config)
- self.all_filenames = self.folder_naming_specs.extract_all_file_names()
+ self.folder_naming_specs.extract_all_file_names()
  self.options = Options(config)

load_suite2p/utils.py

@@ -1,3 +1,4 @@
+import logging
 import sys
 
 import rich
@@ -34,7 +35,7 @@ def inner_function(*args, **kwargs):
  func(*args, **kwargs)
  except Exception as e:
  rich.print("Something went wrong 😱")
- rich.print(e)
+ logging.exception(e)
 
  return inner_function