Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 13 additions & 11 deletions spineps/entrypoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@

import argparse
import cProfile
import os
from argparse import Namespace
from pathlib import Path
from time import perf_counter
Expand Down Expand Up @@ -276,20 +275,23 @@ def run_sample(opt: Namespace):
skipped or failed -- the reason is logged.

Raises:
ValueError: If only a filename was given instead of a path to the file.
ValueError: If the input is not a gzipped NIfTI image (``.nii.gz``).
FileNotFoundError: If the input path's parent directory is missing, or the input file does not exist.
"""
input_path = Path(opt.input).absolute()
if not input_path.exists() and not input_path.name.endswith(".nii.gz"):
input_path = input_path.with_name(input_path.name + ".nii.gz") # allow --input without the extension
if not input_path.parent.exists():
raise FileNotFoundError(f"--input/-i: the folder {input_path.parent} does not exist")
if not input_path.is_file():
raise FileNotFoundError(f"--input/-i: {input_path} does not exist or is not a file")
if not input_path.name.endswith(".nii.gz"):
raise ValueError(
f"--input/-i: SPINEPS reads gzipped NIfTI images (.nii.gz), got {input_path.name}. "
f"Compress it first, e.g. 'gzip {input_path.name}'."
)
dataset = str(input_path.parent)
if dataset == "":
raise ValueError(f"-input you only gave a filename, not a path to the file, got {input_path}")
if not os.path.exists(dataset): # noqa: PTH110
raise FileNotFoundError(f"-input parent directory does not exist, got {dataset}")
input_path = str(input_path)
if not input_path.endswith(".nii.gz"):
input_path += ".nii.gz"
if not os.path.isfile(input_path): # noqa: PTH113
raise FileNotFoundError(f"-input does not exist or is not a file, got {input_path}")
# model semantic
if looks_like_model_path(opt.model_semantic):
model_semantic = get_actual_model(opt.model_semantic, use_cpu=opt.cpu).load()
Expand Down Expand Up @@ -409,7 +411,7 @@ def run_dataset(opt: Namespace):
model_labeling = get_labeling_model(opt.model_labeling, use_cpu=opt.cpu).load()

if model_instance is None:
raise ValueError("-model_instance/-mv resolved to None; pass a valid instance model id or path")
raise ValueError("--model-instance/-mi resolved to None; pass a valid instance model id or path")

kwargs = {
"dataset_path": input_dir,
Expand Down
8 changes: 4 additions & 4 deletions spineps/seg_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,7 @@ def process_dataset( # noqa: C901

# Activate logger. Log the plain options plus the model ids -- the model objects stringify to their
# entire inference config, which floods the log file.
_not_logged = ("model_instance", "model_semantic", "model_labeling", "compatible")
_not_logged = ("model_instance", "model_semantic", "model_labeling", "compatible", "_not_logged")
args = {k: v for k, v in locals().items() if k not in _not_logged}
args["model_instance"] = model_instance.modelid()
args["model_semantic"] = [m.modelid() for m in model_semantic]
Expand Down Expand Up @@ -620,7 +620,7 @@ def segment_image( # noqa: C901
logger.print(f"Predict semantic took: {perf_counter() - start_time2:.2f} seconds", Log_Type.OK, verbose=log_inference_time)
start_time2 = perf_counter()
else:
logger.print("Subreg Mask already exists. Set -override_subreg to create it anew")
logger.print("Semantic mask already exists, loading it. Pass --override-semantic/-os to redo it")
seg_nii_modelres = NII.load(out_spine_raw, seg=True)
logger.print("seg_nii", seg_nii_modelres.zoom, seg_nii_modelres.orientation, seg_nii_modelres.shape)
# Second stage
Expand Down Expand Up @@ -651,7 +651,7 @@ def segment_image( # noqa: C901
logger.print(f"Predict instance took: {perf_counter() - start_time2:.2f} seconds", Log_Type.OK, verbose=log_inference_time)
start_time2 = perf_counter()
else:
logger.print("Vert Mask already exists. Set -override_vert to create it anew")
logger.print("Vertebra mask already exists, loading it. Pass --override-instance/-oi to redo it")
whole_vert_nii = NII.load(out_vert_raw, seg=True)

# Cleanup Step
Expand Down Expand Up @@ -716,7 +716,7 @@ def segment_image( # noqa: C901
logger.print(f"Centroids took: {perf_counter() - start_time2:.2f} seconds", Log_Type.OK, verbose=log_inference_time)
start_time2 = perf_counter()
else:
logger.print("Centroids already exists, will load instead. Set -override_ctd = True to create it anew")
logger.print("Centroids already exist, loading them. Pass --override-ctd/-oc to redo them")
ctd = POI.load(out_ctd)

# return_output_instead_of_save:
Expand Down
124 changes: 124 additions & 0 deletions unit_tests/test_cli_messages.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
# Call 'python -m unittest' on this folder
"""Every user-facing message must name a flag that exists and a cause the user can act on.

The hints told users to set flags that do not exist (`-override_subreg`, `-override_vert`,
`-override_ctd`, `-model_instance`; the real ones are `--override-semantic`, `--override-instance`,
`--override-ctd`, `--model-instance`), and `--input scan.nii` -- an uncompressed NIfTI, which SPINEPS
cannot read -- silently became `scan.nii.nii.gz` and then reported *"-input does not exist or is not a
file"* about a path the user never typed.
"""

from __future__ import annotations

import argparse
import inspect
import re
import tempfile
import unittest
from pathlib import Path

from spineps import entrypoint, seg_run


def _all_flags() -> set[str]:
"""Every option string the CLI actually accepts, including the --no-* forms argparse generates."""
parser = argparse.ArgumentParser()
entrypoint.parser_arguments(parser)
flags: set[str] = set()
for action in parser._actions:
flags.update(action.option_strings)
if isinstance(action, argparse.BooleanOptionalAction):
flags.update("--no-" + s[2:] for s in action.option_strings if s.startswith("--"))
# the per-subcommand options, which parser_arguments does not register
flags.update(
[
*("--input", "-i", "--directory", "-d", "--rawdata-name", "-rn", "--model-semantic", "-ms"),
*("--model-instance", "-mi", "-mv", "--model-labeling", "-ml", "--ignore-bids-filter", "-ibf"),
*("--ignore-model-compatibility", "-imc", "--save-log", "-sl", "--save-snaps-folder", "-ssf"),
]
)
return flags


class Test_Messages_Name_Real_Flags(unittest.TestCase):
"""Any `-flag` mentioned in a message or exception must be one the parser knows."""

FLAG_PATTERN = re.compile(r"(?<![\w-])--?[a-z][a-z0-9]*(?:[-_][a-z0-9]+)*")

def _check(self, module) -> None:
flags = _all_flags()
source = inspect.getsource(module)
# only user-facing strings, i.e. logger.print(...) / raise ...(f"...")
for line in source.splitlines():
if "logger.print(" not in line and "raise " not in line and 'f"' not in line:
continue
for quoted in re.findall(r'"([^"]*)"', line):
for match in self.FLAG_PATTERN.finditer(quoted):
candidate = match.group()
# "--model-{kind}" is assembled at runtime, so the literal "--model" prefix in front
# of the placeholder is not a flag on its own.
rest = quoted[match.end() :]
if rest.startswith("{") or (rest[:1] in "-_" and rest[1:2] == "{"):
continue
if "_" in candidate or candidate.startswith("--"):
with self.subTest(module=module.__name__, flag=candidate, line=line.strip()[:90]):
self.assertIn(candidate, flags, f"message mentions unknown flag {candidate}")

def test_entrypoint_messages(self):
self._check(entrypoint)

def test_seg_run_messages(self):
self._check(seg_run)


class Test_Input_Validation_Messages(unittest.TestCase):
@staticmethod
def _opt(input_path: str) -> argparse.Namespace:
parser = argparse.ArgumentParser()
entrypoint.parser_arguments(parser)
opt = parser.parse_args([])
opt.input = input_path
opt.model_semantic = "t2w"
opt.model_instance = "instance"
opt.model_labeling = "none"
return opt

def test_uncompressed_nifti_says_so(self):
with tempfile.TemporaryDirectory() as td:
path = Path(td) / "scan.nii"
path.write_bytes(b"")
with self.assertRaises(ValueError) as cm:
entrypoint.run_sample(self._opt(str(path)))
message = str(cm.exception)
self.assertIn("scan.nii", message)
self.assertIn(".nii.gz", message)
# and it must not invent a path the user never typed
self.assertNotIn("scan.nii.nii.gz", message)

def test_missing_file_names_the_path_given(self):
with tempfile.TemporaryDirectory() as td:
missing = Path(td) / "sub-01_T2w.nii.gz"
with self.assertRaises(FileNotFoundError) as cm:
entrypoint.run_sample(self._opt(str(missing)))
self.assertIn(str(missing), str(cm.exception))

def test_missing_folder_names_the_folder(self):
with tempfile.TemporaryDirectory() as td:
# a native path, so the assertion holds on Windows too
missing_folder = Path(td) / "does" / "not" / "exist"
with self.assertRaises(FileNotFoundError) as cm:
entrypoint.run_sample(self._opt(str(missing_folder / "scan.nii.gz")))
self.assertIn(str(missing_folder), str(cm.exception))

def test_extension_may_still_be_omitted(self):
# `--input sub-01_T2w` keeps working: the extension is appended only if nothing is there already.
with tempfile.TemporaryDirectory() as td:
path = Path(td) / "sub-01_T2w.nii.gz"
path.write_bytes(b"")
with self.assertRaises(Exception) as cm: # fails later, in model loading
entrypoint.run_sample(self._opt(str(path)[: -len(".nii.gz")]))
self.assertNotIsInstance(cm.exception, (FileNotFoundError, ValueError))


if __name__ == "__main__":
unittest.main()
Loading