DIE physiological.py DIE (electrodes import refactor) (PR 2)

.github/workflows/static.yml

@@ -82,4 +82,4 @@ jobs:
       uses: ./.github/actions/setup-python
 
     - name: Run Pytest
-      run: pytest python/tests/unit
+      run: pytest

docs/python/Tooling.md

@@ -1,42 +1,42 @@
-# LORIS-MRI Python tooling
+# LORIS Python tooling
 
 ## Virtual environment
 
 LORIS-MRI uses a Python virtual environment to manage its execution context and dependencies. To activate the virtual environment, use the command `source environment` in the LORIS-MRI root directory. The dependencies of the virtual environment are listed in the `pyproject.toml` file.
 
 ## Configuration
 
-The LORIS-MRI Python tools are configured in the `pyproject.toml` file in the LORIS-MRI root directory.
+The LORIS Python tools are configured in the `pyproject.toml` file in the LORIS-MRI root directory.
 
 ## Linting
 
-LORIS-MRI Python uses the Ruff linter to ensure a consistent coding style that adheres to the Python community guidelines.
+LORIS Python uses the Ruff linter to ensure a consistent coding style that adheres to the Python community guidelines.
 
 To run Ruff, use the following commands in the LORIS-MRI root directory:
 * `ruff check` to run the linter and report style errors.
 * `ruff check --fix` to automatically fix trivial linting errors such as unsorted imports.
 
 ## Type checking
 
-LORIS-MRI Python uses type hints and the Pyright type checker [^3] to improve the robustness and maintainability of the code.
+LORIS Python uses type hints and the Pyright type checker [^3] to improve the robustness and maintainability of the code.
 
 To run Pyright, use the following commands in the LORIS-MRI root directory:
-* `pyright` to run the strict type checker, which type checks modern LORIS-MRI Python type-hinted modules with a strict configuration.
-* `pyright --project test` to run the global type checker, which type checks all the LORIS-MRI Python modules including untyped legacy code with a (very) loose configuration.
+* `pyright` to run the strict type checker, which type checks modern LORIS Python type-hinted modules with a strict configuration.
+* `pyright --project test` to run the global type checker, which type checks all the LORIS Python modules including untyped legacy code with a (very) loose configuration.
 
 When interacting with legacy code or untyped libraries in modern modules, you can use `# type: ignore` or `cast(type, value)` to ignore type checking errors.
 
 [^3]: Although we use Pyright, Mypy is also an excellent alternative type checker.
 
 ## Testing
 
-LORIS-MRI Python uses the Pytest library to handle unit and integration tests.
+LORIS Python uses the Pytest library to handle unit and integration tests.
 
 ### Unit testing
 
-The LORIS-MRI Python unit tests are located in the `python/tests/unit` directory.
+The core LORIS Python unit tests are located in the `python/tests/unit` directory, but each Python subpackage can have its unit tests in its own `tests` directory.
 
-To run the LORIS-MRI Python unit tests, use the command `pytest` in the root LORIS-MRI directory.
+To run the LORIS Python unit tests, use the command `pytest` in the root LORIS-MRI directory or in a package directory to only run that package tests.
 
 ### Integration testing
 

pyproject.toml

@@ -112,7 +112,7 @@ exclude = [
     # DICOM to BIDS converter pipeline
     "python/loris_dicom_to_bids_converter",
     # Untyped EEG BIDS import code
-    "python/loris_bids_importer/src/loris_bids_importer/eeg",
+    "python/loris_bids_importer/src/loris_bids_importer/eeg/main.py",
     # Generated protocol buffer code
     "python/loris_ephys_chunker/src/loris_ephys_chunker/protocol_buffers",
 ]
@@ -131,14 +131,19 @@ locale = "en-us"
 
 [tool.typos.default.extend-words]
 Centre = "Centre"  # McGill Centre for Integrative Neuroscience
+Colour = "Colour"  # SQL column name in sex table
 HED = "HED"  # Hierarchical Event Descriptors
 ND = "ND"  # Image Type DICOM attribute
-Colour = "Colour"  # SQL column name in sex table
 
 [tool.typos.default.extend-identifiers]
 ba = "ba"  # A CLI option of dcm2niix
+CapTrak = "CapTrak"  # BrainVision CapTrak
 
 [tool.pytest.ini_options]
 # Integration tests are located in `python/tests/integration`, but they should be ran from the
 # appropriate Docker image (usually using GitHub Actions).
-testpaths = ["python/tests/unit"]
+addopts = ["--import-mode=importlib"]
+testpaths = [
+    "python/tests/unit",
+    "python/loris_*/tests",
+]

python/lib/database_lib/physiological_coord_system.py

@@ -8,10 +8,13 @@
 - physiological_coord_system_point_3d_rel
 """
 
+from typing_extensions import deprecated
+
 # from lib.point_3d import Point3D
 # from lib.database_lib.point_3d import Point3DDB
 
 
+@deprecated('Use `lib.db.models.physio_coord_system.DbPhysioCoordSystem` and `lib.physio.coord_system` instead.')
 class PhysiologicalCoordSystem:
 
     def __init__(self, db, verbose):

python/lib/database_lib/point_3d.py

@@ -1,8 +1,13 @@
 """This class performs database queries for point_3d table"""
 
+from typing_extensions import deprecated
+
 from lib.point_3d import Point3D
 
 
+@deprecated(
+    'Use `lib.db.models.point_3d.DbPoint3D` and `lib.physio.points.get_or_create_point` instead.'
+)
 class Point3DDB:
     def __init__(self, db, verbose):
         """
@@ -35,7 +40,7 @@
                     " AND Y = %s"
                     " AND Z = %s",
             args=(x, y, z,)
         )
         return Point3D(cp[0]['Point3DID'], x, y, z) if cp else None
 
     def grep_point_by_id(self, point_id: int):
@@ -51,9 +56,9 @@
                     "FROM point_3d "
                     "WHERE Point3DID = %s",
             args=(point_id,)
         )
         return Point3D(point_id, cp[0]['X'], cp[0]['Y'], cp[0]['Z']) if cp else None
 
     def insert_point(self, p: Point3D):
         """
         Wrapper for insert_point_by_coordinates.
@@ -83,9 +88,9 @@
             column_names = ('X', 'Y', 'Z'),
             values = (x, y, z),
             get_last_id = True
         )
         return Point3D(pid, x, y, z)
 
     def grep_or_insert_point(self, point: Point3D):
         """
         Wrapper around grep_or_insert_point_by_coordinates.

python/lib/db/models/physio_coord_system.py

@@ -3,6 +3,7 @@
 from sqlalchemy import ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+import lib.db.models.bids_file as db_bids_file
 import lib.db.models.physio_coord_system_name as db_physio_coord_system_name
 import lib.db.models.physio_coord_system_type as db_physio_coord_system_type
 import lib.db.models.physio_coord_system_unit as db_physio_coord_system_unit
@@ -20,8 +21,10 @@ class DbPhysioCoordSystem(Base):
     unit_id     : Mapped[int]         = mapped_column('UnitID', ForeignKey('physiological_coord_system_unit.PhysiologicalCoordSystemUnitID'))
     modality_id : Mapped[int]         = mapped_column('ModalityID', ForeignKey('physiological_modality.PhysiologicalModalityID'))
     file_path   : Mapped[Path | None] = mapped_column('FilePath', StringPath)
+    bids_file_id: Mapped[int | None]  = mapped_column('BidsFileID', ForeignKey('bids_file.ID', ondelete='SET NULL'))
 
     name     : Mapped['db_physio_coord_system_name.DbPhysioCoordSystemName'] = relationship('DbPhysioCoordSystemName')
     type     : Mapped['db_physio_coord_system_type.DbPhysioCoordSystemType'] = relationship('DbPhysioCoordSystemType')
     unit     : Mapped['db_physio_coord_system_unit.DbPhysioCoordSystemUnit'] = relationship('DbPhysioCoordSystemUnit')
     modality : Mapped['db_physio_modality.DbPhysioModality']                 = relationship('DbPhysioModality')
+    bids_file: Mapped['db_bids_file.DbBidsFile | None']                      = relationship('DbBidsFile')

python/lib/db/models/physio_coord_system_electrode.py

@@ -1,14 +1,36 @@
 from datetime import datetime
 
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+import lib.db.models.physio_coord_system as db_physio_coord_system
+import lib.db.models.physio_electrode as db_physio_electrode
+import lib.db.models.physio_file as db_physio_file
 from lib.db.base import Base
 
 
 class DbPhysioCoordSystemElectrode(Base):
     __tablename__ = 'physiological_coord_system_electrode_rel'
 
-    coord_system_id : Mapped[int]      = mapped_column('PhysiologicalCoordSystemID', primary_key=True)
-    electrode_id    : Mapped[int]      = mapped_column('PhysiologicalElectrodeID', primary_key=True)
-    physio_file_id  : Mapped[int]      = mapped_column('PhysiologicalFileID')
+    coord_system_id: Mapped[int] = mapped_column(
+        'PhysiologicalCoordSystemID',
+        ForeignKey('physiological_coord_system.PhysiologicalCoordSystemID'),
+        primary_key=True,
+    )
+
+    electrode_id: Mapped[int] = mapped_column(
+        'PhysiologicalElectrodeID',
+        ForeignKey('physiological_electrode.PhysiologicalElectrodeID'),
+        primary_key=True,
+    )
+
+    physio_file_id: Mapped[int] = mapped_column(
+        'PhysiologicalFileID',
+        ForeignKey('physiological_file.PhysiologicalFileID', ondelete='CASCADE'),
+    )
+
     insert_time     : Mapped[datetime] = mapped_column('InsertTime', default=datetime.now)
+
+    coord_system : Mapped['db_physio_coord_system.DbPhysioCoordSystem'] = relationship('DbPhysioCoordSystem')
+    electrode    : Mapped['db_physio_electrode.DbPhysioElectrode']      = relationship('DbPhysioElectrode')
+    physio_file  : Mapped['db_physio_file.DbPhysioFile']                = relationship('DbPhysioFile')

python/lib/db/models/physio_coord_system_point_3d.py

@@ -1,11 +1,27 @@
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
+import lib.db.models.physio_coord_system as db_physio_coord_system
+import lib.db.models.point_3d as db_point_3d
 from lib.db.base import Base
 
 
 class DbPhysioCoordSystemPoint3d(Base):
     __tablename__ = 'physiological_coord_system_point_3d_rel'
 
-    coord_system_id : Mapped[int]        = mapped_column('PhysiologicalCoordSystemID', primary_key=True)
-    point_3d_id     : Mapped[int]        = mapped_column('Point3DID', primary_key=True)
+    coord_system_id : Mapped[int]        = mapped_column(
+        'PhysiologicalCoordSystemID',
+        ForeignKey('physiological_coord_system.PhysiologicalCoordSystemID'),
+        primary_key=True,
+    )
+
+    point_3d_id: Mapped[int] = mapped_column(
+        'Point3DID',
+        ForeignKey('point_3d.Point3DID'),
+        primary_key=True,
+    )
+
     name            : Mapped[str | None] = mapped_column('Name')
+
+    coord_system : Mapped['db_physio_coord_system.DbPhysioCoordSystem'] = relationship('DbPhysioCoordSystem')
+    point        : Mapped['db_point_3d.DbPoint3D']                      = relationship('DbPoint3D')

python/lib/db/queries/physio_coord_system.py

@@ -0,0 +1,115 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.physio_coord_system import DbPhysioCoordSystem
+from lib.db.models.physio_coord_system_electrode import DbPhysioCoordSystemElectrode
+from lib.db.models.physio_coord_system_name import DbPhysioCoordSystemName
+from lib.db.models.physio_coord_system_point_3d import DbPhysioCoordSystemPoint3d
+from lib.db.models.physio_coord_system_type import DbPhysioCoordSystemType
+from lib.db.models.physio_coord_system_unit import DbPhysioCoordSystemUnit
+from lib.db.models.point_3d import DbPoint3D
+
+
+def try_get_coord_system_type_with_name(db: Database, name: str) -> DbPhysioCoordSystemType | None:
+    """
+    Get a coordinate system type using its name, or return `None` if no coordinate system was found.
+    """
+
+    return db.execute(select(DbPhysioCoordSystemType)
+        .where(DbPhysioCoordSystemType.name == name)
+    ).scalar_one_or_none()
+
+
+def try_get_coord_system_name_with_name(db: Database, name: str) -> DbPhysioCoordSystemName | None:
+    """
+    Get a coordinate system name using its name, or return `None` if no coordinate system was found.
+    """
+
+    return db.execute(select(DbPhysioCoordSystemName)
+        .where(DbPhysioCoordSystemName.name == name)
+    ).scalar_one_or_none()
+
+
+def try_get_coord_system_unit_with_symbol(db: Database, symbol: str) -> DbPhysioCoordSystemUnit | None:
+    """
+    Get a coordinate system unit using its symbol, or return `None` if no coordinate system was
+    found.
+    """
+
+    return db.execute(select(DbPhysioCoordSystemUnit)
+        .where(DbPhysioCoordSystemUnit.symbol == symbol)
+    ).scalar_one_or_none()
+
+
+def try_get_coord_system_unit_with_name(db: Database, name: str) -> DbPhysioCoordSystemUnit | None:
+    """
+    Get a coordinate system unit using its name, or return `None` if no coordinate system was found.
+    """
+
+    return db.execute(select(DbPhysioCoordSystemUnit)
+        .where(DbPhysioCoordSystemUnit.name == name)
+    ).scalar_one_or_none()
+
+
+def try_get_coord_system(
+    db: Database,
+    bids_info_id: int | None,
+    modality_id: int,
+    type_id: int,
+    name_id: int,
+    unit_id: int,
+) -> DbPhysioCoordSystem | None:
+    """
+    Get a coordinate system using its identifying fields, or return `None` if no coordinate system
+    was found.
+    """
+
+    return db.execute(select(DbPhysioCoordSystem).where(
+        DbPhysioCoordSystem.bids_file_id == bids_info_id,
+        DbPhysioCoordSystem.modality_id == modality_id,
+        DbPhysioCoordSystem.type_id == type_id,
+        DbPhysioCoordSystem.name_id == name_id,
+        DbPhysioCoordSystem.unit_id == unit_id,
+    )).scalar_one_or_none()
+
+
+def try_get_point_with_coordinates(db: Database, x: float | None, y: float | None, z: float | None) -> DbPoint3D | None:
+    """
+    Get a three-dimensional point using its coordinates, or return `None` if none was found.
+    """
+
+    return db.execute(select(DbPoint3D).where(
+        DbPoint3D.x == x,
+        DbPoint3D.y == y,
+        DbPoint3D.z == z,
+    )).scalar_one_or_none()
+
+
+def try_get_coord_system_point_relation(
+    db: Database,
+    coord_system_id: int,
+    point_id: int,
+) -> DbPhysioCoordSystemPoint3d | None:
+    """
+    Get a coordinate system point relation using its IDs, or return `None` if none was found.
+    """
+
+    return db.execute(select(DbPhysioCoordSystemPoint3d).where(
+        DbPhysioCoordSystemPoint3d.coord_system_id == coord_system_id,
+        DbPhysioCoordSystemPoint3d.point_3d_id == point_id,
+    )).scalar_one_or_none()
+
+
+def try_get_coord_system_electrode_relation(
+    db: Database,
+    coord_system_id: int,
+    electrode_id: int,
+) -> DbPhysioCoordSystemElectrode | None:
+    """
+    Get a coordinate system electrode relation using its IDs, or return `None` if none was found.
+    """
+
+    return db.execute(select(DbPhysioCoordSystemElectrode).where(
+        DbPhysioCoordSystemElectrode.coord_system_id == coord_system_id,
+        DbPhysioCoordSystemElectrode.electrode_id == electrode_id,
+    )).scalar_one_or_none()

python/lib/db/queries/physio_electrode.py

@@ -0,0 +1,45 @@
+from collections.abc import Sequence
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.physio_coord_system_electrode import DbPhysioCoordSystemElectrode
+from lib.db.models.physio_electrode import DbPhysioElectrode
+from lib.db.models.physio_electrode_material import DbPhysioElectrodeMaterial
+from lib.db.models.physio_electrode_type import DbPhysioElectrodeType
+
+
+def get_physio_electrodes_with_file_id(db: Database, physio_file_id: int) -> Sequence[DbPhysioElectrode]:
+    """
+    Get the electrodes associated with a physiological file.
+    """
+
+    return db.execute(
+        select(DbPhysioElectrode)
+        .join(
+            DbPhysioCoordSystemElectrode,
+            DbPhysioCoordSystemElectrode.electrode_id == DbPhysioElectrode.id,
+        )
+        .where(DbPhysioCoordSystemElectrode.physio_file_id == physio_file_id)
+        .distinct()
+    ).scalars().all()
+
+
+def try_get_electrode_type_with_name(db: Database, name: str) -> DbPhysioElectrodeType | None:
+    """
+    Get an electrode type using its name, or return `None` if no electrode type was found.
+    """
+
+    return db.execute(select(DbPhysioElectrodeType)
+        .where(DbPhysioElectrodeType.name == name)
+    ).scalar_one_or_none()
+
+
+def try_get_electrode_material_with_name(db: Database, name: str) -> DbPhysioElectrodeMaterial | None:
+    """
+    Get an electrode material using its name, or return `None` if no electrode material was found.
+    """
+
+    return db.execute(select(DbPhysioElectrodeMaterial)
+        .where(DbPhysioElectrodeMaterial.name == name)
+    ).scalar_one_or_none()