Migrate some EEG channel queries to typed database abstraction

python/lib/config.py

@@ -79,6 +79,84 @@ def get_dicom_archive_dir_path_config(env: Env) -> Path:
     return dicom_archive_dir_path
 
 
+def get_default_bids_visit_label_config(env: Env) -> str | None:
+    """
+    Get the default BIDS visit label from the in-database configuration.
+    """
+
+    return _try_get_config_value(env, 'default_bids_vl')
+
+
+def get_eeg_viz_enabled_config(env: Env) -> bool:
+    """
+    Get whether the EEG visualization is enabled from the in-database configuration.
+    """
+
+    eeg_viz_enabled = _try_get_config_value(env, 'useEEGBrowserVisualizationComponents')
+    return eeg_viz_enabled == 'true' or eeg_viz_enabled == '1'
+
+
+def get_eeg_chunks_dir_path_config(env: Env) -> Path | None:
+    """
+    Get the EEG chunks directory path configuration value from the in-database configuration.
+    """
+
+    eeg_chunks_path = _try_get_config_value(env, 'EEGChunksPath')
+    if eeg_chunks_path is None:
+        return None
+
+    eeg_chunks_path = Path(eeg_chunks_path)
+
+    if not eeg_chunks_path.is_dir():
+        log_error_exit(
+            env,
+            (
+                f"The configuration value for the LORIS EEG chunks directory path '{eeg_chunks_path}' does not refer to"
+                " an existing directory."
+            ),
+        )
+
+    if not os.access(eeg_chunks_path, os.R_OK) or not os.access(eeg_chunks_path, os.W_OK):
+        log_error_exit(
+            env,
+            f"Missing read or write permission on the LORIS EEG chunks directory '{eeg_chunks_path}'.",
+        )
+
+    return eeg_chunks_path
+
+
+def get_eeg_pre_package_download_dir_path_config(env: Env) -> Path | None:
+    """
+    Get the EEG pre-packaged download path configuration value from the in-database configuration.
+    """
+
+    eeg_pre_package_path = _try_get_config_value(env, 'prePackagedDownloadPath')
+    if eeg_pre_package_path is None:
+        return None
+
+    eeg_pre_package_path = Path(eeg_pre_package_path)
+
+    if not eeg_pre_package_path.is_dir():
+        log_error_exit(
+            env,
+            (
+                "The configuration value for the LORIS EEG pre-packaged download directory path"
+                f" '{eeg_pre_package_path}' does not refer to an existing directory."
+            ),
+        )
+
+    if not os.access(eeg_pre_package_path, os.R_OK) or not os.access(eeg_pre_package_path, os.W_OK):
+        log_error_exit(
+            env,
+            (
+                "Missing read or write permission on the LORIS EEG pre-packaged download directory"
+                f" '{eeg_pre_package_path}'."
+            ),
+        )
+
+    return eeg_pre_package_path
+
+
 def _get_config_value(env: Env, setting_name: str) -> str:
     """
     Get a configuration value from the database using a configuration setting name, or exit the
@@ -99,3 +177,13 @@ def _get_config_value(env: Env, setting_name: str) -> str:
         )
 
     return config.value
+
+
+def _try_get_config_value(env: Env, setting_name: str) -> str | None:
+    """
+    Get a configuration value from the database using a configuration setting name, or return
+    `None` if that configuration setting or value does not exist in the database.
+    """
+
+    config = try_get_config_with_setting_name(env.db, setting_name)
+    return config.value if config is not None else None

python/lib/db/models/physio_channel_type.py

@@ -6,6 +6,6 @@
 class DbPhysioChannelType(Base):
     __tablename__ = 'physiological_channel_type'
 
-    id                  : Mapped[int]        = mapped_column('PhysiologicalChannelTypeID', primary_key=True)
-    channel_type_name   : Mapped[str]        = mapped_column('ChannelTypeName')
-    channel_description : Mapped[str | None] = mapped_column('ChannelDescription')
+    id          : Mapped[int]        = mapped_column('PhysiologicalChannelTypeID', primary_key=True)
+    name        : Mapped[str]        = mapped_column('ChannelTypeName')
+    description : Mapped[str | None] = mapped_column('ChannelDescription')

python/lib/db/models/physio_event_file.py

@@ -24,8 +24,8 @@ class DbPhysioEventFile(Base):
     last_update    : Mapped[datetime]    = mapped_column('LastUpdate')
     last_written   : Mapped[datetime]    = mapped_column('LastWritten')
 
-    physio_file       : Mapped['db_physio_file.DbPhysioFile | None']                     = relationship('PhysiologicalFile')
-    project           : Mapped['db_project.DbProject | None']                            = relationship('Project')
-    imaging_file_type : Mapped['db_imaging_file_type.DbImagingFileType | None']          = relationship('ImagingFileTypes')
-    task_events       : Mapped[list['db_physio_task_event.DbPhysioTaskEvent']]           = relationship('PhysiologicalTaskEvent', back_populates='event_file')
-    event_parameters  : Mapped[list['db_physio_event_parameter.DbPhysioEventParameter']] = relationship('PhysiologicalEventParameter', back_populates='event_file')
+    physio_file       : Mapped['db_physio_file.DbPhysioFile | None']                     = relationship('DbPhysioFile')
+    project           : Mapped['db_project.DbProject | None']                            = relationship('DbProject')
+    imaging_file_type : Mapped['db_imaging_file_type.DbImagingFileType | None']          = relationship('DbImagingFileType')
+    task_events       : Mapped[list['db_physio_task_event.DbPhysioTaskEvent']]           = relationship('DbPhysioTaskEvent', back_populates='event_file')
+    event_parameters  : Mapped[list['db_physio_event_parameter.DbPhysioEventParameter']] = relationship('DbPhysioEventParameter', back_populates='event_file')

python/lib/db/models/physio_event_parameter.py

@@ -19,5 +19,5 @@ class DbPhysioEventParameter(Base):
     is_categorical : Mapped[bool | None] = mapped_column('isCategorical', YNBool)
     hed            : Mapped[str | None]  = mapped_column('HED')
 
-    event_file      : Mapped['db_physio_event_file.DbPhysioEventFile']                                             = relationship('PhysiologicalEventFile')
-    category_levels : Mapped[list['db_physio_event_parameter_category_level.DbPhysioEventParameterCategoryLevel']] = relationship('PhysiologicalEventParameterCategoryLevel', back_populates='event_parameter')
+    event_file      : Mapped['db_physio_event_file.DbPhysioEventFile']                                             = relationship('DbPhysioEventFile')
+    category_levels : Mapped[list['db_physio_event_parameter_category_level.DbPhysioEventParameterCategoryLevel']] = relationship('DbPhysioEventParameterCategoryLevel', back_populates='event_parameter')

python/lib/db/models/physio_event_parameter_category_level.py

@@ -14,4 +14,4 @@ class DbPhysioEventParameterCategoryLevel(Base):
     description        : Mapped[str | None] = mapped_column('Description')
     hed                : Mapped[str | None] = mapped_column('HED')
 
-    event_parameter: Mapped['db_physio_event_parameter.DbPhysioEventParameter'] = relationship('PhysiologicalEventParameter', back_populates='category_levels')
+    event_parameter: Mapped['db_physio_event_parameter.DbPhysioEventParameter'] = relationship('DbPhysioEventParameter', back_populates='category_levels')

python/lib/db/models/physio_file.py

@@ -4,9 +4,11 @@
 from sqlalchemy import ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+import lib.db.models.physio_channel as db_physio_channel
 import lib.db.models.physio_file_parameter as db_phyiso_file_parameter
 import lib.db.models.physio_modality as db_physio_modality
 import lib.db.models.physio_output_type as db_physio_output_type
+import lib.db.models.session as db_session
 from lib.db.base import Base
 from lib.db.decorators.string_path import StringPath
 
@@ -16,10 +18,10 @@ class DbPhysioFile(Base):
 
     id               : Mapped[int]             = mapped_column('PhysiologicalFileID', primary_key=True)
     modality_id      : Mapped[int | None]      = mapped_column('PhysiologicalModalityID', ForeignKey('physiological_modality.PhysiologicalModalityID'))
-    output_type_id   : Mapped[int ]            = mapped_column('PhysiologicalOutputTypeID', ForeignKey('physiological_output_type.PhysiologicalOutputTypeID'))
-    session_id       : Mapped[int ]            = mapped_column('SessionID')
-    insert_time      : Mapped[datetime]        = mapped_column('InsertTime')
-    file_type        : Mapped[str | None]      = mapped_column('FileType')
+    output_type_id   : Mapped[int]             = mapped_column('PhysiologicalOutputTypeID', ForeignKey('physiological_output_type.PhysiologicalOutputTypeID'))
+    session_id       : Mapped[int]             = mapped_column('SessionID', ForeignKey('session.ID'))
+    insert_time      : Mapped[datetime]        = mapped_column('InsertTime', default=datetime.now)
+    type             : Mapped[str | None]      = mapped_column('FileType')
     acquisition_time : Mapped[datetime | None] = mapped_column('AcquisitionTime')
     inserted_by_user : Mapped[str]             = mapped_column('InsertedByUser')
     path             : Mapped[Path]            = mapped_column('FilePath', StringPath)
@@ -28,4 +30,6 @@ class DbPhysioFile(Base):
 
     output_type : Mapped['db_physio_output_type.DbPhysioOutputType']       = relationship('DbPhysioOutputType')
     modality    : Mapped['db_physio_modality.DbPhysioModality | None']     = relationship('DbPhysioModality')
+    session     : Mapped['db_session.DbSession']                           = relationship('DbSession')
+    channels    : Mapped['db_physio_channel.DbPhysioChannel']              = relationship('DbPhysioChannel', back_populates='physio_file')
     parameters  : Mapped['db_phyiso_file_parameter.DbPhysioFileParameter'] = relationship('DbPhysioFileParameter', back_populates='file')

python/lib/db/models/physio_status_type.py

@@ -6,5 +6,5 @@
 class DbPhysioStatusType(Base):
     __tablename__ = 'physiological_status_type'
 
-    id             : Mapped[int] = mapped_column('PhysiologicalStatusTypeID', primary_key=True)
-    channel_status : Mapped[str] = mapped_column('ChannelStatus')
+    id   : Mapped[int] = mapped_column('PhysiologicalStatusTypeID', primary_key=True)
+    name : Mapped[str] = mapped_column('ChannelStatus')

python/lib/db/models/physio_task_event.py

@@ -26,5 +26,5 @@ class DbPhysioTaskEvent(Base):
     trial_type     : Mapped[str | None]     = mapped_column('TrialType')
     response_time  : Mapped[time | None]    = mapped_column('ResponseTime')
 
-    physio_file : Mapped['db_physio_file.DbPhysioFile']            = relationship('PhysiologicalFile')
-    event_file  : Mapped['db_physio_event_file.DbPhysioEventFile'] = relationship('PhysiologicalEventFile', back_populates='task_events')
+    physio_file : Mapped['db_physio_file.DbPhysioFile']            = relationship('DbPhysioFile')
+    event_file  : Mapped['db_physio_event_file.DbPhysioEventFile'] = relationship('DbPhysioEventFile', back_populates='task_events')

python/lib/db/queries/physio_channel.py

@@ -0,0 +1,27 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.physio_channel_type import DbPhysioChannelType
+from lib.db.models.physio_status_type import DbPhysioStatusType
+
+
+def try_get_channel_type_with_name(db: Database, name: str) -> DbPhysioChannelType | None:
+    """
+    Get a physiological channel type from the database using its name, or return `None` if no
+    physiological channel type is found.
+    """
+
+    return db.execute(select(DbPhysioChannelType)
+        .where(DbPhysioChannelType.name == name)
+    ).scalar_one_or_none()
+
+
+def try_get_status_type_with_name(db: Database, name: str) -> DbPhysioStatusType | None:
+    """
+    Get a physiological status type from the database using its name, or return `None` if no
+    physiological status type is found.
+    """
+
+    return db.execute(select(DbPhysioStatusType)
+        .where(DbPhysioStatusType.name == name)
+    ).scalar_one_or_none()

python/lib/eeg.py

@@ -8,11 +8,12 @@
 import lib.exitcode
 import lib.utilities as utilities
 from lib.candidate import Candidate
-from lib.database_lib.config import Config
+from lib.config import get_eeg_pre_package_download_dir_path_config, get_eeg_viz_enabled_config
 from lib.database_lib.physiological_event_archive import PhysiologicalEventArchive
 from lib.database_lib.physiological_event_file import PhysiologicalEventFile
 from lib.database_lib.physiological_modality import PhysiologicalModality
 from lib.database_lib.physiological_output_type import PhysiologicalOutputType
+from lib.env import Env
 from lib.physiological import Physiological
 from lib.scanstsv import ScansTSV
 from lib.session import Session
@@ -75,7 +76,7 @@ class Eeg:
         db.disconnect()
     """
 
-    def __init__(self, bids_reader, bids_sub_id, bids_ses_id, bids_modality, db,
+    def __init__(self, env: Env, bids_reader, bids_sub_id, bids_ses_id, bids_modality, db,
                  verbose, data_dir, default_visit_label, loris_bids_eeg_rel_dir,
                  loris_bids_root_dir, dataset_tag_dict, dataset_type):
         """
@@ -108,8 +109,7 @@ def __init__(self, bids_reader, bids_sub_id, bids_ses_id, bids_modality, db,
          :type dataset_type          : string
         """
 
-        # config
-        self.config_db_obj = Config(db, verbose)
+        self.env = env
 
         # load bids objects
         self.bids_reader   = bids_reader
@@ -281,7 +281,7 @@ def register_data(self, derivatives=False, detect=True):
         if not inserted_eegs:
             return
 
-        physiological = Physiological(self.db, self.verbose)
+        physiological = Physiological(self.env, self.db, self.verbose)
 
         for inserted_eeg in inserted_eegs:
             eeg_file_id        = inserted_eeg['file_id']
@@ -340,8 +340,8 @@ def register_data(self, derivatives=False, detect=True):
             )
 
             # create data chunks for React visualization
-            eeg_viz_enabled = self.config_db_obj.get_config("useEEGBrowserVisualizationComponents")
-            if eeg_viz_enabled == 'true' or eeg_viz_enabled == '1':
+            eeg_viz_enabled = get_eeg_viz_enabled_config(self.env)
+            if eeg_viz_enabled:
                 physiological.create_chunks_for_visualization(eeg_file_id, self.data_dir)
 
     def fetch_and_insert_eeg_files(self, derivatives=False, detect=True):
@@ -366,7 +366,7 @@ def fetch_and_insert_eeg_files(self, derivatives=False, detect=True):
         inserted_eegs = []
         # load the Physiological object that will be used to insert the
         # physiological data into the database
-        physiological = Physiological(self.db, self.verbose)
+        physiological = Physiological(self.env, self.db, self.verbose)
 
         if detect:
             # TODO if derivatives, grep the source file as well as the input file ID???
@@ -552,7 +552,7 @@ def fetch_and_insert_electrode_file(
 
         # load the Physiological object that will be used to insert the
         # physiological data into the database
-        physiological = Physiological(self.db, self.verbose)
+        physiological = Physiological(self.env, self.db, self.verbose)
 
         electrode_files = self.bids_layout.get_nearest(
             original_physiological_file_path,
@@ -662,7 +662,7 @@ def fetch_and_insert_channel_file(
 
         # load the Physiological object that will be used to insert the
         # physiological data into the database
-        physiological = Physiological(self.db, self.verbose)
+        physiological = Physiological(self.env, self.db, self.verbose)
 
         channel_file = self.bids_layout.get_nearest(
             original_physiological_file_path,
@@ -726,7 +726,7 @@ def fetch_and_insert_event_files(
 
         # load the Physiological object that will be used to insert the
         # physiological data into the database
-        physiological = Physiological(self.db, self.verbose)
+        physiological = Physiological(self.env, self.db, self.verbose)
 
         event_data_file = self.bids_layout.get_nearest(
             original_physiological_file_path,
@@ -886,7 +886,7 @@ def create_and_insert_archive(self, files_to_archive: list[str], archive_rel_nam
 
         # load the Physiological object that will be used to insert the
         # physiological archive into the database
-        physiological = Physiological(self.db, self.verbose)
+        physiological = Physiological(self.env, self.db, self.verbose)
 
         # check if archive is on the filesystem
         (archive_rel_name, archive_full_path) = self.get_archive_paths(archive_rel_name)
@@ -985,7 +985,7 @@ def create_and_insert_event_archive(self, files_to_archive: list[str], archive_r
         physiological_event_archive_obj.insert(eeg_file_id, blake2, archive_rel_name)
 
     def get_archive_paths(self, archive_rel_name):
-        package_path = self.config_db_obj.get_config("prePackagedDownloadPath")
+        package_path = get_eeg_pre_package_download_dir_path_config(self.env)
         if package_path:
             raw_package_dir = os.path.join(package_path, 'raw')
             os.makedirs(raw_package_dir, exist_ok=True)