-
Notifications
You must be signed in to change notification settings - Fork 51
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Improve subject configuration structure (#1150)
* refacor subject config * remove suspicious phantom code * change naming * fix rebase lints * cleaner database url * Cecile CandID path join bug fix Co-authored-by: Cécile Madjar <[email protected]> --------- Co-authored-by: Cécile Madjar <[email protected]>
- Loading branch information
1 parent
27f6ba9
commit 06c08e6
Showing
14 changed files
with
258 additions
and
256 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,18 +1,24 @@ | ||
from typing import Any | ||
from urllib.parse import quote | ||
|
||
from sqlalchemy import create_engine | ||
from sqlalchemy import URL, create_engine | ||
from sqlalchemy.orm import Session | ||
|
||
default_port = 3306 | ||
from lib.config_file import DatabaseConfig | ||
|
||
|
||
def connect_to_database(config: DatabaseConfig): | ||
""" | ||
Connect to the database and get an SQLAlchemy session to interract with it using the provided | ||
credentials. | ||
""" | ||
|
||
# The SQLAlchemy URL object notably escapes special characters in the configuration attributes | ||
url = URL.create( | ||
drivername = 'mysql+mysqldb', | ||
host = config.host, | ||
port = config.port, | ||
username = config.username, | ||
password = config.password, | ||
database = config.database, | ||
) | ||
|
||
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 | ||
engine = create_engine(f'mysql+mysqldb://{username}:{password}@{host}:{port}/{database}') | ||
engine = create_engine(url) | ||
return Session(engine) |
Oops, something went wrong.