Merge pull request #11 from NeuroHackademy2024/first-implementation

First implementation

Author: aptinis

.github/workflows/lint.yaml

@@ -0,0 +1,51 @@
+name: Code Style
+
+on: [push, pull_request]
+
+jobs:
+  build:
+    name: "Lint Code Base"
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python 3.10
+        uses: actions/setup-python@v5
+        with:
+          # change this to a version matrix
+          python-version: '3.10'
+
+      - name: Install and configure Poetry
+        uses: snok/install-poetry@v1
+        with:
+          version: 1.8.3
+          virtualenvs-create: true
+
+      - name: Install dependencies
+        run: |
+          poetry install --only dev --no-root --no-interaction
+
+      - name: Lint with flake8
+        run: |
+          # stop the build if there are Python syntax errors or undefined names
+          poetry run flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+          poetry run flake8 . --count --max-complexity=10 --max-line-length=88 --statistics
+
+      - name: Check with black
+        run: |
+          poetry run black . --check --diff
+
+      - name: Run GitHub super-linter
+        uses: github/super-linter/slim@v6
+        env:
+          DEFAULT_BRANCH: master
+          # To report GitHub Actions status checks
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          VALIDATE_ALL_CODEBASE: true
+          VALIDATE_YAML: true
+          VALIDATE_JSON: true
+          VALIDATE_MD: true
+          VALIDATE_BASH: true
+          VALIDATE_DOCKER: true
+          VALIDATE_ENV: true

bidsinspec/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.0.1"

bidsinspec/__version__.py

@@ -0,0 +1,43 @@
+import argparse
+from pathlib import Path
+
+from bidsinspec.__version__ import __version__
+from bidsinspec.workflows.report import ReportPipeline
+
+
+def handler(args: argparse.Namespace) -> int:
+
+    bids_dir: Path = args.bids_dir
+    output_dir: Path = args.output_dir
+
+    pipeline = ReportPipeline(
+        bids_dir=bids_dir,
+        output_dir=output_dir,
+    )
+    pipeline.create_workflow()
+    pipeline.run()
+
+    return 0
+
+
+def create_parser() -> argparse.ArgumentParser:
+
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument("bids_dir", type=Path, help="The BIDS directory to inspec.")
+    parser.add_argument("output_dir", type=Path, help="The output directory.")
+    parser.add_argument("-v", "--version", action="version", version=__version__)
+
+    parser.set_defaults(handler=handler)
+
+    return parser
+
+
+def main() -> int:
+    parser = create_parser()
+    args = parser.parse_args()
+    if hasattr(args, "handler"):
+        return args.handler(args)
+
+    parser.print_help()
+    return 1

bidsinspec/cli.py

@@ -0,0 +1,32 @@
+# WIP
+# Create a custom Nipype interface that plots the bvec file on a sphere and saves figure
+from nipype.interfaces.base import BaseInterfaceInputSpec
+from nipype.interfaces.base import File
+from nipype.interfaces.base import SimpleInterface
+from nipype.interfaces.base import TraitedSpec
+from nipype.interfaces.base import traits
+
+
+class _PlotBvecsInputSpec(BaseInterfaceInputSpec):
+    in_file = File(exists=True, mandatory=True, desc="DWI bvec file to be plotted")
+    title = traits.Str(desc="a title string for the plot")
+    out_file = File("bvec.svg", usedefault=True, desc="output file name")
+
+
+class _PlotBvecsOutputSpec(TraitedSpec):
+    out_file = File(exists=True, desc="written file path")
+
+
+class PlotBvecs(SimpleInterface):
+    """Plot DWI bvecs on a sphere"""
+
+    input_spec = _PlotBvecsInputSpec
+    output_spec = _PlotBvecsOutputSpec
+
+    def _run_interface(self, runtime):
+        self._results["out_file"] = str(out_file)  # noqa wip
+
+        # need to create a function that plots bvecs on a sphere
+        # plot_bvecs()
+
+        return runtime

bidsinspec/interfaces/__init__.py

@@ -0,0 +1,64 @@
+from pathlib import Path
+
+from nipype import IdentityInterface
+from nipype.interfaces.io import BIDSDataGrabber
+from nipype.pipeline import MapNode
+from nipype.pipeline import Node
+from nipype.pipeline import Workflow
+from nireports.interfaces.fmri import FMRISummary
+from nireports.interfaces.mosaic import PlotMosaic
+
+
+class ReportPipeline:
+    def __init__(
+        self,
+        bids_dir: str | Path,
+        output_dir: str | Path,
+        name: str = "report_processing",
+    ):
+        self.name = name
+        self.bids_dir = bids_dir
+        self.base_dir = output_dir
+
+    def create_workflow(self):
+        workflow = Workflow(name=self.name, base_dir=self.base_dir)
+        workflow.config["execution"] = {"remove_unnecessary_outputs": False}
+
+        i = 0
+        input_node = Node(
+            IdentityInterface(
+                fields=[
+                    "bids_dir",
+                ],
+            ),
+            name="inputnode",
+        )
+
+        i += 1
+        bids_grab = Node(BIDSDataGrabber(), name=f"{i:02}_BIDSGrabber")
+
+        workflow.connect(input_node, "bids_dir", bids_grab, "base_dir")
+
+        i += 1
+        plot_anat = MapNode(
+            PlotMosaic(title="Raw T1w"), name=f"{i:02}_PlotAnat", iterfield="in_file"
+        )
+        workflow.connect(bids_grab, "T1w", plot_anat, "in_file")
+
+        i += 1
+        fmri_summary = MapNode(
+            FMRISummary(), name=f"{i:02}_FMRISummary", iterfield="in_func"
+        )
+        workflow.connect(bids_grab, "bold", fmri_summary, "in_func")
+
+        return workflow
+
+    def run(self):
+        workflow = self.create_workflow()
+        workflow.write_graph("pipeline_flat.dot", graph2use="flat")
+        workflow.write_graph("pipeline_hier.dot", graph2use="hierarchical")
+        workflow.write_graph("pipeline_col.dot", graph2use="colored")
+
+        in_node = workflow.get_node("inputnode")
+        in_node.inputs.bids_dir = self.bids_dir
+        workflow.run()