Improve subject configuration structure

dicom-archive/database_config_template.py

@@ -1,43 +1,53 @@
 #!/usr/bin/env python
 
 import re
+from lib.database import Database
 from lib.imaging import Imaging
+from lib.config_file import CreateVisitInfo, DatabaseConfig, S3Config, SubjectInfo
 
-mysql = {
-    'host'    : 'DBHOST',
-    'username': 'DBUSER',
-    'passwd'  : 'DBPASS',
-    'database': 'DBNAME',
-    'port'    : ''
-}
 
-s3 = {
-    'aws_access_key_id'    : 'AWS_ACCESS_KEY_ID',
-    'aws_secret_access_key': 'AWS_SECRET_ACCESS_KEY',
-    'aws_s3_endpoint_url'  : 'AWS_S3_ENDPOINT',
-    'aws_s3_bucket_name'   : 'AWS_S3_BUCKET_NAME',
-}
+mysql: DatabaseConfig = DatabaseConfig(
+    host     = 'DBHOST',
+    username = 'DBUSER',
+    password = 'DBPASS',
+    database = 'DBNAME',
+    port     = 3306,
+)
 
+# This statement can be omitted if the project does not use AWS S3.
+s3: S3Config = S3Config(
+    aws_access_key_id     = 'AWS_ACCESS_KEY_ID',
+    aws_secret_access_key = 'AWS_SECRET_ACCESS_KEY',
+    aws_s3_endpoint_url   = 'AWS_S3_ENDPOINT',
+    aws_s3_bucket_name    = 'AWS_S3_BUCKET_NAME',
+)
 
-def get_subject_ids(db, dicom_value=None, scanner_id=None):
-
-    subject_id_dict = {}
 
+def get_subject_info(db: Database, subject_name: str, scanner_id: int | None = None) -> SubjectInfo | None:
     imaging = Imaging(db, False)
 
-    phantom_match   = re.search(r'(pha)|(test)', dicom_value, re.IGNORECASE)
-    candidate_match = re.search(r'([^_]+)_(\d+)_([^_]+)', dicom_value, re.IGNORECASE)
+    phantom_match   = re.search(r'(pha)|(test)', subject_name, re.IGNORECASE)
+    candidate_match = re.search(r'([^_]+)_(\d+)_([^_]+)', subject_name, re.IGNORECASE)
 
     if phantom_match:
-        subject_id_dict['isPhantom']  = True
-        subject_id_dict['CandID']     = imaging.get_scanner_candid(scanner_id)
-        subject_id_dict['visitLabel'] = dicom_value.strip()
-        subject_id_dict['createVisitLabel'] = 1
+        return SubjectInfo.from_phantom(
+            name         = subject_name,
+            # Pass the scanner candidate CandID. If the scanner candidate does not exist in the
+            # database yet, create it in this function.
+            cand_id      = imaging.get_scanner_candid(scanner_id),
+            visit_label  = subject_name.strip(),
+            create_visit = CreateVisitInfo(
+               project_id = 1, # Change to relevant project ID
+               cohort_id  = 1, # Change to relevant cohort ID
+            ),
+        )
     elif candidate_match:
-        subject_id_dict['isPhantom']  = False
-        subject_id_dict['PSCID']      = candidate_match.group(1)
-        subject_id_dict['CandID']     = candidate_match.group(2)
-        subject_id_dict['visitLabel'] = candidate_match.group(3)
-        subject_id_dict['createVisitLabel'] = 0
-
-    return subject_id_dict
+        return SubjectInfo.from_candidate(
+            name         = subject_name,
+            psc_id       = candidate_match.group(1),
+            cand_id      = int(candidate_match.group(2)),
+            visit_label  = candidate_match.group(3),
+            create_visit = None,
+        )
+
+    return None

pyproject.toml

@@ -17,7 +17,8 @@ include = [
     "python/tests",
     "python/lib/db",
     "python/lib/exception",
-    "python/lib/validate_subject_ids.py",
+    "python/lib/config_file.py",
+    "python/lib/validate_subject_info.py",
 ]
 typeCheckingMode = "strict"
 reportMissingTypeStubs = "none"

python/lib/config_file.py

@@ -0,0 +1,80 @@
+"""
+This module stores the classes used in the Python configuration file of LORIS-MRI.
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class DatabaseConfig:
+    """
+    Class wrapping the MariaDB / MySQL database access configuration.
+    """
+
+    host:     str
+    username: str
+    password: str
+    database: str
+    port:     int = 3306  # Default database port.
+
+
+@dataclass
+class S3Config:
+    """
+    Class wrapping AWS S3 access configuration.
+    """
+
+    aws_access_key_id:     str
+    aws_secret_access_key: str
+    aws_s3_endpoint_url:   str | None = None  # Can also be obtained from the database.
+    aws_s3_bucket_name:    str | None = None  # Can also be obtained from the database.
+
+
+@dataclass
+class CreateVisitInfo:
+    """
+    Class wrapping the parameters for automated visit creation (in the `Visit_Windows` table).
+    """
+
+    project_id: int
+    cohort_id:  int
+
+
+@dataclass
+class SubjectInfo:
+    """
+    Dataclass wrapping information about a subject configuration, including information about the
+    candidate, the visit label, and the automated visit creation (or not).
+    """
+
+    # The name of the subject may be either the DICOM's PatientName or PatientID depending on the
+    # LORIS configuration.
+    name: str
+    is_phantom: bool
+    # For a phantom scan, the PSCID is 'scanner'.
+    psc_id: str
+    # For a phantom scan, the CandID is that of the scanner.
+    cand_id: int
+    visit_label: str
+    # `CreateVisitInfo` means that a visit can be created automatically using the parameters
+    # provided, `None` means that the visit needs to already exist in the database.
+    create_visit: CreateVisitInfo | None
+
+    @staticmethod
+    def from_candidate(
+        name: str,
+        psc_id: str,
+        cand_id: int,
+        visit_label: str,
+        create_visit: CreateVisitInfo | None,
+    ):
+        return SubjectInfo(name, False, psc_id, cand_id, visit_label, create_visit)
+
+    @staticmethod
+    def from_phantom(
+        name: str,
+        cand_id: int,
+        visit_label: str,
+        create_visit: CreateVisitInfo | None,
+    ):
+        return SubjectInfo(name, True, 'scanner', cand_id, visit_label, create_visit)

python/lib/database.py

@@ -5,6 +5,7 @@
 import MySQLdb
 
 import lib.exitcode
+from lib.config_file import DatabaseConfig
 
 __license__ = "GPLv3"
 
@@ -62,25 +63,22 @@ class Database:
         db.disconnect()
     """
 
-    def __init__(self, credentials, verbose):
+    def __init__(self, config: DatabaseConfig, verbose: bool):
         """
         Constructor method for the Database class.
 
-        :param credentials: LORIS database credentials
-         :type credentials: dict
-        :param verbose    : whether to be verbose or not
-         :type verbose    : bool
+        :param config:  LORIS database credentials
+        :param verbose: whether to be verbose or not
         """
 
         self.verbose = verbose
 
         # grep database credentials
-        default_port   = 3306
-        self.db_name   = credentials['database']
-        self.user_name = credentials['username']
-        self.password  = credentials['passwd']
-        self.host_name = credentials['host']
-        port           = credentials['port']
+        self.db_name   = config.database
+        self.user_name = config.username
+        self.password  = config.password
+        self.host_name = config.host
+        self.port      = config.port
 
         if not self.user_name:
             raise Exception("\nUser name cannot be empty string.\n")
@@ -89,8 +87,6 @@ def __init__(self, credentials, verbose):
         if not self.host_name:
             raise Exception("\nDatabase host cannot be empty string.\n")
 
-        self.port = int(port) if port else default_port
-
     def connect(self):
         """
         Attempts to connect to the database using the connection parameters

python/lib/database_lib/candidate_db.py

@@ -38,7 +38,7 @@ def __init__(self, db, verbose):
         self.verbose = verbose
 
     @deprecated('Use `lib.db.query.candidate.try_get_candidate_with_cand_id` instead')
-    def get_candidate_psc_id(self, cand_id: str | int) -> str | None:
+    def get_candidate_psc_id(self, cand_id: int) -> str | None:
         """
         Return a candidate PSCID and based on its CandID, or `None` if no candidate is found in
         the database.

python/lib/db/connect.py

@@ -1,18 +1,21 @@
-from typing import Any
 from urllib.parse import quote
 
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 
-default_port = 3306
+from lib.config_file import DatabaseConfig
 
 
-def connect_to_db(credentials: dict[str, Any]):
-    host     = credentials['host']
-    port     = credentials['port']
-    username = quote(credentials['username'])
-    password = quote(credentials['passwd'])
-    database = credentials['database']
-    port     = int(port) if port else default_port
+def connect_to_database(config: DatabaseConfig):
+    """
+    Connect to the database and get an SQLAlchemy session to interract with it using the provided
+    credentials.
+    """
+
+    host     = config.host
+    port     = config.port
+    username = quote(config.username)
+    password = quote(config.password)
+    database = config.database
     engine = create_engine(f'mysql+mysqldb://{username}:{password}@{host}:{port}/{database}')
     return Session(engine)