diff --git a/.github/workflows/static.yml b/.github/workflows/static.yml index 8e73e58f1..2a64ef47c 100644 --- a/.github/workflows/static.yml +++ b/.github/workflows/static.yml @@ -82,4 +82,4 @@ jobs: uses: ./.github/actions/setup-python - name: Run Pytest - run: pytest python/tests/unit + run: pytest diff --git a/docs/python/Tooling.md b/docs/python/Tooling.md index 02bd74e4c..0f0ac26a1 100644 --- a/docs/python/Tooling.md +++ b/docs/python/Tooling.md @@ -1,4 +1,4 @@ -# LORIS-MRI Python tooling +# LORIS Python tooling ## Virtual environment @@ -6,11 +6,11 @@ LORIS-MRI uses a Python virtual environment to manage its execution context and ## 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. @@ -18,11 +18,11 @@ To run Ruff, use the following commands in the LORIS-MRI root directory: ## 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. @@ -30,13 +30,13 @@ When interacting with legacy code or untyped libraries in modern modules, you ca ## 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 diff --git a/pyproject.toml b/pyproject.toml index 579160143..723ba4000 100644 --- a/pyproject.toml +++ b/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", +] diff --git a/python/lib/database_lib/physiological_coord_system.py b/python/lib/database_lib/physiological_coord_system.py index 614562002..38df81189 100644 --- a/python/lib/database_lib/physiological_coord_system.py +++ b/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): diff --git a/python/lib/database_lib/point_3d.py b/python/lib/database_lib/point_3d.py index 15adb2f5e..30334450d 100644 --- a/python/lib/database_lib/point_3d.py +++ b/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): """ diff --git a/python/lib/db/models/physio_coord_system.py b/python/lib/db/models/physio_coord_system.py index e28beb624..a73a51b98 100644 --- a/python/lib/db/models/physio_coord_system.py +++ b/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') diff --git a/python/lib/db/models/physio_coord_system_electrode.py b/python/lib/db/models/physio_coord_system_electrode.py index 29eb605c1..61420a0e9 100644 --- a/python/lib/db/models/physio_coord_system_electrode.py +++ b/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') diff --git a/python/lib/db/models/physio_coord_system_point_3d.py b/python/lib/db/models/physio_coord_system_point_3d.py index c045f7b4b..8c4d511d7 100644 --- a/python/lib/db/models/physio_coord_system_point_3d.py +++ b/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') diff --git a/python/lib/db/queries/physio_coord_system.py b/python/lib/db/queries/physio_coord_system.py new file mode 100644 index 000000000..efee91622 --- /dev/null +++ b/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() diff --git a/python/lib/db/queries/physio_electrode.py b/python/lib/db/queries/physio_electrode.py new file mode 100644 index 000000000..f93c37a18 --- /dev/null +++ b/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() diff --git a/python/lib/physio/coord_system.py b/python/lib/physio/coord_system.py new file mode 100644 index 000000000..ca98f318e --- /dev/null +++ b/python/lib/physio/coord_system.py @@ -0,0 +1,163 @@ +from lib.db.models.bids_file import DbBidsFile +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.physio_file import DbPhysioFile +from lib.db.models.physio_modality import DbPhysioModality +from lib.db.models.point_3d import DbPoint3D +from lib.db.queries.physio import try_get_physio_modality_with_name +from lib.db.queries.physio_coord_system import ( + try_get_coord_system, + try_get_coord_system_electrode_relation, + try_get_coord_system_name_with_name, + try_get_coord_system_point_relation, + try_get_coord_system_type_with_name, + try_get_coord_system_unit_with_name, + try_get_coord_system_unit_with_symbol, +) +from lib.env import Env + + +def get_or_create_coord_system( + env: Env, + bids_file: DbBidsFile | None, + modality: DbPhysioModality, + coord_type: DbPhysioCoordSystemType, + coord_name: DbPhysioCoordSystemName, + coord_unit: DbPhysioCoordSystemUnit, +) -> DbPhysioCoordSystem: + """ + Get a matching coordinate system or create it if it does not already exist. + """ + + coord_system = try_get_coord_system( + env.db, + bids_file.id if bids_file is not None else None, + modality.id, + coord_type.id, + coord_name.id, + coord_unit.id, + ) + + if coord_system is not None: + return coord_system + + coord_system = DbPhysioCoordSystem( + name_id = coord_name.id, + type_id = coord_type.id, + unit_id = coord_unit.id, + modality_id = modality.id, + file_path = bids_file.dataset.path / bids_file.path if bids_file is not None else None, + bids_file_id = bids_file.id if bids_file is not None else None, + ) + + env.db.add(coord_system) + env.db.flush() + return coord_system + + +def get_coord_system_type(env: Env, name: str) -> DbPhysioCoordSystemType: + """ + Get a coordinate system type by name, falling back to `'Not registered'`. + """ + + coord_type = try_get_coord_system_type_with_name(env.db, name) + if coord_type is None and name != 'Not registered': + return get_coord_system_type(env, 'Not registered') + + if coord_type is None: + raise ValueError("Missing 'Not registered' physiological coordinate system type") + + return coord_type + + +def get_coord_system_name(env: Env, name: str) -> DbPhysioCoordSystemName: + """ + Get a coordinate system name, falling back to `'Not registered'`. + """ + + coord_name = try_get_coord_system_name_with_name(env.db, name) + if coord_name is None and name != 'Not registered': + return get_coord_system_name(env, 'Not registered') + + if coord_name is None: + raise ValueError("Missing 'Not registered' physiological coordinate system name") + + return coord_name + + +def get_coord_system_unit(env: Env, symbol: str | None) -> DbPhysioCoordSystemUnit: + """ + Get a coordinate system unit by symbol, falling back to `'Not registered'`. + """ + + if symbol is not None: + coord_unit = try_get_coord_system_unit_with_symbol(env.db, symbol) + if coord_unit is not None: + return coord_unit + + coord_unit = try_get_coord_system_unit_with_name(env.db, 'Not registered') + if coord_unit is None: + raise ValueError("Missing 'Not registered' physiological coordinate system unit") + + return coord_unit + + +def get_or_create_point_relation( + env: Env, + coord_system: DbPhysioCoordSystem, + point: DbPoint3D, + name: str, +) -> DbPhysioCoordSystemPoint3d: + """ + Get a coordinate system point relation or create it if it does not already exist. + """ + + relation = try_get_coord_system_point_relation(env.db, coord_system.id, point.id) + if relation is None: + relation = DbPhysioCoordSystemPoint3d( + coord_system_id = coord_system.id, + point_3d_id = point.id, + name = name, + ) + env.db.add(relation) + + return relation + + +def get_or_create_electrode_relation( + env: Env, + coord_system: DbPhysioCoordSystem, + electrode_id: int, + physio_file: DbPhysioFile, +) -> DbPhysioCoordSystemElectrode: + """ + Get a coordinate system electrode relation or create it if it does not already exist. + """ + + relation = try_get_coord_system_electrode_relation(env.db, coord_system.id, electrode_id) + if relation is None: + relation = DbPhysioCoordSystemElectrode( + coord_system_id = coord_system.id, + electrode_id = electrode_id, + physio_file_id = physio_file.id, + ) + env.db.add(relation) + + return relation + + +def get_physio_modality(env: Env, name: str) -> DbPhysioModality: + """ + Get a physiological modality by name, falling back to `'Not registered'`. + """ + + modality = try_get_physio_modality_with_name(env.db, name) + if modality is None and name != 'Not registered': + return get_physio_modality(env, 'Not registered') + if modality is None: + raise ValueError("Missing 'Not registered' physiological modality") + return modality diff --git a/python/lib/physio/electrodes.py b/python/lib/physio/electrodes.py new file mode 100644 index 000000000..06682a61a --- /dev/null +++ b/python/lib/physio/electrodes.py @@ -0,0 +1,68 @@ +from pathlib import Path + +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 +from lib.db.queries.physio_electrode import ( + try_get_electrode_material_with_name, + try_get_electrode_type_with_name, +) +from lib.env import Env +from lib.physio.points import get_or_create_point + + +def get_or_create_electrode_type(env: Env, name: str) -> DbPhysioElectrodeType: + """ + Get an electrode type by name or create it if it does not already exist. + """ + + electrode_type = try_get_electrode_type_with_name(env.db, name) + if electrode_type is None: + electrode_type = DbPhysioElectrodeType(name=name) + env.db.add(electrode_type) + env.db.flush() + return electrode_type + + +def get_or_create_electrode_material(env: Env, name: str) -> DbPhysioElectrodeMaterial: + """ + Get an electrode material by name or create it if it does not already exist. + """ + + material = try_get_electrode_material_with_name(env.db, name) + if material is None: + material = DbPhysioElectrodeMaterial(name=name) + env.db.add(material) + env.db.flush() + return material + + +def insert_physio_electrode( + env: Env, + file_path: Path, + name: str, + x: float | None, + y: float | None, + z: float | None, + electrode_type: DbPhysioElectrodeType | None, + material: DbPhysioElectrodeMaterial | None, + impedance: int | None, + flush: bool = True, +) -> DbPhysioElectrode: + """ + Insert a physiological electrode and its three-dimensional point. + """ + + point = get_or_create_point(env, x, y, z) + electrode = DbPhysioElectrode( + type_id = electrode_type.id if electrode_type is not None else None, + material_id = material.id if material is not None else None, + name = name, + point_3d_id = point.id, + impedance = impedance, + file_path = file_path, + ) + env.db.add(electrode) + if flush: + env.db.flush() + return electrode diff --git a/python/lib/physio/points.py b/python/lib/physio/points.py new file mode 100644 index 000000000..ebdb2b898 --- /dev/null +++ b/python/lib/physio/points.py @@ -0,0 +1,18 @@ +from lib.db.models.point_3d import DbPoint3D +from lib.db.queries.physio_coord_system import try_get_point_with_coordinates +from lib.env import Env + + +def get_or_create_point(env: Env, x: float | None, y: float | None, z: float | None) -> DbPoint3D: + """ + Get a point with matching coordinates or create it if it does not already exist. + """ + + point = try_get_point_with_coordinates(env.db, x, y, z) + if point is not None: + return point + + point = DbPoint3D(x=x, y=y, z=z) + env.db.add(point) + env.db.flush() + return point diff --git a/python/lib/point_3d.py b/python/lib/point_3d.py index d0d03b42a..374835438 100644 --- a/python/lib/point_3d.py +++ b/python/lib/point_3d.py @@ -1,6 +1,9 @@ """This class represents a Point with 3D coordinates""" +from typing_extensions import deprecated + +@deprecated('Use `lib.db.models.point_3d.DbPoint3D` instead.') class Point3D: def __init__(self, pid: int, x: float, y: float, z: float): """ diff --git a/python/loris_bids_importer/src/loris_bids_importer/coord_system.py b/python/loris_bids_importer/src/loris_bids_importer/coord_system.py new file mode 100644 index 000000000..304f12e54 --- /dev/null +++ b/python/loris_bids_importer/src/loris_bids_importer/coord_system.py @@ -0,0 +1,139 @@ +from lib.db.models.bids_file import DbBidsFile +from lib.db.models.physio_coord_system import DbPhysioCoordSystem +from lib.db.models.physio_electrode import DbPhysioElectrode +from lib.db.models.physio_file import DbPhysioFile +from lib.db.models.physio_modality import DbPhysioModality +from lib.env import Env +from lib.physio.coord_system import ( + get_coord_system_name, + get_coord_system_type, + get_coord_system_unit, + get_or_create_coord_system, + get_or_create_electrode_relation, + get_or_create_point_relation, + get_physio_modality, +) +from lib.physio.points import get_or_create_point +from loris_bids_utils.eeg.coord_system import BidsCoordSystem, BidsCoordSystemJsonFile +from loris_utils.iter import find + + +def import_bids_coord_systems( + env: Env, + file: BidsCoordSystemJsonFile | None, + bids_file: DbBidsFile | None, + physio_file: DbPhysioFile, +) -> dict[str, DbPhysioCoordSystem]: + """ + Import the coordinate systems described by a BIDS coordinate system file. + """ + + if physio_file.modality is not None: + default_modality = physio_file.modality + else: + default_modality = get_physio_modality(env, 'Not registered') + + if file is not None: + definitions = file.get_coord_systems() + else: + definitions = [] + + db_coord_systems: dict[str, DbPhysioCoordSystem] = {} + for definition in definitions: + db_coord_systems[definition.kind] = import_bids_coord_system( + env, + bids_file, + definition, + default_modality, + ) + + env.db.flush() + return db_coord_systems + + +def get_or_create_bids_electrode_coord_system( + env: Env, + physio_file: DbPhysioFile, + bids_file: DbBidsFile | None, + coord_systems: dict[str, DbPhysioCoordSystem], +) -> DbPhysioCoordSystem: + """ + Get the electrode coordinate system from an import, creating a fallback when required. + """ + + if physio_file.modality is not None: + modality = physio_file.modality + else: + modality = get_physio_modality(env, 'Not registered') + + # An electrodes.tsv file in a MEG dataset can describe simultaneously recorded EEG electrodes. + if modality.name == 'meg' and 'EEG' in coord_systems: + electrode_kind = 'EEG' + else: + electrode_kind = find(['MEG', 'EEG', 'iEEG'], lambda kind: kind.lower() == modality.name) + + if electrode_kind is not None and electrode_kind in coord_systems: + return coord_systems[electrode_kind] + + return get_or_create_coord_system( + env, + bids_file, + modality, + get_coord_system_type(env, 'Not registered'), + get_coord_system_name(env, 'Not registered'), + get_coord_system_unit(env, None), + ) + + +def import_bids_coord_system( + env: Env, + bids_file: DbBidsFile | None, + coord_system: BidsCoordSystem, + default_modality: DbPhysioModality, +) -> DbPhysioCoordSystem: + """ + Import a BIDS coordinate system into LORIS. + """ + + is_sensor = coord_system.kind in ['MEG', 'EEG', 'iEEG'] + if is_sensor: + modality_name = coord_system.kind.lower() + else: + modality_name = default_modality.name + + modality = get_physio_modality(env, modality_name) + + coord_type = get_coord_system_type(env, 'Not registered' if is_sensor else coord_system.kind) + coord_name = get_coord_system_name(env, coord_system.name) + coord_unit = get_coord_system_unit(env, coord_system.unit) + + db_coord_system = get_or_create_coord_system( + env, + bids_file, + modality, + coord_type, + coord_name, + coord_unit, + ) + + for point_name, coordinates in coord_system.points.items(): + point = get_or_create_point(env, *coordinates) + get_or_create_point_relation(env, db_coord_system, point, point_name) + + return db_coord_system + + +def link_electrodes_to_coord_system( + env: Env, + physio_file: DbPhysioFile, + coord_system: DbPhysioCoordSystem, + electrodes: list[DbPhysioElectrode], +) -> None: + """ + Associate imported electrodes with their physiological file and coordinate system. + """ + + for electrode in electrodes: + get_or_create_electrode_relation(env, coord_system, electrode.id, physio_file) + + env.db.flush() diff --git a/python/loris_bids_importer/src/loris_bids_importer/eeg/main.py b/python/loris_bids_importer/src/loris_bids_importer/eeg/main.py index aadee5f1a..05a117c08 100644 --- a/python/loris_bids_importer/src/loris_bids_importer/eeg/main.py +++ b/python/loris_bids_importer/src/loris_bids_importer/eeg/main.py @@ -1,6 +1,5 @@ """Deals with EEG BIDS datasets and register them into the database.""" -import json import os from pathlib import Path @@ -9,6 +8,7 @@ from lib.db.models.physio_file import DbPhysioFile from lib.db.models.session import DbSession from lib.db.queries.hed_schema_node import get_all_hed_schema_nodes +from lib.db.queries.physio_electrode import get_physio_electrodes_with_file_id from lib.db.queries.physio_file import try_get_physio_file_with_path from lib.env import Env from lib.logging import log, log_warning @@ -17,6 +17,8 @@ from lib.physio.file import insert_physio_file from lib.physio.parameters import register_physio_file_parameters from loris_bids_utils.eeg.channels import BidsEegChannelsTsvFile +from loris_bids_utils.eeg.coord_system import BidsCoordSystemJsonFile +from loris_bids_utils.eeg.electrodes import BidsEegElectrodesTsvFile from loris_bids_utils.eeg.sidecar import BidsEegSidecarJsonFile from loris_bids_utils.files.events import BidsEventsTsvFile from loris_bids_utils.files.scans import BidsScansTsvFile @@ -26,6 +28,10 @@ from loris_bids_importer.archive import import_physio_event_archive, import_physio_file_archive from loris_bids_importer.channels import insert_bids_channels_file +from loris_bids_importer.coord_system import ( + get_or_create_bids_electrode_coord_system, + import_bids_coord_systems, +) from loris_bids_importer.copy_files import ( add_bids_scan_row, copy_loris_bids_file, @@ -33,7 +39,7 @@ get_loris_scans_path, ) from loris_bids_importer.dataset import get_or_create_loris_bids_file -from loris_bids_importer.eeg.physiological import Physiological +from loris_bids_importer.electrodes import insert_bids_electrodes_file from loris_bids_importer.events import insert_bids_event_dict_file, insert_bids_events_file from loris_bids_importer.file_type import get_check_bids_imaging_file_type_from_extension from loris_bids_importer.importer import BidsImporter @@ -47,12 +53,11 @@ class Eeg: """ - This class reads the BIDS EEG data structure and register the EEG datasets - into the database by calling the loris_bids_importer.eeg.physiological class. + Read a BIDS electrophysiology data structure and register its datasets in LORIS. """ def __init__(self, env: Env, importer: BidsImporter, bids_layout, bids_info: BidsDataTypeInfo, - session: DbSession, db, dataset_tag_dict): + session: DbSession, dataset_tag_dict): """ Constructor method for the Eeg class. @@ -60,8 +65,6 @@ def __init__(self, env: Env, importer: BidsImporter, bids_layout, bids_info: Bid :type bids_reader : dict :param bids_info : the BIDS data type information :param session : The LORIS session the EEG datasets are linked to - :param db : Database class object - :type db : object :param info : The BIDS import pipeline information :param dataset_tag_dict : Dict of dataset-inherited HED tags :type dataset_tag_dict : dict @@ -83,9 +86,6 @@ def __init__(self, env: Env, importer: BidsImporter, bids_layout, bids_info: Bid # load dataset tag dict. Used to ensure HED tags aren't duplicated self.dataset_tag_dict = dataset_tag_dict - # load database handler object - self.db = db - # find corresponding CandID and SessionID in LORIS self.session = session @@ -422,97 +422,80 @@ def fetch_and_insert_electrode_file( :rtype: str """ - # load the Physiological object that will be used to insert the - # physiological data into the database - physiological = Physiological(self.env, self.db, self.env.verbose) - - electrode_files = self.bids_layout.get_nearest( + electrode_file = self.bids_layout.get_nearest( original_physiological_file_path, return_type = 'tuple', strict = False, extension = 'tsv', suffix = 'electrodes', - all_ = True, # get all existing electrode files + all_ = False, full_search = False, ) - if not electrode_files: - message = "WARNING: no electrode file associated with " \ - "physiological file ID " + str(physiological_file.id) - print(message) + if electrode_file is None: + print(f"WARNING: no electrode file associated with physiological file ID {physiological_file.id}") return None + + existing_electrodes = get_physio_electrodes_with_file_id(self.env.db, physiological_file.id) + if existing_electrodes: + return existing_electrodes[0].file_path + + electrodes_file = BidsEegElectrodesTsvFile(Path(electrode_file.path)) + self.copy_file_to_loris_bids_dir(electrodes_file.path, derivatives) + + coordsystem_metadata_file = self.bids_layout.get_nearest( + electrode_file.path, + return_type = 'tuple', + strict = False, + extension = 'json', + suffix = 'coordsystem', + all_ = False, + full_search = False, + subject=self.bids_info.subject, + ) + + if coordsystem_metadata_file is None: + print( + '\nWARNING: no electrode metadata files (coordsystem.json) ' + f'associated with physiological file ID {physiological_file.id}' + ) + coord_systems = import_bids_coord_systems(self.env, None, None, physiological_file) + coord_system_bids_file = None else: - # maybe several electrode files - for electrode_file in electrode_files: - result = physiological.grep_electrode_from_physiological_file_id( - physiological_file.id - ) - if not result: - electrode_data = utilities.read_tsv_file(electrode_file.path) - # copy the electrode file to the LORIS BIDS import directory - electrode_path = self.copy_file_to_loris_bids_dir( - electrode_file.path, derivatives - ) - # get the blake2b hash of the electrode file - blake2 = compute_file_blake2b_hash(electrode_file.path) + coord_system_file = BidsCoordSystemJsonFile(Path(coordsystem_metadata_file.path)) + coord_system_path = self.copy_file_to_loris_bids_dir(coord_system_file.path, derivatives) - get_or_create_loris_bids_file(self.env, self.importer, Path(electrode_file.path), electrode_path) - # insert the electrode data in the database - electrode_ids = physiological.insert_electrode_file( - electrode_data, electrode_path, physiological_file, blake2 - ) + coord_system_bids_file = get_or_create_loris_bids_file( + self.env, + self.importer, + coord_system_file.path, + coord_system_path, + ) - # get coordsystem.json file - # subject-specific metadata - coordsystem_metadata_file = self.bids_layout.get_nearest( - electrode_file.path, - return_type = 'tuple', - strict = False, - extension = 'json', - suffix = 'coordsystem', - all_ = False, - full_search = False, - subject=self.bids_info.subject, - ) - if not coordsystem_metadata_file: - message = '\nWARNING: no electrode metadata files (coordsystem.json) ' \ - f'associated with physiological file ID {physiological_file.id}' - print(message) - - # insert default (not registered) coordsystem in the database - physiological.insert_electrode_metadata( - None, - None, - physiological_file, - None, - electrode_ids - ) - else: - # copy the electrode metadata file to the LORIS BIDS import directory - electrode_metadata_path = self.copy_file_to_loris_bids_dir( - coordsystem_metadata_file.path, derivatives - ) - # load json data - with open(coordsystem_metadata_file.path) as metadata_file: - electrode_metadata = json.load(metadata_file) - # get the blake2b hash of the json events file - blake2 = compute_file_blake2b_hash(coordsystem_metadata_file.path) - - get_or_create_loris_bids_file( - self.env, - self.importer, - Path(coordsystem_metadata_file.path), - electrode_metadata_path, - ) - - # insert event metadata in the database - physiological.insert_electrode_metadata( - electrode_metadata, - electrode_metadata_path, - physiological_file, - blake2, - electrode_ids - ) + coord_systems = import_bids_coord_systems( + self.env, + coord_system_file, + coord_system_bids_file, + physiological_file, + ) + + electrode_coord_system = get_or_create_bids_electrode_coord_system( + self.env, + physiological_file, + coord_system_bids_file, + coord_systems, + ) + + return insert_bids_electrodes_file( + self.env, + self.importer, + physiological_file, + self.session, + self.bids_info, + electrodes_file, + electrode_coord_system, + derivatives, + ) def fetch_and_insert_channel_file( self, physiological_file: DbPhysioFile, original_physiological_file_path, derivatives=False) -> Path: diff --git a/python/loris_bids_importer/src/loris_bids_importer/eeg/physiological.py b/python/loris_bids_importer/src/loris_bids_importer/eeg/physiological.py deleted file mode 100644 index da9f75dc4..000000000 --- a/python/loris_bids_importer/src/loris_bids_importer/eeg/physiological.py +++ /dev/null @@ -1,276 +0,0 @@ -"""This class performs database queries for BIDS physiological dataset (EEG, MEG...)""" - -from lib.database_lib.physiological_coord_system import PhysiologicalCoordSystem -from lib.database_lib.point_3d import Point3DDB -from lib.db.models.physio_file import DbPhysioFile -from lib.env import Env -from lib.physio.parameters import register_physio_file_parameter -from lib.point_3d import Point3D - - -class Physiological: - """ - This class performs database queries for BIDS physiological dataset (EEG, - MEG...). - - :Example: - - from loris_bids_importer.eeg.physiological import Physiological - from lib.database import Database - - # database connection - db = Database(config.mysql, verbose) - db.connect() - - physiological = Physiological(env, db, verbose) - - # Get file type for the physiological file - file_type = physiological.get_file_type(eeg_file) - - # grep a PhysiologicalFileID based on a blake2b hash - file_id = physiological.grep_file_id_from_hash(blake2) - - # insert electrode file into physiological_electrode - physiological.insert_electrode_file( - electrode_data, electrode_path, physiological_file_id, blake2 - ) - - ... - """ - - def __init__(self, env: Env, db, verbose): - """ - Constructor method for the Physiological class. - - :param db : Database class object - :type db : object - :param verbose: whether to be verbose - :type verbose: bool - """ - - self.env = env - self.db = db - self.verbose = verbose - - self.physiological_coord_system_db = PhysiologicalCoordSystem(self.db, self.verbose) - self.point_3d_db = Point3DDB(self.db, self.verbose) - - def grep_electrode_from_physiological_file_id(self, physiological_file_id): - """ - Greps all entries present in the physiological_electrode table for a - given PhysiologicalFileID and returns its result. - - :param physiological_file_id: physiological file's ID - :type physiological_file_id: int - - :return: tuple of dictionaries with one entry in the tuple - corresponding to one entry in physiological_electrode - :rtype: tuple - """ - - results = self.db.pselect( - query = "SELECT * " - "FROM physiological_electrode " - "WHERE PhysiologicalElectrodeID " - "IN (" - " SELECT PhysiologicalElectrodeID " - " FROM physiological_coord_system_electrode_rel " - " WHERE PhysiologicalFileID = %s)", - args = (physiological_file_id,) - ) - - return results - - def insert_electrode_file(self, electrode_data, electrode_file, - physiological_file: DbPhysioFile, blake2): - """ - Inserts the electrode information read from the file *electrode.tsv - into the physiological_electrode table, linking it to the - physiological file ID already inserted in physiological_file. - - :param electrode_data : list with dictionaries of electrodes - information to insert into - physiological_electrode - :type electrode_data : list - :param electrode_file : name of the electrode file - :type electrode_file : str - :param physiological_file : Physiological file object to link the - electrode information to - :param blake2 : blake2b hash of the electrode file - :type blake2 : str - """ - - # gather values that need to be inserted into physiological_electrode table - electrode_fields = ( - 'PhysiologicalElectrodeTypeID', - 'PhysiologicalElectrodeMaterialID', - 'Name', - 'Point3DID', - 'Impedance', - 'FilePath' - ) - electrode_ids = [] - optional_fields = ('type', 'material', 'impedance') - for row in electrode_data: - for field in optional_fields: - if field not in row.keys(): - continue - - if field == 'type': - row['type_id'] = self.db.grep_id_from_lookup_table( - id_field_name = 'PhysiologicalElectrodeTypeID', - table_name = 'physiological_electrode_type', - where_field_name = 'ElectrodeType', - where_value = row['type'], - insert_if_not_found = True - ) - if field == 'material': - row['material_id'] = self.db.grep_id_from_lookup_table( - id_field_name = 'PhysiologicalElectrodeMaterialID', - table_name = 'physiological_electrode_material', - where_field_name = 'ElectrodeMaterial', - where_value = row['material'], - insert_if_not_found = True - ) - - # map the X, Y and Z 'n/a' values to NULL - x_value = None if row['x'] == 'n/a' else row['x'] - y_value = None if row['y'] == 'n/a' else row['y'] - z_value = None if row['z'] == 'n/a' else row['z'] - p = Point3D(None, x_value, y_value, z_value) - point = self.point_3d_db.grep_or_insert_point(p) - - # insert into physiological_electrode table - values_tuple = ( - row.get('type_id'), - row.get('material_id'), - row['name'], - point.id, - row.get('impedance'), - electrode_file - ) - - inserted_electrode_id = self.db.insert( - table_name = 'physiological_electrode', - column_names = electrode_fields, - values = values_tuple, - get_last_id = True - ) - electrode_ids.append(inserted_electrode_id) - - # insert blake2b hash of electrode file into physiological_parameter_file - register_physio_file_parameter(self.env, physiological_file, 'electrode_file_blake2b_hash', blake2) - return electrode_ids - - def insert_electrode_metadata(self, electrode_metadata, electrode_metadata_file, - physiological_file: DbPhysioFile, blake2, electrode_ids): - """ - Inserts the electrode metadata information read from the file *coordsystem.json - into the physiological_coord_system, physiological_coord_system_point_3d_rel - and physiological_coord_system_electrode_rel tables, linking it to the - physiological file ID already inserted in physiological_file. - :param electrode_metadata : dictionaries of electrode metadata to insert - into the database - :type electrode_metadata : dict - :param electrode_metadata_file : PhysiologicalFileID to link the electrode info to - :type electrode_metadata_file : int - :param physiological_file : Physiological file object to link the electrode info to - :param blake2 : blake2b hash of the event file - :type blake2 : str - :param electrode_ids : blake2b hash of the event file - :type electrode_ids : str - """ - - # define modality (MEG, iEEG, EEG) - try: - modality = next( - k for k in electrode_metadata.keys() - if k.endswith('CoordinateSystem') - ).rstrip('CoordinateSystem') - modality_id = self.physiological_coord_system_db.grep_coord_system_modality_from_name(modality.lower()) - if modality_id is None: - print(f"Modality {modality} unknown in DB") - # force default - raise IndexError - except Exception: - modality_id = self.physiological_coord_system_db.grep_coord_system_modality_from_name("Not registered") - - # type (Fiducials, AnatomicalLandmark, HeadCoil, DigitizedHeapPoints) - try: - coord_system_type = next( - k for k in electrode_metadata.keys() - if k.endswith('CoordinateSystem') and not k.startswith(modality) - ).rstrip('CoordinateSystem') - type_id = self.physiological_coord_system_db.grep_coord_system_type_from_name(coord_system_type) - if type_id is None: - print(f"Type {coord_system_type} unknown in DB") - # force default - raise IndexError - except Exception: - coord_system_type = None - type_id = self.physiological_coord_system_db.grep_coord_system_type_from_name("Not registered") - - # unit - try: - unit_data = electrode_metadata[f'{modality}CoordinateUnits'] - unit_id = self.physiological_coord_system_db.grep_coord_system_unit_from_symbol(unit_data) - if unit_id is None: - print(f"Unit {unit_data} unknown in DB") - # force default - raise IndexError - except Exception: - unit_id = self.physiological_coord_system_db.grep_coord_system_unit_from_name("Not registered") - - # name - try: - coord_system_name = electrode_metadata[f'{modality}CoordinateSystem'] - name_id = self.physiological_coord_system_db.grep_coord_system_name_from_name(coord_system_name) - if name_id is None: - print(f"Name {coord_system_name} unknown in DB") - # force default - raise IndexError - except Exception: - name_id = self.physiological_coord_system_db.grep_coord_system_name_from_name("Not registered") - - # get or create coord system in db - coord_system_id = self.physiological_coord_system_db.grep_or_insert_coord_system( - name_id, - unit_id, - type_id, - modality_id, - str(electrode_metadata_file) - ) - - # define coord system referential points (e.g. LPA, RPA) + points - is_ok_ref_coords = True - try: - if coord_system_type is None: - raise KeyError - ref_coords = electrode_metadata[f'{coord_system_type}Coordinates'] - ref_points = { - ref_key : Point3D(None, *ref_val) - for ref_key, ref_val in ref_coords.items() - } - except Exception: - # no ref points - is_ok_ref_coords = False - # insert ref points if found - if is_ok_ref_coords: - # insert ref points - point_ids = {} - for rk, rv in ref_points.items(): - p = self.point_3d_db.grep_or_insert_point(rv) - point_ids[rk] = p.id - # insert ref point/coord system relations - self.physiological_coord_system_db.insert_coord_system_point_3d_relation(coord_system_id, point_ids) - - # insert the relation between coordinate file electrode and physio file - self.physiological_coord_system_db.insert_coord_system_electrodes_relation( - physiological_file.id, - coord_system_id, - electrode_ids - ) - - if blake2: - # insert blake2b hash of task event file into physiological_parameter_file - register_physio_file_parameter(self.env, physiological_file, 'coordsystem_file_json_blake2b_hash', blake2) diff --git a/python/loris_bids_importer/src/loris_bids_importer/electrodes.py b/python/loris_bids_importer/src/loris_bids_importer/electrodes.py new file mode 100644 index 000000000..305ce2698 --- /dev/null +++ b/python/loris_bids_importer/src/loris_bids_importer/electrodes.py @@ -0,0 +1,107 @@ +from pathlib import Path + +from lib.db.models.physio_coord_system import DbPhysioCoordSystem +from lib.db.models.physio_electrode import DbPhysioElectrode +from lib.db.models.physio_file import DbPhysioFile +from lib.db.models.session import DbSession +from lib.env import Env +from lib.physio.electrodes import ( + get_or_create_electrode_material, + get_or_create_electrode_type, + insert_physio_electrode, +) +from lib.physio.parameters import register_physio_file_parameter +from loris_bids_utils.eeg.electrodes import BidsEegElectrodesTsvFile, BidsEegElectrodeTsvRow +from loris_bids_utils.info import BidsAcquisitionInfo +from loris_utils.crypto import compute_file_blake2b_hash +from loris_utils.error import group_errors + +from loris_bids_importer.coord_system import link_electrodes_to_coord_system +from loris_bids_importer.copy_files import get_loris_bids_file_path +from loris_bids_importer.dataset import get_or_create_loris_bids_file +from loris_bids_importer.importer import BidsImporter + + +def insert_bids_electrodes_file( + env: Env, + importer: BidsImporter, + physio_file: DbPhysioFile, + session: DbSession, + acquisition: BidsAcquisitionInfo, + electrodes_file: BidsEegElectrodesTsvFile, + coord_system: DbPhysioCoordSystem, + derivative: bool = False, +) -> Path: + """ + Insert the electrodes from a BIDS electrode file and link them to their coordinate system. + """ + + loris_electrodes_file_path = get_loris_bids_file_path( + importer, + session, + acquisition.data_type, + electrodes_file.path, + derivative, + ) + + blake2_hash = compute_file_blake2b_hash(electrodes_file.path) + + electrodes = group_errors( + f"Could not import electrodes from file '{electrodes_file.path.name}'.", + ( + lambda: insert_bids_electrode( + env, + loris_electrodes_file_path, + electrode, + flush=False, + ) for electrode in electrodes_file.rows + ), + ) + + env.db.flush() + + link_electrodes_to_coord_system(env, physio_file, coord_system, electrodes) + get_or_create_loris_bids_file( + env, + importer, + electrodes_file.path, + loris_electrodes_file_path, + ) + register_physio_file_parameter( + env, + physio_file, + 'electrode_file_blake2b_hash', + blake2_hash, + ) + + env.db.flush() + + return loris_electrodes_file_path + + +def insert_bids_electrode( + env: Env, + loris_file_path: Path, + electrode: BidsEegElectrodeTsvRow, + flush: bool = True, +) -> DbPhysioElectrode: + """ + Insert an electrode from a BIDS electrode TSV row into the database. + """ + + electrode_type = get_or_create_electrode_type(env, electrode.type) if electrode.type is not None else None + + material = get_or_create_electrode_material(env, electrode.material) if electrode.material is not None else None + + return insert_physio_electrode( + env, + loris_file_path, + electrode.name, + electrode.x, + electrode.y, + electrode.z, + electrode_type, + material, + electrode.impedance, + flush, + ) diff --git a/python/loris_bids_importer/src/loris_bids_importer/main.py b/python/loris_bids_importer/src/loris_bids_importer/main.py index ba6971ece..ef56cd85f 100644 --- a/python/loris_bids_importer/src/loris_bids_importer/main.py +++ b/python/loris_bids_importer/src/loris_bids_importer/main.py @@ -1,7 +1,6 @@ from typing import Any from lib.config import get_default_bids_visit_label_config -from lib.database import Database from lib.db.models.session import DbSession from lib.db.queries.candidate import try_get_candidate_with_psc_id from lib.db.queries.session import try_get_session_with_cand_id_visit_label @@ -25,7 +24,7 @@ from loris_bids_importer.validation.subjects import validate_bids_subjects -def import_bids_dataset(env: Env, args: BidsImporterArgs, legacy_db: Database): +def import_bids_dataset(env: Env, args: BidsImporterArgs): """ Read the provided BIDS dataset and import it into LORIS. """ @@ -84,7 +83,7 @@ def import_bids_dataset(env: Env, args: BidsImporterArgs, legacy_db: Database): # Process each session directory. for bids_session in bids.sessions: - import_bids_session(env, importer, bids_session, dataset_tag_dict, legacy_db) + import_bids_session(env, importer, bids_session, dataset_tag_dict) # Print import summary. @@ -96,7 +95,6 @@ def import_bids_session( importer: BidsImporter, bids_session: BidsSessionReader, dataset_tag_dict: dict[Any, Any], - legacy_db: Database, ): """ Read the provided BIDS session directory and import it into LORIS. @@ -127,7 +125,7 @@ def import_bids_session( # Process each data type directory. for data_type in bids_session.data_types: - import_bids_data_type(env, importer, session, data_type, dataset_tag_dict, legacy_db) + import_bids_data_type(env, importer, session, data_type, dataset_tag_dict) def import_bids_data_type( @@ -136,7 +134,6 @@ def import_bids_data_type( session: DbSession, data_type: BidsDataTypeReader, dataset_tag_dict: dict[Any, Any], - legacy_db: Database, ): """ Read the provided BIDS data type directory and import it into LORIS. @@ -151,7 +148,7 @@ def import_bids_data_type( case BidsMriDataTypeReader(): import_bids_mri_data_type(env, importer, session, data_type) case BidsDataTypeReader(): - import_bids_eeg_data_type_files(env, importer, session, data_type, dataset_tag_dict, legacy_db) + import_bids_eeg_data_type_files(env, importer, session, data_type, dataset_tag_dict) def import_bids_eeg_data_type_files( @@ -160,7 +157,6 @@ def import_bids_eeg_data_type_files( session: DbSession, data_type: BidsDataTypeReader, dataset_tag_dict: dict[Any, Any], - legacy_db: Database, ): """ Read the provided BIDS EEG data type directory and import it into LORIS. @@ -172,7 +168,6 @@ def import_bids_eeg_data_type_files( importer = importer, bids_layout = data_type.session.subject.dataset.layout, bids_info = data_type.info, - db = legacy_db, session = session, dataset_tag_dict = dataset_tag_dict, ) diff --git a/python/loris_bids_importer/src/loris_bids_importer/scripts/import_bids_dataset.py b/python/loris_bids_importer/src/loris_bids_importer/scripts/import_bids_dataset.py index fb7262a52..49005665b 100755 --- a/python/loris_bids_importer/src/loris_bids_importer/scripts/import_bids_dataset.py +++ b/python/loris_bids_importer/src/loris_bids_importer/scripts/import_bids_dataset.py @@ -90,12 +90,7 @@ def main(): args = pack_args(loris_getopt_obj.options_dict) - # read and insert BIDS data - import_bids_dataset( - env, - args, - loris_getopt_obj.db, - ) + import_bids_dataset(env, args) print("Success !") diff --git a/python/loris_bids_importer/tests/conftest.py b/python/loris_bids_importer/tests/conftest.py new file mode 100644 index 000000000..c244ece05 --- /dev/null +++ b/python/loris_bids_importer/tests/conftest.py @@ -0,0 +1 @@ +pytest_plugins = ['tests.fixtures'] diff --git a/python/loris_bids_importer/tests/test_coord_system.py b/python/loris_bids_importer/tests/test_coord_system.py new file mode 100644 index 000000000..2cd6cb749 --- /dev/null +++ b/python/loris_bids_importer/tests/test_coord_system.py @@ -0,0 +1,164 @@ +import json +from pathlib import Path + +from lib.db.models.bids_dataset import DbBidsDataset +from lib.db.models.bids_file import DbBidsFile +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.physio_electrode import DbPhysioElectrode +from lib.db.models.physio_file import DbPhysioFile +from lib.db.models.physio_modality import DbPhysioModality +from lib.db.models.point_3d import DbPoint3D +from lib.env import Env +from loris_bids_utils.eeg.coord_system import BidsCoordSystemJsonFile +from sqlalchemy import select + +from loris_bids_importer.coord_system import ( + get_or_create_bids_electrode_coord_system, + import_bids_coord_systems, + link_electrodes_to_coord_system, +) + + +def test_import_bids_coord_systems_deduplicates_within_bids_file(env: Env, tmp_path: Path): + _add_lookups(env) + dataset = _add_dataset(env) + bids_file = _add_bids_file(env, dataset, Path('sub-01/meg/sub-01_coordsystem.json')) + first_physio_file = _add_physio_file(env, 1) + second_physio_file = _add_physio_file(env, 2) + coord_system_file = _make_coord_system_file(tmp_path, { + 'MEGCoordinateSystem': 'CTF', + 'MEGCoordinateUnits': 'm', + 'HeadCoilCoordinateSystem': 'CTF', + 'HeadCoilCoordinateUnits': 'cm', + 'HeadCoilCoordinates': {'coil1': [1, 2, 3]}, + }) + + first = import_bids_coord_systems(env, coord_system_file, bids_file, first_physio_file) + second = import_bids_coord_systems(env, coord_system_file, bids_file, second_physio_file) + first_electrode = _add_electrode(env, 101) + second_electrode = _add_electrode(env, 102) + link_electrodes_to_coord_system(env, first_physio_file, first['MEG'], [first_electrode]) + link_electrodes_to_coord_system(env, second_physio_file, second['MEG'], [second_electrode]) + + assert first.keys() == second.keys() + assert [item.id for item in first.values()] == [item.id for item in second.values()] + assert len(env.db.scalars(select(DbPhysioCoordSystem)).all()) == 2 + assert len(env.db.scalars(select(DbPhysioCoordSystemElectrode)).all()) == 2 + assert len(env.db.scalars(select(DbPhysioCoordSystemPoint3d)).all()) == 1 + assert all(item.bids_file_id == bids_file.id for item in first.values()) + + +def test_import_bids_coord_systems_keeps_different_bids_files_separate(env: Env, tmp_path: Path): + _add_lookups(env) + dataset = _add_dataset(env) + first_bids_file = _add_bids_file(env, dataset, Path('sub-01/meg/sub-01_coordsystem.json')) + second_bids_file = _add_bids_file(env, dataset, Path('sub-02/meg/sub-02_coordsystem.json')) + first_physio_file = _add_physio_file(env, 1) + second_physio_file = _add_physio_file(env, 2) + coord_system_file = _make_coord_system_file( + tmp_path, + {'MEGCoordinateSystem': 'CTF', 'MEGCoordinateUnits': 'm'}, + ) + + first = import_bids_coord_systems(env, coord_system_file, first_bids_file, first_physio_file) + second = import_bids_coord_systems(env, coord_system_file, second_bids_file, second_physio_file) + + assert first['MEG'].id != second['MEG'].id + assert len(env.db.scalars(select(DbPhysioCoordSystem)).all()) == 2 + + +def test_electrode_import_creates_default_coord_system_when_required(env: Env): + _add_lookups(env) + physio_file = _add_physio_file(env, 1) + + coord_systems = import_bids_coord_systems(env, None, None, physio_file) + + assert coord_systems == {} + + electrode_system = get_or_create_bids_electrode_coord_system(env, physio_file, None, coord_systems) + + assert electrode_system.name.name == 'Not registered' + assert electrode_system.type.name == 'Not registered' + assert electrode_system.unit.name == 'Not registered' + + +def _make_coord_system_file(tmp_path: Path, metadata: dict[str, object]) -> BidsCoordSystemJsonFile: + path = tmp_path / 'coordsystem.json' + path.write_text(json.dumps(metadata)) + return BidsCoordSystemJsonFile(path) + + +def _add_lookups(env: Env): + env.db.add_all([ + DbPhysioModality(id=1, name='meg'), + DbPhysioModality(id=2, name='eeg'), + DbPhysioModality(id=3, name='ieeg'), + DbPhysioModality(id=4, name='Not registered'), + DbPhysioCoordSystemName(id=1, name='Not registered'), + DbPhysioCoordSystemName(id=2, name='CTF'), + DbPhysioCoordSystemType(id=1, name='Not registered'), + DbPhysioCoordSystemType(id=2, name='HeadCoil'), + DbPhysioCoordSystemType(id=3, name='AnatomicalLandmark'), + DbPhysioCoordSystemType(id=4, name='DigitizedHeadPoints'), + DbPhysioCoordSystemType(id=5, name='Fiducials'), + DbPhysioCoordSystemUnit(id=1, name='Not registered', symbol=None), + DbPhysioCoordSystemUnit(id=2, name='Meter', symbol='m'), + DbPhysioCoordSystemUnit(id=3, name='Centimeter', symbol='cm'), + DbPhysioCoordSystemUnit(id=4, name='Millimeter', symbol='mm'), + ]) + env.db.flush() + + +def _add_dataset(env: Env) -> DbBidsDataset: + dataset = DbBidsDataset(path=Path('bids')) + env.db.add(dataset) + env.db.flush() + return dataset + + +def _add_bids_file(env: Env, dataset: DbBidsDataset, path: Path) -> DbBidsFile: + bids_file = DbBidsFile( + dataset_id = dataset.id, + path = path, + source_path = path, + blake2b_hash = 'hash', + derivative = False, + ) + env.db.add(bids_file) + env.db.flush() + return bids_file + + +def _add_physio_file(env: Env, physio_file_id: int) -> DbPhysioFile: + physio_file = DbPhysioFile( + id = physio_file_id, + modality_id = 1, + output_type_id = 1, + session_id = 1, + type = 'ctf', + inserted_by_user = 'pytest', + path = Path(f'file-{physio_file_id}.ds'), + ) + env.db.add(physio_file) + env.db.flush() + return physio_file + + +def _add_electrode(env: Env, electrode_id: int) -> DbPhysioElectrode: + point = DbPoint3D(x=float(electrode_id), y=None, z=None) + env.db.add(point) + env.db.flush() + electrode = DbPhysioElectrode( + id = electrode_id, + name = f'E{electrode_id}', + point_3d_id = point.id, + file_path = Path('electrodes.tsv'), + ) + env.db.add(electrode) + env.db.flush() + return electrode diff --git a/python/loris_bids_importer/tests/test_electrodes.py b/python/loris_bids_importer/tests/test_electrodes.py new file mode 100644 index 000000000..159ecc765 --- /dev/null +++ b/python/loris_bids_importer/tests/test_electrodes.py @@ -0,0 +1,45 @@ +from pathlib import Path + +from lib.db.models.physio_electrode_material import DbPhysioElectrodeMaterial +from lib.db.models.physio_electrode_type import DbPhysioElectrodeType +from lib.env import Env +from loris_bids_utils.eeg.electrodes import BidsEegElectrodeTsvRow +from sqlalchemy import select + +from loris_bids_importer.electrodes import insert_bids_electrode + + +def test_insert_bids_electrode_uses_typed_orm_models(env: Env): + row_data: dict[str, str | None] = { + 'name': 'E1', + 'x': '1.5', + 'y': 'n/a', + 'z': '3', + 'type': 'depth', + 'material': 'platinum', + 'impedance': '12', + } + electrode = insert_bids_electrode( + env, + Path('sub-01/eeg/sub-01_electrodes.tsv'), + BidsEegElectrodeTsvRow(row_data), + ) + + assert electrode.name == 'E1' + assert (electrode.point_3d.x, electrode.point_3d.y, electrode.point_3d.z) == (1.5, None, 3.0) + assert electrode.type is not None + assert electrode.type.name == 'depth' + assert electrode.material is not None + assert electrode.material.name == 'platinum' + assert electrode.impedance == 12 + assert row_data == { + 'name': 'E1', + 'x': '1.5', + 'y': 'n/a', + 'z': '3', + 'type': 'depth', + 'material': 'platinum', + 'impedance': '12', + } + assert len(env.db.scalars(select(DbPhysioElectrodeType)).all()) == 1 + assert len(env.db.scalars(select(DbPhysioElectrodeMaterial)).all()) == 1 diff --git a/python/loris_bids_utils/src/loris_bids_utils/eeg/coord_system.py b/python/loris_bids_utils/src/loris_bids_utils/eeg/coord_system.py new file mode 100644 index 000000000..91160688f --- /dev/null +++ b/python/loris_bids_utils/src/loris_bids_utils/eeg/coord_system.py @@ -0,0 +1,77 @@ +from dataclasses import dataclass +from typing import Any, cast + +from loris_bids_utils.json import BidsJsonFile + +BIDS_COORD_SYSTEM_KINDS: list[str] = [ + 'MEG', + 'EEG', + 'iEEG', + 'HeadCoil', + 'DigitizedHeadPoints', + 'AnatomicalLandmark', + 'Fiducials', +] + + +@dataclass(frozen=True) +class BidsCoordSystem: + """ + A coordinate system description parsed from a BIDS `coordsystem.json` file. + """ + + kind: str + name: str + unit: str | None + points: dict[str, tuple[float, float, float]] + + +class BidsCoordSystemJsonFile(BidsJsonFile): + """ + Class representing a BIDS `coordsystem.json` file. + """ + + def get_coord_systems(self) -> list[BidsCoordSystem]: + """ + Parse every coordinate system description in this file. + """ + + definitions: list[BidsCoordSystem] = [] + for kind in BIDS_COORD_SYSTEM_KINDS: + system_key = f'{kind}CoordinateSystem' + if system_key not in self.data: + continue + + definitions.append(BidsCoordSystem( + kind = kind, + name = str(self.data[system_key]), + unit = _parse_unit(self.data.get(f'{kind}CoordinateUnits')), + points = _parse_points(self.data.get(f'{kind}Coordinates')), + )) + + return definitions + + +def _parse_unit(value: Any) -> str | None: + return value if isinstance(value, str) and value != 'n/a' else None + + +def _parse_points(value: Any) -> dict[str, tuple[float, float, float]]: + if not isinstance(value, dict): + return {} + + points: dict[str, tuple[float, float, float]] = {} + for name, untyped_coordinates in cast(dict[object, object], value).items(): + coordinates = cast(list[object], untyped_coordinates) if isinstance(untyped_coordinates, list) else None + if ( + isinstance(name, str) + and coordinates is not None + and len(coordinates) == 3 + and all(isinstance(coordinate, int | float) for coordinate in coordinates) + ): + points[name] = ( + float(cast(int | float, coordinates[0])), + float(cast(int | float, coordinates[1])), + float(cast(int | float, coordinates[2])), + ) + return points diff --git a/python/loris_bids_utils/src/loris_bids_utils/eeg/electrodes.py b/python/loris_bids_utils/src/loris_bids_utils/eeg/electrodes.py new file mode 100644 index 000000000..8d72af3b5 --- /dev/null +++ b/python/loris_bids_utils/src/loris_bids_utils/eeg/electrodes.py @@ -0,0 +1,60 @@ +from pathlib import Path + +from loris_utils.parse import try_parse_float, try_parse_int + +from loris_bids_utils.tsv import BidsTsvFile, BidsTsvRow + + +class BidsEegElectrodeTsvRow(BidsTsvRow): + """ + Class representing a BIDS EEG or iEEG `electrodes.tsv` row. + + Documentation: + - https://bids-specification.readthedocs.io/en/stable/modality-specific-files/electroencephalography.html#electrode-locations-electrodestsv + - https://bids-specification.readthedocs.io/en/stable/modality-specific-files/intracranial-electroencephalography.html#electrode-description-electrodestsv + """ + + name: str + x: float | None + y: float | None + z: float | None + type: str | None + material: str | None + impedance: int | None + + def __init__(self, data: dict[str, str | None]): + super().__init__(data) + + name = data.get('name') + if name is None: + raise Exception("Missing electrode name in BIDS electrode file.") + + self.name = name + self.x = _try_parse_bids_float(data.get('x')) + self.y = _try_parse_bids_float(data.get('y')) + self.z = _try_parse_bids_float(data.get('z')) + self.type = _nullify_bids_missing_value(data.get('type')) + self.material = _nullify_bids_missing_value(data.get('material')) + + impedance = _nullify_bids_missing_value(data.get('impedance')) + self.impedance = try_parse_int(impedance) if impedance is not None else None + + +class BidsEegElectrodesTsvFile(BidsTsvFile[BidsEegElectrodeTsvRow]): + """ + Class representing a BIDS EEG or iEEG `electrodes.tsv` file. + """ + + def __init__(self, path: Path): + super().__init__(BidsEegElectrodeTsvRow, path) + + +def _try_parse_bids_float(value: str | None) -> float | None: + value = _nullify_bids_missing_value(value) + return try_parse_float(value) if value is not None else None + + +def _nullify_bids_missing_value(value: str | None) -> str | None: + if value is None or value.strip().lower() == 'n/a': + return None + return value diff --git a/python/loris_bids_utils/tests/eeg/test_coord_system.py b/python/loris_bids_utils/tests/eeg/test_coord_system.py new file mode 100644 index 000000000..6b7922d65 --- /dev/null +++ b/python/loris_bids_utils/tests/eeg/test_coord_system.py @@ -0,0 +1,41 @@ +import json +from pathlib import Path + +from loris_bids_utils.eeg.coord_system import BidsCoordSystemJsonFile + + +def test_parse_bids_coord_systems_keeps_every_description(tmp_path: Path): + path = tmp_path / 'coordsystem.json' + path.write_text(json.dumps({ + 'HeadCoilCoordinateSystem': 'CTF', + 'HeadCoilCoordinateUnits': 'cm', + 'HeadCoilCoordinates': {'coil1': [1, 2, 3]}, + 'MEGCoordinateSystem': 'CTF', + 'MEGCoordinateUnits': 'm', + 'EEGCoordinateSystem': 'CapTrak', + 'EEGCoordinateUnits': 'mm', + 'AnatomicalLandmarkCoordinateSystem': 'CTF', + 'AnatomicalLandmarkCoordinateUnits': 'mm', + 'AnatomicalLandmarkCoordinates': {'NAS': [4, 5, 6]}, + })) + definitions = BidsCoordSystemJsonFile(path).get_coord_systems() + + assert [item.kind for item in definitions] == [ + 'MEG', + 'EEG', + 'HeadCoil', + 'AnatomicalLandmark', + ] + assert definitions[2].points == {'coil1': (1.0, 2.0, 3.0)} + + +def test_parse_bids_coord_systems_adds_missing_sensor_system(tmp_path: Path): + path = tmp_path / 'coordsystem.json' + path.write_text(json.dumps({ + 'FiducialsCoordinateSystem': 'CTF', + 'FiducialsCoordinateUnits': 'mm', + })) + definitions = BidsCoordSystemJsonFile(path).get_coord_systems() + + assert len(definitions) == 1 + assert definitions[0].kind == 'Fiducials' diff --git a/python/loris_bids_utils/tests/eeg/test_electrodes.py b/python/loris_bids_utils/tests/eeg/test_electrodes.py new file mode 100644 index 000000000..32a20263d --- /dev/null +++ b/python/loris_bids_utils/tests/eeg/test_electrodes.py @@ -0,0 +1,23 @@ +from pathlib import Path + +from loris_bids_utils.eeg.electrodes import BidsEegElectrodesTsvFile + + +def test_parse_electrodes_file(tmp_path: Path): + path = tmp_path / 'electrodes.tsv' + path.write_text( + 'name\tx\ty\tz\ttype\tmaterial\timpedance\n' + 'E1\t1.5\t2\t3\tdepth\tplatinum\t12\n' + 'E2\tn/a\tN/A\tNaN\tn/a\t\tn/a\n' + ) + + electrodes = BidsEegElectrodesTsvFile(path).rows + + assert (electrodes[0].x, electrodes[0].y, electrodes[0].z) == (1.5, 2.0, 3.0) + assert electrodes[0].type == 'depth' + assert electrodes[0].material == 'platinum' + assert electrodes[0].impedance == 12 + assert (electrodes[1].x, electrodes[1].y, electrodes[1].z) == (None, None, None) + assert electrodes[1].type is None + assert electrodes[1].material is None + assert electrodes[1].impedance is None diff --git a/python/tests/fixtures.py b/python/tests/fixtures.py new file mode 100644 index 000000000..e5095e154 --- /dev/null +++ b/python/tests/fixtures.py @@ -0,0 +1,53 @@ +from collections.abc import Iterator +from pathlib import Path +from types import SimpleNamespace + +import pytest +from sqlalchemy import Engine, create_engine +from sqlalchemy.orm import Session + +from lib.db.base import Base +from lib.env import Env + + +@pytest.fixture +def db_engine() -> Iterator[Engine]: + """ + Create an in-memory SQLite database engine based on the ORM schema. + """ + + engine = create_engine('sqlite:///:memory:') + Base.metadata.create_all(engine) + + yield engine + + engine.dispose() + + +@pytest.fixture +def db(db_engine: Engine) -> Iterator[Session]: + """ + Create an in-memory SQLite database session to run a unit test. + """ + + with Session(db_engine) as db: + yield db + db.rollback() + + +@pytest.fixture +def env(db_engine: Engine, db: Session, tmp_path: Path) -> Env: + """ + Create a LORIS environment with an in-memory SQLite database to run a unit test. + """ + + return Env( + db_engine = db_engine, + db = db, + script_name = 'pytest', + config_info = SimpleNamespace(), + tmp_dir_path = tmp_path, + log_file_path = tmp_path / 'test.log', + verbose = False, + cleanups = [], + ) diff --git a/python/tests/unit/conftest.py b/python/tests/unit/conftest.py index e5095e154..c244ece05 100644 --- a/python/tests/unit/conftest.py +++ b/python/tests/unit/conftest.py @@ -1,53 +1 @@ -from collections.abc import Iterator -from pathlib import Path -from types import SimpleNamespace - -import pytest -from sqlalchemy import Engine, create_engine -from sqlalchemy.orm import Session - -from lib.db.base import Base -from lib.env import Env - - -@pytest.fixture -def db_engine() -> Iterator[Engine]: - """ - Create an in-memory SQLite database engine based on the ORM schema. - """ - - engine = create_engine('sqlite:///:memory:') - Base.metadata.create_all(engine) - - yield engine - - engine.dispose() - - -@pytest.fixture -def db(db_engine: Engine) -> Iterator[Session]: - """ - Create an in-memory SQLite database session to run a unit test. - """ - - with Session(db_engine) as db: - yield db - db.rollback() - - -@pytest.fixture -def env(db_engine: Engine, db: Session, tmp_path: Path) -> Env: - """ - Create a LORIS environment with an in-memory SQLite database to run a unit test. - """ - - return Env( - db_engine = db_engine, - db = db, - script_name = 'pytest', - config_info = SimpleNamespace(), - tmp_dir_path = tmp_path, - log_file_path = tmp_path / 'test.log', - verbose = False, - cleanups = [], - ) +pytest_plugins = ['tests.fixtures']