"""Configuration settings to pass throught the pipeline."""
# Authors: The Lightweight Pipeline developers
# SPDX-License-Identifier: BSD-3-Clause
import importlib
import logging
import os
import sys
from lw_pipeline import Pipeline_Exception
class _ConsoleFormatter(logging.Formatter):
"""Compact console formatting with full details for warnings/errors."""
def format(self, record):
if record.levelno <= logging.INFO:
return record.getMessage()
return f"{record.levelname}: {record.getMessage()}"
[docs]
class Config:
"""A class representing the configuration settings."""
config_file_path = None
"""The path to an external configuration file."""
[docs]
def __init__(self, config_file_path=None, verbose=False):
"""
Initialize the Config object.
Parameters
----------
config_file_path : str, optional
The path to the configuration file. If provided, the configuration
settings will be updated based on the variables defined in the file.
verbose : bool, optional
If True, print messages about the configuration file being used.
Default is False.
"""
self.loaded_config_files = []
self._config_auto_detected = False
# Determine config file path
if config_file_path:
self.config_file_path = os.path.abspath(config_file_path)
if not os.path.isfile(self.config_file_path):
raise FileNotFoundError(
f"Specified configuration file not found: {config_file_path}."
)
else:
default_config = os.path.join(os.getcwd(), "config.py")
self.config_file_path = (
default_config if os.path.isfile(default_config) else None
)
self._config_auto_detected = self.config_file_path is not None
# Load main config file
if self.config_file_path:
config_dir = os.path.dirname(self.config_file_path)
sys.path.insert(0, config_dir)
self._load_file_to_variables(self.config_file_path, verbose=verbose)
# Load local config file if it exists
config_basename, config_ext = os.path.splitext(
os.path.basename(self.config_file_path)
)
local_config_file = os.path.join(
config_dir, f"{config_basename}_local{config_ext}"
)
if os.path.isfile(local_config_file):
self._load_file_to_variables(local_config_file, verbose=verbose)
# Set up logging after configuration is loaded
self.setup_logging()
self._log_startup_messages(verbose=verbose)
[docs]
def setup_logging(self):
"""
Set up logging based on configuration settings.
This method configures the root logger with console and optionally
file handlers based on the log_level, log_to_file, and log_file settings.
"""
# Get the root logger
logger = logging.getLogger()
logger.setLevel(getattr(logging, self.log_level.upper()))
# Remove existing handlers to avoid duplicates
logger.handlers.clear()
# Create console handler
console_handler = logging.StreamHandler(stream=sys.stdout)
console_handler.setLevel(getattr(logging, self.log_level.upper()))
console_formatter = _ConsoleFormatter()
console_handler.setFormatter(console_formatter)
logger.addHandler(console_handler)
# Create file handler if requested
if self.log_to_file:
log_file_path = self._resolve_log_file_path()
# Ensure the directory exists
os.makedirs(os.path.dirname(log_file_path), exist_ok=True)
file_handler = logging.FileHandler(log_file_path)
file_handler.setLevel(getattr(logging, self.log_level.upper()))
file_formatter = logging.Formatter(self.log_format)
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)
logger.info(f"Logging to file: {log_file_path}")
def _resolve_log_file_path(self):
"""Resolve effective log file path from config/cwd and logging options."""
base_dir = (
os.path.dirname(self.config_file_path)
if self.config_file_path is not None
else os.getcwd()
)
if self.log_file is None:
return os.path.join(base_dir, "pipeline.log")
log_file_path = os.path.expanduser(self.log_file)
if not os.path.isabs(log_file_path):
log_file_path = os.path.join(base_dir, log_file_path)
return log_file_path
def _log_startup_messages(self, verbose=False):
"""Emit startup and configuration diagnostics via logger."""
logger = logging.getLogger(__name__)
if self.loaded_config_files:
source = "auto-detected" if self._config_auto_detected else "specified"
noun = "file" if len(self.loaded_config_files) == 1 else "files"
logger.info(
"Using configuration %s (%s): %s",
noun,
source,
", ".join(self.loaded_config_files),
)
else:
logger.info("No configuration file detected; using default settings.")
def _load_file_to_variables(self, file_path, verbose=False):
"""Load variables from a specified configuration file into the current class."""
module_name = os.path.splitext(os.path.basename(file_path))[0]
config_module = importlib.import_module(module_name)
# Update the current variables in the this class with the ones from
# the specified configuration file
vars(self).update(
{k: v for k, v in vars(config_module).items() if not k.startswith("_")}
)
self.check_steps_dir()
self.loaded_config_files.append(file_path)
[docs]
def check_steps_dir(self):
"""
Make sure steps dir is absolute.
Notes
-----
- If `config_file_path` is not `None`, the relative `steps_dir` is
resolved relative to the directory containing the configuration file.
- If `config_file_path` is `None`, the relative `steps_dir` is resolved
relative to the current working directory.
Parameters
----------
steps_dir : str
The directory path for steps, which will be converted to an absolute path.
config_file_path : str or None
The path to the configuration file, used to resolve relative paths.
"""
value = self.steps_dir
# check if steps dir is relative in that case make it relative to config file
# or the current working directory
if not os.path.isabs(value):
# check if there is an externatl config file or if default config is used
if self.config_file_path is not None:
value = os.path.join(os.path.dirname(self.config_file_path), value)
else:
value = os.path.join(os.getcwd(), value)
# make steps dir absolute
value = os.path.abspath(value)
self.steps_dir = value
[docs]
def ask(self, message, default="n"):
"""
Ask to do something, e.g. before potentially deleting data, etc.
Make sure to specify options, e.g. (y/n), in the message.
"""
if self.auto_response == "off":
try:
response = input(f"\u26a0 Question: {message}: ")
except EOFError:
# e.g. if not run interactively
raise Pipeline_Exception(
f"Could not obtain response to question: ({message}). "
"Make sure to specify auto_response in the config, or run "
"with --ignore-questions to use the default response."
)
return response
elif self.auto_response == "default":
return default
else:
return self.auto_response
[docs]
def set_variable_and_write_to_config_file(self, variable, value):
"""
Set a variable in this class and write to config file, if not defined there.
For safety, only allow to write variables that are not already set.
Args:
variable (str): The name of the variable to update.
value (mixed): The value to set the variable to.
"""
if hasattr(self, variable) and getattr(self, variable):
raise Pipeline_Exception(
"Cannot overwrite already set variable in configuration file."
)
return
if not self.config_file_path:
raise Pipeline_Exception("No configuration file specified .")
return
setattr(self, variable, value)
with open(self.config_file_path, "a") as f:
f.write(f"\n{variable} = {value}\n")
logging.getLogger(__name__).info(
"Configuration file updated: %s", self.config_file_path
)
[docs]
def get_version(self):
"""
Get a version of the pipeline by getting last commit hash from the git.
Cave: This only works if the pipeline is in a git repository.
If not, it will return "unknown".
"""
try:
import subprocess
# make sure to execute git commands in the root directory of the repository
root_dir = os.path.dirname(os.path.abspath(__file__))
git_hash = (
subprocess.check_output(
["git", "rev-parse", "--short", "HEAD"], cwd=root_dir
)
.strip()
.decode("utf-8")
)
version = f"git-{git_hash}"
except Exception:
version = "unknown"
return version
# general default variables
# -------------------------
steps_dir = "steps/"
"""Steps directory relative to config file or current working directory if no """
"""external config file is used."""
auto_response = "off"
"""Decide how questions are answered (off/y/n/default)"""
data_dir = os.path.join(os.path.expanduser("~"), "data")
"""Default data directory"""
bids_root = os.path.join(data_dir, "bids")
"""Root directory for BIDS formatted data"""
subjects = []
"""List of subjects to include in the pipeline processing. If empty list, \
include all subjects"""
sessions = []
"""List of sessions to include in the pipeline processing. If empty list, \
include all sessions"""
tasks = []
"""List of tasks to include in the pipeline processing. If empty list, \
include all tasks"""
# variables for PipelineData class
# --------------------------------
deriv_root = os.path.join(data_dir, "derivatives")
"""Root directory for derivatives"""
overwrite = False
"""Overwrite existing derivative files, if False they are skipped"""
overwrite_mode = "never"
"""
Overwrite mode for output files. Options:
- "always": Always overwrite existing files
- "never": Never overwrite, skip existing files
- "ask": Prompt user for each file
- "ifnewer": Overwrite if source file is newer than output
"""
outputs_to_generate = None
"""
List or dict specifying which outputs to generate.
- None: Generate all outputs enabled by default
- List of patterns: Generate outputs matching patterns (e.g., ["plot*", "stats"])
- Dict: Step-specific patterns (e.g., {"01": ["plot"], "02": ["*"]})
Supports wildcards via fnmatch (e.g., "*" for all, "plot*": all starting with plot)
"""
output_root = None
"""
Root directory for non-BIDS outputs.
If None, defaults to deriv_root.
"""
sidecar_auto_generate = True
"""Automatically generate sidecar JSON files for all outputs"""
output_profiling = False
"""Include timing and file size information in sidecar metadata"""
eeg_path = {}
"""Path to the eeg data which should be converted to BIDS
Structure: subject -> condition -> task -> list of eeg files (runs)
File names expected relative to data_dir"""
bids_acquisition = None
"""EEG information that should be included in the BIDS file"""
bids_datatype = "eeg"
"""BIDS datatype of the data created as derivatives in the pipeline"""
bids_extension = ".edf"
"""Extension of the BIDS files in the bids root directory"""
n_jobs = 1
"""Number of parallel jobs to run"""
# Logging configuration
# ---------------------
log_level = "INFO"
"""
Logging level for the pipeline.
Options: "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
"""
log_to_file = True
"""Whether to write logs to a file in addition to console output"""
log_file = None
"""
Path to the log file. If None and log_to_file is True,
defaults to 'pipeline.log' in the directory of the used config file.
If no config file is used, defaults to the current working directory.
If a relative path is provided, it is resolved relative to the used
config file directory (or current working directory if no config file
is used).
"""
log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
"""Format string for log messages"""
# default variables for conversion ...
# default variables preprocessing ...
# default values analysis ...