Automated tests for run_nifti_insertion.py

python/lib/db/models/candidate.py

@@ -3,6 +3,8 @@
 from sqlalchemy import ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+import lib.db.models.mri_protocol_violated_scan as db_mri_protocol_violated_scan
+import lib.db.models.mri_violation_log as db_mri_violation_log
 import lib.db.models.project as db_project
 import lib.db.models.session as db_session
 import lib.db.models.site as db_site
@@ -45,3 +47,7 @@ class DbCandidate(Base):
         = relationship('DbSite')
     registration_project : Mapped['db_project.DbProject'] \
         = relationship('DbProject')
+    violated_scans       : Mapped[list['db_mri_protocol_violated_scan.DbMriProtocolViolatedScan']] \
+        = relationship('DbMriProtocolViolatedScan', back_populates='candidate')
+    violations_log       : Mapped[list['db_mri_violation_log.DbMriViolationLog']] \
+        = relationship('DbMriViolationLog', back_populates='candidate')

python/lib/db/models/dicom_archive.py

@@ -6,7 +6,9 @@
 
 import lib.db.models.dicom_archive_file as db_dicom_archive_file
 import lib.db.models.dicom_archive_series as db_dicom_archive_series
+import lib.db.models.mri_protocol_violated_scan as db_mri_protocol_violated_scan
 import lib.db.models.mri_upload as db_mri_upload
+import lib.db.models.mri_violation_log as db_mri_violation_log
 import lib.db.models.session as db_session
 from lib.db.base import Base
 
@@ -55,3 +57,7 @@ class DbDicomArchive(Base):
         = relationship('DbMriUpload', back_populates='dicom_archive')
     session     : Mapped[Optional['db_session.DbSession']] \
         = relationship('DbSession')
+    violated_scans: Mapped[list['db_mri_protocol_violated_scan.DbMriProtocolViolatedScan']] \
+        = relationship('DbMriProtocolViolatedScan', back_populates='archive')
+    violations_log       : Mapped[list['db_mri_violation_log.DbMriViolationLog']] \
+        = relationship('DbMriViolationLog', back_populates='archive')

python/lib/db/models/file.py

@@ -3,6 +3,7 @@
 from sqlalchemy import ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+import lib.db.models.parameter_file as db_parameter_file
 import lib.db.models.session as db_session
 from lib.db.base import Base
 
@@ -34,4 +35,6 @@ class DbFile(Base):
     acquisition_order_per_modality : Mapped[int | None]   = mapped_column('AcqOrderPerModality')
     acquisition_date               : Mapped[date | None]  = mapped_column('AcquisitionDate')
 
-    session : Mapped['db_session.DbSession'] = relationship('DbSession', back_populates='files')
+    session   : Mapped['db_session.DbSession'] = relationship('DbSession', back_populates='files')
+    parameters: Mapped[list['db_parameter_file.DbParameterFile']] \
+        = relationship('DbParameterFile', back_populates='file')

python/lib/db/models/mri_protocol_check.py

@@ -0,0 +1,28 @@
+from decimal import Decimal
+
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+import lib.db.models.mri_protocol_check_group as db_mri_protocol_check_group
+import lib.db.models.mri_scan_type as db_mri_scan_type
+from lib.db.base import Base
+
+
+class DbMriProtocolCheck(Base):
+    __tablename__ = 'mri_protocol_checks'
+
+    id                      : Mapped[int]            = mapped_column('ID', primary_key=True)
+    scan_type_id            : Mapped[int | None] \
+        = mapped_column('MriScanTypeID', ForeignKey('mri_scan_type.MriScanTypeID'))
+    severity                : Mapped[str | None]     = mapped_column('Severity')
+    header                  : Mapped[str | None]     = mapped_column('Header')
+    valid_min               : Mapped[Decimal | None] = mapped_column('ValidMin')
+    valid_max               : Mapped[Decimal | None] = mapped_column('ValidMax')
+    valid_regex             : Mapped[str | None]     = mapped_column('ValidRegex')
+    protocol_check_group_id : Mapped[int] \
+        = mapped_column('MriProtocolChecksGroupID', ForeignKey('mri_protocol_checks_group.MriProtocolChecksGroupID'))
+
+    scan_type            : Mapped['db_mri_scan_type.DbMriScanType'] \
+        = relationship('DbMriScanType', back_populates='protocol_checks')
+    protocol_check_group: Mapped['db_mri_protocol_check_group.DbMriProtocolCheckGroup'] \
+        = relationship('DbMriProtocolCheckGroup', back_populates='protocol_check')

python/lib/db/models/mri_protocol_check_group.py

@@ -0,0 +1,17 @@
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+import lib.db.models.mri_protocol_check as db_mri_protocol_check
+import lib.db.models.mri_violation_log as db_mri_violation_log
+from lib.db.base import Base
+
+
+class DbMriProtocolCheckGroup(Base):
+    __tablename__ = 'mri_protocol_checks_group'
+
+    id  : Mapped[int] = mapped_column('MriProtocolChecksGroupID', primary_key=True)
+    name: Mapped[str] = mapped_column('Name')
+
+    protocol_check: Mapped['db_mri_protocol_check.DbMriProtocolCheck'] \
+        = relationship('DbMriProtocolCheck', back_populates='protocol_check_group')
+    violations_log: Mapped['db_mri_violation_log.DbMriViolationLog'] \
+        = relationship('DbMriViolationLog', back_populates='protocol_check_group')

python/lib/db/models/mri_protocol_group.py

@@ -0,0 +1,14 @@
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+import lib.db.models.mri_protocol_violated_scan as db_mri_protocol_violated_scan
+from lib.db.base import Base
+
+
+class DbMriProtocolGroup(Base):
+    __tablename__ = 'mri_protocol_group'
+
+    id   : Mapped[int] = mapped_column('MriProtocolGroupID', primary_key=True)
+    name : Mapped[str] = mapped_column('Name')
+
+    violated_scans: Mapped['db_mri_protocol_violated_scan.DbMriProtocolViolatedScan'] \
+        = relationship('DbMriProtocolViolatedScan', back_populates='protocol_group')

python/lib/db/models/mri_protocol_violated_scan.py

@@ -0,0 +1,47 @@
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+import lib.db.models.candidate as db_candidate
+import lib.db.models.dicom_archive as db_dicom_archive
+import lib.db.models.mri_protocol_group as db_mri_protocol_group
+from lib.db.base import Base
+
+
+class DbMriProtocolViolatedScan(Base):
+    __tablename__ = 'mri_protocol_violated_scans'
+
+    id                       : Mapped[int]             = mapped_column('ID', primary_key=True)
+    candidate_id             : Mapped[int | None]      = mapped_column('CandidateID', ForeignKey('candidate.ID'))
+    pscid                    : Mapped[str | None]      = mapped_column('PSCID')
+    dicom_archive_id         : Mapped[int | None]      = mapped_column('TarchiveID', ForeignKey('tarchive.TarchiveID'))
+    time_run                 : Mapped[datetime | None] = mapped_column('time_run')
+    series_description       : Mapped[str | None]      = mapped_column('series_description')
+    minc_location            : Mapped[str | None]      = mapped_column('minc_location')
+    patient_name             : Mapped[str | None]      = mapped_column('PatientName')
+    tr_range                 : Mapped[str | None]      = mapped_column('TR_range')
+    te_range                 : Mapped[str | None]      = mapped_column('TE_range')
+    ti_range                 : Mapped[str | None]      = mapped_column('TI_range')
+    slice_thickness_range    : Mapped[str | None]      = mapped_column('slice_thickness_range')
+    xspace_range             : Mapped[str | None]      = mapped_column('xspace_range')
+    yspace_range             : Mapped[str | None]      = mapped_column('yspace_range')
+    zspace_range             : Mapped[str | None]      = mapped_column('zspace_range')
+    xstep_range              : Mapped[str | None]      = mapped_column('xstep_range')
+    ystep_range              : Mapped[str | None]      = mapped_column('ystep_range')
+    zstep_range              : Mapped[str | None]      = mapped_column('zstep_range')
+    time_range               : Mapped[str | None]      = mapped_column('time_range')
+    series_uid               : Mapped[str | None]      = mapped_column('SeriesUID')
+    image_type               : Mapped[str | None]      = mapped_column('image_type')
+    phase_encoding_direction : Mapped[str | None]      = mapped_column('PhaseEncodingDirection')
+    echo_number              : Mapped[str | None]      = mapped_column('EchoNumber')
+    protocol_group_id        : Mapped[int | None]      \
+        = mapped_column('MriProtocolGroupID', ForeignKey('mri_protocol_group.MriProtocolGroupID'))
+
+    candidate     : Mapped[Optional['db_candidate.DbCandidate']] \
+        = relationship('DbCandidate', back_populates='violated_scans')
+    archive       : Mapped[Optional['db_dicom_archive.DbDicomArchive']] \
+        = relationship('DbDicomArchive', back_populates='violated_scans')
+    protocol_group: Mapped['db_mri_protocol_group.DbMriProtocolGroup']\
+        = relationship('DbMriProtocolGroup', back_populates='violated_scans')

python/lib/db/models/mri_scan_type.py

@@ -0,0 +1,17 @@
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+import lib.db.models.mri_protocol_check as db_mri_protocol_check
+import lib.db.models.mri_violation_log as db_mri_violation_log
+from lib.db.base import Base
+
+
+class DbMriScanType(Base):
+    __tablename__ = 'mri_scan_type'
+
+    id   : Mapped[int] = mapped_column('MriScanTypeID', primary_key=True)
+    name : Mapped[str] = mapped_column('MriScanTypeName')
+
+    protocol_checks : Mapped[list['db_mri_protocol_check.DbMriProtocolCheck']] \
+        = relationship('DbMriProtocolCheck', back_populates='scan_type')
+    violations_log  : Mapped[list['db_mri_violation_log.DbMriViolationLog']] \
+        = relationship('DbMriViolationLog', back_populates='scan_type')

python/lib/db/models/mri_violation_log.py

@@ -0,0 +1,47 @@
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+import lib.db.models.candidate as db_candidate
+import lib.db.models.dicom_archive as db_dicom_archive
+import lib.db.models.mri_protocol_check_group as db_mri_protocol_check_group
+import lib.db.models.mri_scan_type as db_mri_scan_type
+from lib.db.base import Base
+
+
+class DbMriViolationLog(Base):
+    __tablename__ = 'mri_violations_log'
+
+    id                      : Mapped[int]              = mapped_column('LogID', primary_key=True)
+    time_run                : Mapped[datetime]         = mapped_column('TimeRun')
+    series_uid              : Mapped[str | None]       = mapped_column('SeriesUID')
+    dicom_archive_id        : Mapped[int | None]       \
+        = mapped_column('TarchiveID', ForeignKey('tarchive.TarchiveID'))
+    minc_file               : Mapped[str | None]       = mapped_column('MincFile')
+    patient_name            : Mapped[str | None]       = mapped_column('PatientName')
+    candidate_id            : Mapped[int | None]       = mapped_column('CandidateID', ForeignKey('candidate.ID'))
+    visit_label             : Mapped[str | None]       = mapped_column('Visit_label')
+    check_id                : Mapped[int | None]       = mapped_column('CheckID')
+    scan_type_id            : Mapped[int | None]       \
+        = mapped_column('MriScanTypeID', ForeignKey('mri_scan_type.MriScanTypeID'))
+    severity                : Mapped[str | None]       = mapped_column('Severity')
+    header                  : Mapped[str | None]       = mapped_column('Header')
+    value                   : Mapped[str | None]       = mapped_column('Value')
+    valid_range             : Mapped[str | None]       = mapped_column('ValidRange')
+    valid_regex             : Mapped[str | None]       = mapped_column('ValidRegex')
+    echo_time               : Mapped[float | None]     = mapped_column('EchoTime')
+    phase_encoding_direction: Mapped[str | None]       = mapped_column('PhaseEncodingDirection')
+    echo_number             : Mapped[str | None]       = mapped_column('EchoNumber')
+    protocol_check_group_id : Mapped[int | None]       \
+        = mapped_column('MriProtocolChecksGroupID', ForeignKey('mri_protocol_checks_group.MriProtocolChecksGroupID'))
+
+    archive              : Mapped[Optional['db_dicom_archive.DbDicomArchive']] \
+        = relationship('DbDicomArchive', back_populates='violations_log')
+    candidate            : Mapped[Optional['db_candidate.DbCandidate']] \
+        = relationship('DbCandidate', back_populates='violations_log')
+    scan_type            : Mapped[Optional['db_mri_scan_type.DbMriScanType']] \
+        = relationship('DbMriScanType', back_populates='violations_log')
+    protocol_check_group: Mapped[Optional['db_mri_protocol_check_group.DbMriProtocolCheckGroup']] \
+        = relationship('DbMriProtocolCheckGroup', back_populates='violations_log')

python/lib/db/models/parameter_file.py

@@ -1,13 +1,19 @@
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+import lib.db.models.file as db_file
+import lib.db.models.parameter_type as db_parameter_type
 from lib.db.base import Base
 
 
 class DbParameterFile(Base):
     __tablename__ = 'parameter_file'
 
-    id                : Mapped[int]        = mapped_column('ParameterFileID', primary_key=True)
-    file_id           : Mapped[int]        = mapped_column('FileID')
-    parameter_type_id : Mapped[int]        = mapped_column('ParameterTypeID')
-    value             : Mapped[str | None] = mapped_column('Value')
-    insert_time       : Mapped[int]        = mapped_column('InsertTime')
+    id          : Mapped[int]        = mapped_column('ParameterFileID', primary_key=True)
+    file_id     : Mapped[int]        = mapped_column('FileID', ForeignKey('files.FileID'))
+    type_id     : Mapped[int]        = mapped_column('ParameterTypeID', ForeignKey('parameter_type.ParameterTypeID'))
+    value       : Mapped[str | None] = mapped_column('Value')
+    insert_time : Mapped[int]        = mapped_column('InsertTime')
+
+    file: Mapped[list['db_file.DbFile']]              = relationship('DbFile', back_populates='parameters')
+    type: Mapped['db_parameter_type.DbParameterType'] = relationship('DbParameterType', back_populates='parameter_file')

python/lib/db/models/parameter_type.py

@@ -0,0 +1,24 @@
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+import lib.db.models.parameter_file as db_parameter_file
+from lib.db.base import Base
+
+
+class DbParameterType(Base):
+    __tablename__ = 'parameter_type'
+
+    id               : Mapped[int]          = mapped_column('ParameterTypeID', primary_key=True)
+    name             : Mapped[str]          = mapped_column('Name')
+    alias            : Mapped[str | None]   = mapped_column('Alias')
+    data_type        : Mapped[str | None]   = mapped_column('Type')
+    description      : Mapped[str | None]   = mapped_column('Description')
+    range_min        : Mapped[float | None] = mapped_column('RangeMin')
+    range_max        : Mapped[float | None] = mapped_column('RangeMax')
+    source_field     : Mapped[str | None]   = mapped_column('SourceField')
+    source_from      : Mapped[str | None]   = mapped_column('SourceFrom')
+    source_condition : Mapped[str | None]   = mapped_column('SourceCondition')
+    queryable        : Mapped[bool | None]  = mapped_column('Queryable')
+    is_file          : Mapped[bool | None]  = mapped_column('IsFile')
+
+    parameter_file: Mapped[list['db_parameter_file.DbParameterFile']] \
+        = relationship('DbParameterFile', back_populates='type')

python/lib/db/queries/file.py

@@ -0,0 +1,42 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.file import DbFile
+from lib.db.models.parameter_file import DbParameterFile
+from lib.db.models.parameter_type import DbParameterType
+
+
+def try_get_file_with_unique_combination(
+        db: Database,
+        series_uid: str,
+        echo_time: str | None,
+        echo_number: str | None,
+        phase_encoding_direction: str | None
+) -> DbFile | None:
+    """
+    Get a file from the database using its SeriesInstanceUID, or return `None` if
+    no file was found.
+    """
+
+    return db.execute(select(DbFile)
+        .where(DbFile.series_uid == series_uid)
+        .where(DbFile.echo_time == echo_time)
+        .where(DbFile.echo_number == echo_number)
+        .where(DbFile.phase_encoding_direction == phase_encoding_direction)
+    ).scalar_one_or_none()
+
+
+def try_get_parameter_value_with_file_id_parameter_name(
+        db: Database,
+        file_id: int,
+        parameter_name: str
+) -> DbParameterFile | None:
+    """
+    Get parameter value from file ID and parameter name, or return `None` if no entry was found
+    """
+
+    return db.execute(select(DbParameterFile)
+        .join(DbParameterFile.type)
+        .where(DbParameterType.name == parameter_name)
+        .where(DbParameterFile.file_id == file_id)
+    ).scalar_one_or_none()

python/lib/db/queries/mri_protocol_violated_scans.py

@@ -0,0 +1,24 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.mri_protocol_violated_scan import DbMriProtocolViolatedScan
+
+
+def try_get_protocol_violated_scans_with_unique_series_combination(
+        db: Database,
+        series_uid: str,
+        echo_time: str | None,
+        echo_number: str | None,
+        phase_encoding_direction: str | None
+) -> DbMriProtocolViolatedScan | None:
+    """
+    Get the protocol violated scans from the database using its SeriesInstanceUID, or return `None` if
+    no protocol violated scan was found.
+    """
+
+    return db.execute(select(DbMriProtocolViolatedScan)
+        .where(DbMriProtocolViolatedScan.series_uid == series_uid)
+        .where(DbMriProtocolViolatedScan.te_range == echo_time)
+        .where(DbMriProtocolViolatedScan.echo_number == echo_number)
+        .where(DbMriProtocolViolatedScan.phase_encoding_direction == phase_encoding_direction)
+    ).scalar_one_or_none()

python/lib/db/queries/mri_violations_log.py

@@ -0,0 +1,24 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.mri_violation_log import DbMriViolationLog
+
+
+def try_get_violations_log_with_unique_series_combination(
+        db: Database,
+        series_uid: str,
+        echo_time: str | None,
+        echo_number: str | None,
+        phase_encoding_direction: str | None
+) -> DbMriViolationLog | None:
+    """
+    Get the violations log from the database using its SeriesInstanceUID, or return `None` if
+    no violations log was found.
+    """
+
+    return db.execute(select(DbMriViolationLog)
+        .where(DbMriViolationLog.series_uid == series_uid)
+        .where(DbMriViolationLog.echo_time == echo_time)
+        .where(DbMriViolationLog.echo_number == echo_number)
+        .where(DbMriViolationLog.phase_encoding_direction == phase_encoding_direction)
+    ).scalar_one_or_none()

python/lib/dcm2bids_imaging_pipeline_lib/nifti_insertion_pipeline.py

@@ -635,7 +635,7 @@ def _register_violations_log(self, violations_list, file_rel_path):
             'TarchiveID': self.dicom_archive.id,
             'MincFile': file_rel_path,
             'PatientName': self.json_file_dict['PatientName'],
-            'CandID': self.session.candidate.cand_id,
+            'CandidateID': self.session.candidate.id,
             'Visit_label': self.session.visit_label,
             'MriScanTypeID': self.scan_type_id,
             'EchoTime': scan_param['EchoTime'] if 'EchoTime' in scan_param.keys() else None,

python/scripts/run_nifti_insertion.py

@@ -117,7 +117,7 @@ def input_error_checking(loris_getopt_obj):
     if not json_path and not scan_type:
         print(
             "[ERROR   ] a json_path or a loris_scan_type need to be provided in order"
-            "to determine the image file protocol.\n"
+            " to determine the image file protocol.\n"
         )
         sys.exit(lib.exitcode.MISSING_ARG)