diff --git a/README.md b/README.md index 8c833c4..c97ae86 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,8 @@ LavLab Python Utils is designed to streamline common tasks in our research workf - **Data Processing:** Functions to handle various data formats used in our research, including DICOM, NIfTI, and more. - **Visualization:** Tools to create publication-quality plots and visualizations. - **Integration:** Seamless integration with other tools and services used in our lab. +- **OMERO Support:** Connect to and work with OMERO servers for image data management. +- **XNAT Support:** Connect to and work with XNAT servers for neuroimaging data management. ## Installation @@ -27,6 +29,11 @@ You can install LavLab Python Utils via pip: python3 -m pip install https://github.com/laviolette-lab/lavlab-python-utils/releases/latest/download/lavlab_python_utils-latest-py3-none-any.whl # optional install targets, must install wheel from github using command above first! python3 -m pip install 'lavlab-python-utils[all]' + +# Or install specific features: +python3 -m pip install 'lavlab-python-utils[omero]' # For OMERO support +python3 -m pip install 'lavlab-python-utils[xnat]' # For XNAT support +python3 -m pip install 'lavlab-python-utils[jupyter]' # For Jupyter tools ``` ## Documentation diff --git a/docs/xnat.md b/docs/xnat.md new file mode 100644 index 0000000..5025d96 --- /dev/null +++ b/docs/xnat.md @@ -0,0 +1,83 @@ +# XNAT Support + +This package now includes support for XNAT servers in addition to OMERO servers. + +## Installation + +To use XNAT features, install with the xnat extra: + +```bash +pip install lavlab-python-utils[xnat] +``` + +## Configuration + +Configure XNAT service in your configuration file (`~/.lavlab.yml` or `/etc/lavlab.yml`): + +```yaml +histology: + service: + name: 'xnat' + host: 'https://your-xnat-server.org' + username: your_username # optional, will prompt if not provided + passwd: your_password # optional, recommended to use keyring instead +``` + +## Usage + +### Basic Connection + +```python +import lavlab.xnat + +# Connect using configuration +session = lavlab.xnat.connect() + +# Or create service provider directly +provider = lavlab.xnat.XNATServiceProvider() +session = provider.login() +``` + +### Helper Functions + +```python +from lavlab.xnat.helpers import get_projects, get_subjects, get_experiments + +# Get available projects +projects = get_projects(session) + +# Get subjects for a project +subjects = get_subjects(session, "PROJECT_ID") + +# Get experiments for a subject +experiments = get_experiments(session, "PROJECT_ID", "SUBJECT_ID") + +# Download scan files +from lavlab.xnat.helpers import download_scan_file + +with download_scan_file(session, "EXPERIMENT_ID", "SCAN_ID", "filename.dcm") as file_path: + # Work with downloaded file + pass +# File is automatically cleaned up +``` + +### Search and Discovery + +```python +from lavlab.xnat.helpers import find_experiments_by_type, search_experiments + +# Find MR experiments in a project +mr_experiments = find_experiments_by_type(session, "PROJECT_ID", "xnat:mrSessionData") + +# Search experiments with custom criteria +results = search_experiments(session, project="PROJECT_ID", modality="MR") +``` + +## Service Provider Architecture + +The XNAT support follows the same service provider pattern as OMERO: + +- `XNATServiceProvider` handles authentication and connection +- Credentials are managed through keyring for security +- The service is registered as an entry point for dynamic loading +- Configuration follows the same patterns as other services \ No newline at end of file diff --git a/examples/xnat_example.py b/examples/xnat_example.py new file mode 100644 index 0000000..6876f28 --- /dev/null +++ b/examples/xnat_example.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +""" +Example script demonstrating XNAT support in lavlab-python-utils. + +This script shows how to: +1. Configure XNAT service +2. Connect to XNAT server +3. List projects, subjects, and experiments +4. Download scan files + +Requirements: +- lavlab-python-utils[xnat] package installed +- XNAT server configuration in ~/.lavlab.yml +""" + +import lavlab +from lavlab.xnat import connect +from lavlab.xnat.helpers import ( + get_projects, + get_subjects, + get_experiments, + get_scan_files, + download_scan_file +) + + +def main(): + """Demonstrate XNAT functionality.""" + + # Configure for XNAT service - this would typically be in ~/.lavlab.yml + # Example configuration: + config = { + "histology": { + "service": { + "name": "xnat", + "host": "https://your-xnat-server.org", + # username/password will be prompted or loaded from keyring + } + } + } + + print("XNAT Support Example") + print("=" * 50) + + try: + # Connect to XNAT server + print("Connecting to XNAT server...") + session = connect() + print("✓ Connected successfully!") + + # List available projects + print("\nAvailable projects:") + projects = get_projects(session) + for i, project_id in enumerate(projects[:5]): # Show first 5 + print(f" {i+1}. {project_id}") + + if projects: + # Use first project as example + project_id = projects[0] + print(f"\nExploring project: {project_id}") + + # List subjects in the project + subjects = get_subjects(session, project_id) + print(f" Subjects found: {len(subjects)}") + + if subjects: + # Use first subject as example + subject_id = subjects[0] + print(f" Exploring subject: {subject_id}") + + # List experiments for the subject + experiments = get_experiments(session, project_id, subject_id) + print(f" Experiments found: {len(experiments)}") + + if experiments: + # Show experiment details + experiment_id = experiments[0] + print(f" Example experiment: {experiment_id}") + + # This would typically be used for actual file downloads: + # with download_scan_file(session, experiment_id, scan_id, filename) as file_path: + # print(f"Downloaded file: {file_path}") + # # Process the file here + + print("\n✓ XNAT exploration completed successfully!") + + except RuntimeError as e: + print(f"✗ Error: {e}") + print("\nMake sure to configure XNAT service in ~/.lavlab.yml:") + print(""" +histology: + service: + name: 'xnat' + host: 'https://your-xnat-server.org' + # username and password will be prompted or loaded from keyring +""") + + except Exception as e: + print(f"✗ Unexpected error: {e}") + + finally: + # Clean up connection + try: + session.disconnect() + print("✓ Disconnected from XNAT server") + except: + pass + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 6d4608d..b569f5c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,18 +48,22 @@ omero = [ "zeroc-ice @ https://github.com/glencoesoftware/zeroc-ice-py-linux-x86_64/releases/download/20240202/zeroc_ice-3.6.5-cp312-cp312-manylinux_2_28_x86_64.whl#sha256=96fb9066912c52f2503e9f9207f98d51de79c475d19ebd3157aae7bc522b5826 ; python_version == '3.12' and platform_system == 'Linux'", "omero-py" ] +xnat = [ + "xnat" +] jupyter = [ "jupyter", "dash-slicer", "dash-bootstrap-components" ] all = [ - "lavlab-python-utils[omero, jupyter]" + "lavlab-python-utils[omero, xnat, jupyter]" ] [project.entry-points."lavlab_python_utils.service_providers"] OMERO = "lavlab.omero:OmeroServiceProvider" IDR = "lavlab.omero:IDRServiceProvider" +XNAT = "lavlab.xnat:XNATServiceProvider" [project.urls] Documentation = "https://github.com/LavLabInfrastructure/lavlab-python-utils#readme" @@ -90,7 +94,7 @@ dependencies = [] build = "hatch build && chmod -R 777 dist/*" [tool.hatch.envs.test] -features = [ "omero", "jupyter" ] +features = [ "omero", "xnat", "jupyter" ] dependencies = [ "toml", "pytest", @@ -104,7 +108,7 @@ test = "pytest {args:test}" cov = "pytest --cov=src --cov-report=xml {args:test}" [tool.hatch.envs.lint] -features = [ "omero", "jupyter" ] +features = [ "omero", "xnat", "jupyter" ] dependencies = [ "toml", "pytest", @@ -117,7 +121,7 @@ format = "black src test" check = "black src test --check" [tool.hatch.envs.types] -features = [ "omero", "jupyter" ] +features = [ "omero", "xnat", "jupyter" ] dependencies = [ "mypy>=1.0.0", ] @@ -125,7 +129,7 @@ dependencies = [ check = "mypy --install-types --non-interactive {args:src/lavlab test}" [tool.hatch.envs.docs] -features = [ "omero", "jupyter" ] +features = [ "omero", "xnat", "jupyter" ] dependencies = [ "mkdocs", "mkdocstrings-python", @@ -164,6 +168,7 @@ exclude_lines = [ [tool.mypy.overrides] module = [ "omero", + "xnat", "scipy" ] ignore_missing_imports = true diff --git a/src/lavlab/xnat/__init__.py b/src/lavlab/xnat/__init__.py new file mode 100644 index 0000000..f9ebb2a --- /dev/null +++ b/src/lavlab/xnat/__init__.py @@ -0,0 +1,82 @@ +"""XNAT Utility module""" + +import logging + +import xnat # type: ignore + +import lavlab +from lavlab.login import AbstractServiceProvider + +LOGGER = lavlab.LOGGER.getChild("xnat") + + +class XNATServiceProvider(AbstractServiceProvider): # pylint: disable=R0903 + """ + Provides a connection to a defined XNAT server using xnatpy. + """ + + SERVICE = "XNAT" + + def login(self) -> xnat.XNATSession: + """ + Logins into configured XNAT server. + + Returns + ------- + xnat.XNATSession + XNAT API session + + Raises + ------ + RuntimeError + Could not login to XNAT server. + """ + details = lavlab.ctx.histology.service.copy() + if details.get("username") is None or details.get("passwd") is None: + username, password = self.cred_provider.get_credentials() + details.update({"username": username, "passwd": password}) + + # Convert field names to match xnatpy expectations + connection_params = { + "server": details.get("host"), + "username": details.get("username"), + "password": details.get("passwd"), + } + + try: + session = xnat.connect(**connection_params) + return session + except Exception as e: + raise RuntimeError(f"Unable to connect to XNAT server: {e}") from e + + +def connect() -> xnat.XNATSession: + """ + Uses the UtilContext to connect to the configured XNAT server + + Returns + ------- + xnat.XNATSession + XNAT API session + """ + if not lavlab.ctx.histology.service.get("name").upper() == "XNAT": + raise RuntimeError("Service is not XNAT.") + return lavlab.ctx.histology.service_provider.login() + + +def set_xnat_logging_level(level: str): + """ + Sets a given python logging._Level in all xnat loggers. + + Parameters + ---------- + level: logging._Level + + Returns + ------- + None + """ + LOGGER.info("Setting XNAT logging level to %s.", level) + for name in logging.root.manager.loggerDict.keys(): # pylint: disable=E1101 + if name.startswith("xnat"): + logging.getLogger(name).setLevel(level) \ No newline at end of file diff --git a/src/lavlab/xnat/helpers.py b/src/lavlab/xnat/helpers.py new file mode 100644 index 0000000..b9fa30f --- /dev/null +++ b/src/lavlab/xnat/helpers.py @@ -0,0 +1,178 @@ +"""General helper functions for writing XNAT utilities.""" + +from contextlib import contextmanager +from typing import Union, List + +import xnat # type: ignore + +import lavlab.xnat + +LOGGER = lavlab.xnat.LOGGER.getChild("helpers") + + +## PROJECT/SUBJECT/SESSION CONTEXT HELPERS + + +def get_projects(session: xnat.XNATSession) -> List[str]: + """Get list of available project IDs. + + Parameters + ---------- + session : xnat.XNATSession + XNAT session object. + + Returns + ------- + List[str] + List of project IDs. + """ + return list(session.projects.keys()) + + +def get_subjects(session: xnat.XNATSession, project_id: str) -> List[str]: + """Get list of subject IDs for a given project. + + Parameters + ---------- + session : xnat.XNATSession + XNAT session object. + project_id : str + Project ID. + + Returns + ------- + List[str] + List of subject IDs. + """ + project = session.projects[project_id] + return list(project.subjects.keys()) + + +def get_experiments(session: xnat.XNATSession, project_id: str, subject_id: str) -> List[str]: + """Get list of experiment IDs for a given project and subject. + + Parameters + ---------- + session : xnat.XNATSession + XNAT session object. + project_id : str + Project ID. + subject_id : str + Subject ID. + + Returns + ------- + List[str] + List of experiment IDs. + """ + project = session.projects[project_id] + subject = project.subjects[subject_id] + return list(subject.experiments.keys()) + + +## DATA ACCESS HELPERS + + +def get_scan_files(session: xnat.XNATSession, experiment_id: str, scan_id: str) -> List: + """Get files associated with a scan. + + Parameters + ---------- + session : xnat.XNATSession + XNAT session object. + experiment_id : str + Experiment ID. + scan_id : str + Scan ID. + + Returns + ------- + List + List of file objects. + """ + experiment = session.experiments[experiment_id] + scan = experiment.scans[scan_id] + return list(scan.files.values()) + + +@contextmanager +def download_scan_file(session: xnat.XNATSession, experiment_id: str, scan_id: str, filename: str): + """Context manager to download and automatically clean up a scan file. + + Parameters + ---------- + session : xnat.XNATSession + XNAT session object. + experiment_id : str + Experiment ID. + scan_id : str + Scan ID. + filename : str + Name of the file to download. + + Yields + ------ + str + Path to the downloaded file. + """ + experiment = session.experiments[experiment_id] + scan = experiment.scans[scan_id] + + try: + file_path = scan.files[filename].download() + yield file_path + finally: + # Clean up downloaded file if needed + import os + if os.path.exists(file_path): + os.remove(file_path) + + +## SEARCH HELPERS + + +def find_experiments_by_type(session: xnat.XNATSession, project_id: str, experiment_type: str) -> List: + """Find experiments of a specific type within a project. + + Parameters + ---------- + session : xnat.XNATSession + XNAT session object. + project_id : str + Project ID to search within. + experiment_type : str + Type of experiment to search for (e.g., 'xnat:mrSessionData'). + + Returns + ------- + List + List of matching experiment objects. + """ + project = session.projects[project_id] + matching_experiments = [] + + for experiment in project.experiments.values(): + if experiment.attrs.get('xsi:type') == experiment_type: + matching_experiments.append(experiment) + + return matching_experiments + + +def search_experiments(session: xnat.XNATSession, **search_criteria) -> List: + """Search for experiments based on criteria. + + Parameters + ---------- + session : xnat.XNATSession + XNAT session object. + **search_criteria + Search criteria as keyword arguments. + + Returns + ------- + List + List of matching experiment objects. + """ + # This is a basic implementation - could be extended with more sophisticated search + table = session.experiments.tabulate(**search_criteria) + return table \ No newline at end of file diff --git a/test/xnat/conftest.py b/test/xnat/conftest.py new file mode 100644 index 0000000..b70cb48 --- /dev/null +++ b/test/xnat/conftest.py @@ -0,0 +1,30 @@ +# pylint: skip-file +# type: ignore +import pytest +from unittest.mock import Mock, patch + +import lavlab + + +@pytest.fixture +def mock_xnat_session(): + """ + A fixture that provides a mock XNAT session for testing. + """ + mock_session = Mock() + mock_session.projects = {} + mock_session.experiments = {} + return mock_session + + +@pytest.fixture +def xnat_service_config(): + """ + A fixture that provides XNAT service configuration for testing. + """ + return { + "name": "XNAT", + "host": "https://example-xnat.org", + "username": "testuser", + "passwd": "testpass" + } \ No newline at end of file diff --git a/test/xnat/test_helpers.py b/test/xnat/test_helpers.py new file mode 100644 index 0000000..0101f3d --- /dev/null +++ b/test/xnat/test_helpers.py @@ -0,0 +1,116 @@ +# pylint: skip-file +# type: ignore +import pytest +from unittest.mock import Mock, patch +from lavlab.xnat.helpers import ( + get_projects, + get_subjects, + get_experiments, + get_scan_files, + download_scan_file, + find_experiments_by_type, + search_experiments +) + + +class TestXNATHelpers: + """Test XNAT helper functions.""" + + def test_get_projects(self, mock_xnat_session): + """Test getting list of projects.""" + mock_xnat_session.projects = {"proj1": Mock(), "proj2": Mock()} + + projects = get_projects(mock_xnat_session) + + assert projects == ["proj1", "proj2"] + + def test_get_subjects(self, mock_xnat_session): + """Test getting list of subjects for a project.""" + mock_project = Mock() + mock_project.subjects = {"subj1": Mock(), "subj2": Mock()} + mock_xnat_session.projects = {"proj1": mock_project} + + subjects = get_subjects(mock_xnat_session, "proj1") + + assert subjects == ["subj1", "subj2"] + + def test_get_experiments(self, mock_xnat_session): + """Test getting list of experiments for a project and subject.""" + mock_experiment = Mock() + mock_subject = Mock() + mock_subject.experiments = {"exp1": mock_experiment, "exp2": mock_experiment} + mock_project = Mock() + mock_project.subjects = {"subj1": mock_subject} + mock_xnat_session.projects = {"proj1": mock_project} + + experiments = get_experiments(mock_xnat_session, "proj1", "subj1") + + assert experiments == ["exp1", "exp2"] + + def test_get_scan_files(self, mock_xnat_session): + """Test getting files for a scan.""" + mock_file1 = Mock() + mock_file2 = Mock() + mock_scan = Mock() + mock_scan.files = {"file1.dcm": mock_file1, "file2.dcm": mock_file2} + mock_experiment = Mock() + mock_experiment.scans = {"scan1": mock_scan} + mock_xnat_session.experiments = {"exp1": mock_experiment} + + files = get_scan_files(mock_xnat_session, "exp1", "scan1") + + assert files == [mock_file1, mock_file2] + + @patch('os.path.exists') + @patch('os.remove') + def test_download_scan_file(self, mock_remove, mock_exists, mock_xnat_session): + """Test downloading and cleanup of scan file.""" + mock_file = Mock() + mock_file.download.return_value = "/tmp/test_file.dcm" + mock_scan = Mock() + mock_scan.files = {"test.dcm": mock_file} + mock_experiment = Mock() + mock_experiment.scans = {"scan1": mock_scan} + mock_xnat_session.experiments = {"exp1": mock_experiment} + + mock_exists.return_value = True + + with download_scan_file(mock_xnat_session, "exp1", "scan1", "test.dcm") as file_path: + assert file_path == "/tmp/test_file.dcm" + mock_file.download.assert_called_once() + + mock_remove.assert_called_once_with("/tmp/test_file.dcm") + + def test_find_experiments_by_type(self, mock_xnat_session): + """Test finding experiments by type.""" + mock_exp1 = Mock() + mock_exp1.attrs = {'xsi:type': 'xnat:mrSessionData'} + mock_exp2 = Mock() + mock_exp2.attrs = {'xsi:type': 'xnat:ctSessionData'} + mock_exp3 = Mock() + mock_exp3.attrs = {'xsi:type': 'xnat:mrSessionData'} + + mock_project = Mock() + mock_project.experiments = { + "exp1": mock_exp1, + "exp2": mock_exp2, + "exp3": mock_exp3 + } + mock_xnat_session.projects = {"proj1": mock_project} + + mr_experiments = find_experiments_by_type(mock_xnat_session, "proj1", "xnat:mrSessionData") + + assert len(mr_experiments) == 2 + assert mock_exp1 in mr_experiments + assert mock_exp3 in mr_experiments + assert mock_exp2 not in mr_experiments + + def test_search_experiments(self, mock_xnat_session): + """Test searching experiments with criteria.""" + mock_table = Mock() + mock_xnat_session.experiments.tabulate.return_value = mock_table + + result = search_experiments(mock_xnat_session, project="test_project") + + assert result == mock_table + mock_xnat_session.experiments.tabulate.assert_called_once_with(project="test_project") \ No newline at end of file diff --git a/test/xnat/test_xnat.py b/test/xnat/test_xnat.py new file mode 100644 index 0000000..d3c2131 --- /dev/null +++ b/test/xnat/test_xnat.py @@ -0,0 +1,108 @@ +# pylint: skip-file +# type: ignore +import pytest +from unittest.mock import Mock, patch +import lavlab +from lavlab.xnat import XNATServiceProvider, connect, set_xnat_logging_level + + +class TestXNATServiceProvider: + """Test the XNAT service provider.""" + + def test_service_name(self): + """Test that the service name is correctly set.""" + provider = XNATServiceProvider() + assert provider.SERVICE == "XNAT" + + @patch('xnat.connect') + def test_login_success(self, mock_connect, xnat_service_config): + """Test successful login to XNAT server.""" + mock_session = Mock() + mock_connect.return_value = mock_session + + provider = XNATServiceProvider() + # Mock the credential provider to avoid keyring issues + provider.cred_provider = Mock() + provider.cred_provider.get_credentials.return_value = ("testuser", "testpass") + + # Mock lavlab context + with patch.object(lavlab.ctx.histology, 'service', xnat_service_config): + session = provider.login() + + assert session == mock_session + mock_connect.assert_called_once_with( + server="https://example-xnat.org", + username="testuser", + password="testpass" + ) + + @patch('xnat.connect') + def test_login_failure(self, mock_connect, xnat_service_config): + """Test login failure handling.""" + mock_connect.side_effect = Exception("Connection failed") + + provider = XNATServiceProvider() + provider.cred_provider = Mock() + provider.cred_provider.get_credentials.return_value = ("testuser", "testpass") + + with patch.object(lavlab.ctx.histology, 'service', xnat_service_config): + with pytest.raises(RuntimeError, match="Unable to connect to XNAT server"): + provider.login() + + @patch('xnat.connect') + def test_login_with_credentials_from_config(self, mock_connect, xnat_service_config): + """Test login when credentials are already in config.""" + mock_session = Mock() + mock_connect.return_value = mock_session + + provider = XNATServiceProvider() + + with patch.object(lavlab.ctx.histology, 'service', xnat_service_config): + session = provider.login() + + assert session == mock_session + mock_connect.assert_called_once_with( + server="https://example-xnat.org", + username="testuser", + password="testpass" + ) + + +class TestXNATConnect: + """Test the XNAT connect function.""" + + @patch.object(lavlab.ctx.histology, 'service_provider') + def test_connect_success(self, mock_provider, xnat_service_config): + """Test successful connection through context.""" + mock_session = Mock() + mock_provider.login.return_value = mock_session + + with patch.object(lavlab.ctx.histology, 'service', xnat_service_config): + session = connect() + + assert session == mock_session + mock_provider.login.assert_called_once() + + def test_connect_wrong_service(self): + """Test connection failure when service is not XNAT.""" + wrong_config = {"name": "OMERO"} + + with patch.object(lavlab.ctx.histology, 'service', wrong_config): + with pytest.raises(RuntimeError, match="Service is not XNAT"): + connect() + + +class TestXNATLogging: + """Test XNAT logging configuration.""" + + @patch('logging.getLogger') + @patch('logging.root.manager.loggerDict', {'xnat.test': Mock(), 'other.logger': Mock()}) + def test_set_xnat_logging_level(self, mock_get_logger): + """Test setting logging level for XNAT loggers.""" + mock_logger = Mock() + mock_get_logger.return_value = mock_logger + + set_xnat_logging_level("DEBUG") + + mock_get_logger.assert_called_with('xnat.test') + mock_logger.setLevel.assert_called_with("DEBUG") \ No newline at end of file