new env log

pyproject.toml

@@ -16,7 +16,11 @@ select = ["E", "W"]
 include = [
     "python/tests",
     "python/lib/db",
+    "python/lib/dataclass",
     "python/lib/exception",
+    "python/lib/file_system.py",
+    "python/lib/log.py",
+    "python/lib/make_env.py",
     "python/lib/validate_subject_ids.py",
 ]
 typeCheckingMode = "strict"

python/extract_eeg_bids_archive.py

@@ -6,11 +6,12 @@
 import sys
 import re
 from lib.lorisgetopt import LorisGetOpt
-from lib.imaging_io import ImagingIO
 from lib.database import Database
 from lib.database_lib.config import Config
 from lib.exitcode import SUCCESS, BAD_CONFIG_SETTING
-from lib.log import Log
+from lib.file_system import copy_file, delete_dir, extract_archive
+from lib.make_env import make_env
+from lib.log import log, log_error, log_error_exit, log_verbose, log_warning
 import lib.utilities as utilities
 
 __license__ = "GPLv3"
@@ -79,16 +80,7 @@ def main():
     # and create the log object (their basename being the name of the script run)
     # ---------------------------------------------------------------------------------------------
     tmp_dir = loris_getopt_obj.tmp_dir
-    data_dir = config_db_obj.get_config("dataDirBasepath")
-    log_obj = Log(
-        db,
-        data_dir,
-        script_name,
-        os.path.basename(tmp_dir),
-        loris_getopt_obj.options_dict,
-        verbose
-    )
-    imaging_io_obj = ImagingIO(log_obj, verbose)
+    env = make_env(loris_getopt_obj)
 
     # ---------------------------------------------------------------------------------------------
     # Grep config settings from the Config module
@@ -116,7 +108,7 @@ def main():
         eeg_archives_list = db.pselect(query, ())
 
     if not eeg_archives_list:
-        print('No new EEG upload to extract.')
+        log(env, "No new EEG upload to extract.")
         sys.exit(SUCCESS)
 
     # ---------------------------------------------------------------------------------------------
@@ -133,31 +125,24 @@ def main():
             try:
                 s3_obj.download_file(eeg_archive_path, eeg_archive_local_path)
             except Exception as err:
-                imaging_io_obj.log_info(
-                    f"{eeg_archive_path} could not be downloaded from S3 bucket. Error was\n{err}",
-                    is_error=True,
-                    is_verbose=False,
-                )
+                log_error(env, f"{eeg_archive_path} could not be downloaded from S3 bucket. Error was\n{err}")
                 error = True
             else:
                 eeg_archive_path = eeg_archive_local_path
 
         elif eeg_incoming_dir.startswith('s3://'):
-            imaging_io_obj.log_error_and_exit(
+            log_error_exit(
+                env,
                 f"{eeg_incoming_dir} is a S3 path but S3 server connection could not be established.",
-                BAD_CONFIG_SETTING
+                BAD_CONFIG_SETTING,
             )
 
         if not error:
             try:
                 # Uncompress archive in tmp location
-                eeg_collection_path = imaging_io_obj.extract_archive(eeg_archive_path, 'EEG', tmp_dir)
+                eeg_collection_path = extract_archive(env, eeg_archive_path, 'EEG', tmp_dir)
             except Exception as err:
-                imaging_io_obj.log_info(
-                    f"Could not extract {eeg_archive_path} - {format(err)}",
-                    is_error=True,
-                    is_verbose=False,
-                )
+                log_error(env, f"Could not extract {eeg_archive_path} - {format(err)}")
                 error = True
 
         if not error:
@@ -173,20 +158,12 @@ def main():
                     if eeg_session_rel_path_re:
                         eeg_session_rel_path = eeg_session_rel_path_re.group()
                     else:
-                        imaging_io_obj.log_info(
-                            f"Could not find a subject folder in the BIDS structure for {eeg_archive_file}.",
-                            is_error=True,
-                            is_verbose=False,
-                        )
+                        log_error(env, f"Could not find a subject folder in the BIDS structure for {eeg_archive_file}.")
                         error = True
                         break
 
         if not error and not tmp_eeg_session_path:
-            imaging_io_obj.log_info(
-                "Could not find a session folder in the bids structure for .",
-                is_error=True,
-                is_verbose=False,
-            )
+            log_error(env, "Could not find a session folder in the bids structure for .")
             error = True
 
         if not error:
@@ -206,9 +183,7 @@ def main():
 
                     file_paths_updated = utilities.update_set_file_path_info(set_full_path, width_fdt_file)
                     if not file_paths_updated:
-                        message = "WARNING: cannot update the set file " \
-                                  + os.path.basename(set_full_path) + " path info"
-                        print(message)
+                        log_warning(env, f"Cannot update the set file {os.path.basename(set_full_path)} path info")
 
                 s3_data_dir = config_db_obj.get_config("EEGS3DataPath")
                 if s3_obj and s3_data_dir and s3_data_dir.startswith('s3://'):
@@ -225,11 +200,11 @@ def main():
                         # Move folder in S3 bucket
                         s3_obj.upload_dir(tmp_eeg_modality_path, s3_data_eeg_modality_path)
                     except Exception as err:
-                        imaging_io_obj.log_info(
+                        log_error(
+                            env,
                             f"{tmp_eeg_modality_path} could not be uploaded to the S3 bucket. Error was\n{err}",
-                            is_error=True,
-                            is_verbose=False,
                         )
+
                         error = True
                 else:
                     assembly_bids_path = config_db_obj.get_config("EEGAssemblyBIDS")
@@ -239,16 +214,14 @@ def main():
 
                     data_eeg_modality_path = os.path.join(assembly_bids_path, eeg_session_rel_path, modality)
 
-                    """
-                    If the suject/session/modality BIDS data already exists
-                    on the destination folder, delete if first
-                    copying the data
-                    """
-                    imaging_io_obj.remove_dir(data_eeg_modality_path)
-                    imaging_io_obj.copy_file(tmp_eeg_modality_path, data_eeg_modality_path)
+                    # If the suject/session/modality BIDS data already exists
+                    # on the destination folder, delete if first
+                    # copying the data
+                    delete_dir(env, data_eeg_modality_path)
+                    copy_file(env, tmp_eeg_modality_path, data_eeg_modality_path)
 
         # Delete tmp location
-        imaging_io_obj.remove_dir(tmp_dir)
+        delete_dir(env, tmp_dir)
 
         if not error:
             # Set Status = Extracted

python/lib/dataclass/env.py

@@ -0,0 +1,66 @@
+from dataclasses import dataclass
+from typing import Callable
+from sqlalchemy.orm import Session as Database
+
+from lib.db.query.notification import get_notification_type_with_name
+
+
+@dataclass
+class NotifInfo:
+    """
+    This class wraps information used to send the script logs to the database.
+    """
+
+    # Noification type ID
+    type_id: int
+    # Notification origin, which is usually the script name
+    origin: str
+    # Process ID, which is usually the MRI upload ID
+    process_id: int
+
+
+@dataclass
+class Env:
+    """
+    This class wraps information about the environmentin which a LORIS-MRI script is executed. It
+    notably stores the database handle and various information used for logging.
+    """
+
+    db: Database
+    script_name: str
+    log_file: str
+    verbose: bool
+    cleanups: list[Callable[[], None]] = []
+    notif_info: NotifInfo | None = None
+
+    def add_cleanup(self, cleanup: Callable[[], None]):
+        """
+        Add a cleanup function to the environment, which will be executed if the program exits
+        early.
+        """
+
+        self.cleanups.append(cleanup)
+
+    def run_cleanups(self):
+        """
+        Run all the cleanup functions of the environment in the reverse insertion order (most
+        recent is ran first). This clears the cleanup functions list.
+        """
+
+        while self.cleanups != []:
+            cleanup = self.cleanups.pop()
+            cleanup()
+
+    def set_process_id(self, process_id: int):
+        """
+        Associate the current script with a given process ID, which notably allows to start logging
+        execution information in the database.
+        """
+
+        notification_type_name = f'PYTHON {self.script_name.replace("_", " ").upper()}'
+        notification_type = get_notification_type_with_name(self.db, notification_type_name)
+        self.notif_info = NotifInfo(
+            notification_type.id,
+            f'{self.script_name}.py',
+            process_id,
+        )

python/lib/db/decorator/y_n_bool.py

@@ -0,0 +1,31 @@
+from sqlalchemy import String
+from sqlalchemy.engine import Dialect
+from sqlalchemy.types import TypeDecorator
+
+
+class YNBool(TypeDecorator[bool]):
+    """
+    Decorator for a database yes/no type.
+    In SQL, the type will appear as 'Y' | 'N'.
+    In Python, the type will appear as a boolean.
+    """
+
+    impl = String
+
+    def process_bind_param(self, value: bool | None, dialect: Dialect):
+        match value:
+            case True:
+                return 'Y'
+            case False:
+                return 'N'
+            case None:
+                return None
+
+    def process_result_value(self, value: str | None, dialect: Dialect):
+        match value:
+            case 'Y':
+                return True
+            case 'N':
+                return False
+            case _:
+                return None

python/lib/db/model/config.py

@@ -0,0 +1,15 @@
+from typing import Optional
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from lib.db.base import Base
+import lib.db.model.config_setting as db_config_setting
+
+
+class DbConfig(Base):
+    __tablename__ = 'Config'
+
+    id         : Mapped[int]           = mapped_column('ID', primary_key=True)
+    setting_id : Mapped[int]           = mapped_column('ConfigID', ForeignKey('ConfigSettings.ID'))
+    value      : Mapped[Optional[str]] = mapped_column('Value')
+
+    setting : Mapped['db_config_setting.DbConfigSetting'] = relationship('DbConfigSetting')

python/lib/db/model/config_setting.py

@@ -0,0 +1,17 @@
+from typing import Optional
+from sqlalchemy.orm import Mapped, mapped_column
+from lib.db.base import Base
+
+
+class DbConfigSetting(Base):
+    __tablename__ = 'ConfigSettings'
+
+    id             : Mapped[int]            = mapped_column('ID', primary_key=True)
+    name           : Mapped[str]            = mapped_column('Name')
+    description    : Mapped[Optional[str]]  = mapped_column('Description')
+    visible        : Mapped[Optional[bool]] = mapped_column('Visible')
+    allow_multiple : Mapped[Optional[bool]] = mapped_column('AllowMultiple')
+    data_type      : Mapped[Optional[str]]  = mapped_column('DataType')
+    parent_id      : Mapped[Optional[int]]  = mapped_column('Parent')
+    label          : Mapped[Optional[str]]  = mapped_column('Label')
+    order_number   : Mapped[Optional[int]]  = mapped_column('OrderNumber')

python/lib/db/model/notification_spool.py

@@ -0,0 +1,21 @@
+from datetime import datetime
+from typing import Optional
+from sqlalchemy.orm import Mapped, mapped_column
+from lib.db.base import Base
+from lib.db.decorator.y_n_bool import YNBool
+
+
+class DbNotificationSpool(Base):
+    __tablename__ = 'notification_spool'
+
+    id           : Mapped[int]                = mapped_column('NotificationID')
+    type_id      : Mapped[int]                = mapped_column('NotificationTypeID')
+    process_id   : Mapped[int]                = mapped_column('ProcessID')
+    time_spooled : Mapped[Optional[datetime]] = mapped_column('TimeSpooled')
+    message      : Mapped[Optional[str]]      = mapped_column('Message')
+    error        : Mapped[Optional[YNBool]]   = mapped_column('Error')
+    verbose      : Mapped[YNBool]             = mapped_column('Verbose')
+    sent         : Mapped[YNBool]             = mapped_column('Sent')
+    site_id      : Mapped[Optional[int]]      = mapped_column('CenterID')
+    origin       : Mapped[Optional[str]]      = mapped_column('Origin')
+    active       : Mapped[YNBool]             = mapped_column('Active')

python/lib/db/model/notification_type.py

@@ -0,0 +1,12 @@
+from typing import Optional
+from sqlalchemy.orm import Mapped, mapped_column
+from lib.db.base import Base
+
+
+class DbNotificationType(Base):
+    __tablename__ = 'notification_types'
+
+    id         : Mapped[int]            = mapped_column('NotificationTypeID')
+    name       : Mapped[str]            = mapped_column('Type')
+    private    : Mapped[Optional[bool]] = mapped_column('private')
+    description: Mapped[Optional[str]]  = mapped_column('Description')

python/lib/db/query/config.py

@@ -0,0 +1,16 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+from lib.db.model.config import DbConfig
+from lib.db.model.config_setting import DbConfigSetting
+
+
+def get_config_with_setting_name(db: Database, name: str):
+    """
+    Get a single configuration entry from the database using its configuration setting name, or
+    raise an exception if no entry or several entries are found.
+    """
+
+    return db.execute(select(DbConfig)
+        .join(DbConfig.setting)
+        .where(DbConfigSetting.name == name)
+    ).scalar_one()

python/lib/db/query/notification.py

@@ -0,0 +1,14 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+from lib.db.model.notification_type import DbNotificationType
+
+
+def get_notification_type_with_name(db: Database, name: str):
+    """
+    Get a notification type from the database using its configuration setting name, or raise an
+    exception if no notification type is found.
+    """
+
+    return db.execute(select(DbNotificationType)
+        .where(DbNotificationType.name == name)
+    ).scalar_one()