diff --git a/docs/source/fedbiomed_node_part_1.md b/docs/source/fedbiomed_node_part_1.md index f52332c..22f4479 100644 --- a/docs/source/fedbiomed_node_part_1.md +++ b/docs/source/fedbiomed_node_part_1.md @@ -4,6 +4,12 @@ We will be using [Fed-BioMed](https://fedbiomed.org/) to conduct the federated l See [this page](https://fedbiomed.org/federated-network/) of the Fed-BioMed website for more information about how it works. Essentially, each site participating in the federated experiment needs to deploy a **node**, and the person responsible for running the experiment (i.e. training the models) deploys a server (**researcher** in Fed-BioMed vocabulary) to which all nodes connect to. +## Record start time + +```bash +brainflex record fedbiomed_node_part_1 start +``` + ## Deploy the node Run this to download the images (if needed) and start them: @@ -76,4 +82,13 @@ brainflex-vpn-default | info: wireguard public key for this peer: ``` Copy the public key (or the entire output of the previous command) and send it to us via email. -Then wait until you hear back before doing [Part 2](fedbiomed_node_part_2). + +## Record end time + +```bash +brainflex record fedbiomed_node_part_1 end +``` + +## Wait + +Once all the above steps are done, wait until you hear back from the FL experiment coordinator before doing [Part 2](fedbiomed_node_part_2). diff --git a/docs/source/fedbiomed_node_part_2.md b/docs/source/fedbiomed_node_part_2.md index d38c0aa..8e51c40 100644 --- a/docs/source/fedbiomed_node_part_2.md +++ b/docs/source/fedbiomed_node_part_2.md @@ -1,5 +1,11 @@ # Fed-BioMed node: Part 2 +## Record start time + +```bash +brainflex record fedbiomed_node_part_2 start +``` + ## Check VPN connection Check that the VPN server and the researcher are reachable: @@ -35,3 +41,9 @@ docker compose exec -u fedbiomed node fedbiomed node -p /fbm-node training-plan ``` Once both training plans are approved, the node is ready for training! + +## Record end time + +```bash +brainflex record fedbiomed_node_part_2 end +``` diff --git a/docs/source/fl_ready_dataset.md b/docs/source/fl_ready_dataset.md index d5478fc..d90ea33 100644 --- a/docs/source/fl_ready_dataset.md +++ b/docs/source/fl_ready_dataset.md @@ -1,5 +1,11 @@ # FL-ready dataset +## Record start time + +```bash +brainflex record fl_ready_data start +``` + ## Pipeline installation Install the Nipoppy-to-Fed-BioMed conversion pipeline: @@ -51,3 +57,9 @@ nipoppy_study/derivatives/nipoppy2fedbiomed ├── dataset.csv └── dataset_info.json ``` + +## Record end time and answer questions + +```bash +brainflex record fl_ready_data end +``` diff --git a/docs/source/freesurfer_idps.md b/docs/source/freesurfer_idps.md index 73ca113..dc4ac7f 100644 --- a/docs/source/freesurfer_idps.md +++ b/docs/source/freesurfer_idps.md @@ -1,5 +1,11 @@ # FreeSurfer IDPs +## Record start time + +```bash +brainflex record idp_extraction start +``` + ## Install pipelines Install the FreeSurfer and FreeSurfer stats extraction pipelines: @@ -95,3 +101,9 @@ nipoppy_study/derivatives/freesurfer/7.3.2/idp ├── fs7.3.2-aparc-thickness.tsv └── fs7.3.2-aseg-volume.tsv ``` + +## Record end time and answer questions + +```bash +brainflex record idp_extraction end +``` diff --git a/docs/source/initial_setup.md b/docs/source/initial_setup.md index ede0bec..cd67f3f 100644 --- a/docs/source/initial_setup.md +++ b/docs/source/initial_setup.md @@ -50,6 +50,14 @@ The above step installs three packages/tools: You can verify that they were installed successfully by running ` --help`, where `` is one of `bagel`, `nipoppy`, and `brainflex`. This should print information about the CLI commands and options. +## Initial survey + +Answer a short survey about your prior experience with Nipoppy, Neurobagel and Fed-BioMed: + +```bash +brainflex initial-survey +``` + ## Node configuration ```{attention} diff --git a/docs/source/phenotypic_files.md b/docs/source/phenotypic_files.md index 2a14d4d..cdb6cad 100644 --- a/docs/source/phenotypic_files.md +++ b/docs/source/phenotypic_files.md @@ -1,5 +1,11 @@ # Phenotypic files +## Record start time + +```bash +brainflex record semantic_harmonization start +``` + ## Annotation Download the following files to your local machine: @@ -49,3 +55,9 @@ sub-8 ses-unnamed snomed:248152002 55.0 sub-9 ses-unnamed snomed:248153007 60.0 ``` + +## Record end time and answer questions + +```bash +brainflex record semantic_harmonization end +``` diff --git a/src/brainflex/cli/console.py b/src/brainflex/cli/console.py index da59f02..24f6074 100644 --- a/src/brainflex/cli/console.py +++ b/src/brainflex/cli/console.py @@ -1,11 +1,14 @@ """Console output shared across CLI commands.""" import warnings -from collections.abc import Iterator +from collections.abc import Iterator, Sequence from contextlib import contextmanager from rich.console import Console from rich.markup import escape +from rich.text import Text + +from brainflex.utils.constants import LIKERT_CHOICES console = Console(stderr=True) @@ -15,6 +18,17 @@ def print_error(message: object) -> None: console.print(f"[red]ERROR:[/red] {escape(str(message))}", soft_wrap=True) +def format_scale(labels: Sequence[str]) -> Text: + """Format a Likert scale legend with one line per choice. + + Raises ValueError if there is not exactly one label per choice. + """ + return Text("\n").join( + Text.assemble(" ", (choice, "prompt.choices"), f" = {label}") + for choice, label in zip(LIKERT_CHOICES, labels, strict=True) + ) + + def _show_warning(message: Warning | str, *args: object, **kwargs: object) -> None: """Print a warning message to stderr.""" console.print(f"[yellow]WARNING:[/yellow] {escape(str(message))}", soft_wrap=True) diff --git a/src/brainflex/cli/initial_survey.py b/src/brainflex/cli/initial_survey.py new file mode 100644 index 0000000..53f574f --- /dev/null +++ b/src/brainflex/cli/initial_survey.py @@ -0,0 +1,43 @@ +"""CLI for ``brainflex initial-survey``.""" + +from pathlib import Path +from typing import Annotated + +import typer +from rich import get_console +from rich.prompt import Prompt + +from brainflex.cli.console import format_scale, print_error +from brainflex.cli.options import RootOption +from brainflex.core.initial_survey import check_initial_survey_path, initial_survey +from brainflex.utils.constants import ( + FAMILIARITY_LABELS, + INITIAL_SURVEY_QUESTIONS, + LIKERT_CHOICES, +) + + +def initial_survey_cli( + overwrite: Annotated[ + bool, + typer.Option("--overwrite", help="Replace existing survey answers."), + ] = False, + path: RootOption = Path(), +) -> None: + """Answer a survey on prior experience with the tools used in the experiment.""" + # check before asking questions so that answers are not lost + try: + fpath = check_initial_survey_path(path, overwrite=overwrite) + except FileExistsError as e: + print_error(e) + raise typer.Exit(1) from e + + typer.echo("Rate your familiarity with each tool:") + # same console as Prompt + get_console().print(format_scale(FAMILIARITY_LABELS), soft_wrap=True) + answers = { + question: int(Prompt.ask(question, choices=LIKERT_CHOICES)) + for question in INITIAL_SURVEY_QUESTIONS + } + initial_survey(path, answers, overwrite=overwrite) + typer.echo(f"Saved answers to {fpath}") diff --git a/src/brainflex/cli/main.py b/src/brainflex/cli/main.py index 6b04fe1..089ab0b 100644 --- a/src/brainflex/cli/main.py +++ b/src/brainflex/cli/main.py @@ -7,6 +7,7 @@ from brainflex import __version__ from brainflex.cli.console import render_warnings from brainflex.cli.get_data import get_data_cli +from brainflex.cli.initial_survey import initial_survey_cli from brainflex.cli.record import record_cli app = typer.Typer( @@ -39,4 +40,5 @@ def main( app.command("get-data")(get_data_cli) +app.command("initial-survey")(initial_survey_cli) app.command("record")(record_cli) diff --git a/src/brainflex/cli/record.py b/src/brainflex/cli/record.py index df856f8..55862ce 100644 --- a/src/brainflex/cli/record.py +++ b/src/brainflex/cli/record.py @@ -1,13 +1,23 @@ """CLI for ``brainflex record``.""" +from datetime import datetime from pathlib import Path from typing import Annotated import typer +from rich import get_console from rich.prompt import Prompt +from rich.text import Text from brainflex.cli.options import RootOption -from brainflex.core.record import Status, Step, record +from brainflex.core.record import ( + Status, + Step, + check_status, + get_step_questions, + get_step_summary, + record, +) def record_cli( @@ -16,12 +26,28 @@ def record_cli( path: RootOption = Path(), ) -> None: """Record the start or end of an experiment step.""" - notes = "" + timestamp = datetime.now().astimezone() + + # warn before asking questions + check_status(path, step, status) + + answers = {} if status == Status.END: - notes = Prompt.ask( - "Notes/feedback (press Enter to leave blank)", - default=notes, - show_default=False, - ) - record(path, step, status, notes=notes) - typer.echo(f"Recorded {step} {status}") + if (summary := get_step_summary(step)) is not None: + # same console and colour as Prompt choices + get_console().print(Text(summary, style="prompt.choices"), end="\n\n") + for question, choices in get_step_questions(step).items(): + if choices is None: + answers[question] = Prompt.ask( + f"{question} (press Enter to leave blank)", + default="", + show_default=False, + ) + else: + answers[question] = Prompt.ask(question, choices=list(choices)) + fpath_timestamps, fpath_survey = record( + path, step, status, answers=answers, timestamp=timestamp + ) + typer.echo(f"Saved timestamp for ({step}, {status}) to {fpath_timestamps}") + if fpath_survey is not None: + typer.echo(f"Saved answers to {fpath_survey}") diff --git a/src/brainflex/core/initial_survey.py b/src/brainflex/core/initial_survey.py new file mode 100644 index 0000000..8b19dbb --- /dev/null +++ b/src/brainflex/core/initial_survey.py @@ -0,0 +1,66 @@ +"""Survey on prior experience, answered before the experiment.""" + +from collections.abc import Mapping +from pathlib import Path + +import pandas as pd + +from brainflex.utils.constants import ( + BRAINFLEX_DIRNAME, + COL_ANSWER, + COL_QUESTION, + FAMILIARITY_LABELS, + INITIAL_SURVEY_FILENAME, + INITIAL_SURVEY_QUESTIONS, + LIKERT_CHOICES, + SURVEY_COLUMNS, +) +from brainflex.utils.likert import label_likert_answer + + +def check_initial_survey_path(root: Path, overwrite: bool = False) -> Path: + """Return the path to the initial survey file of the project. + + Raises FileExistsError if the file already exists and ``overwrite`` is False. + """ + fpath = root / BRAINFLEX_DIRNAME / INITIAL_SURVEY_FILENAME + if fpath.exists() and not overwrite: + raise FileExistsError( + f"Initial survey already completed: {fpath} exists." + " Use --overwrite to replace it." + ) + return fpath + + +def initial_survey( + root: Path, answers: Mapping[str, int], overwrite: bool = False +) -> None: + """Save the answers to the initial survey. + + ``answers`` maps each question in ``INITIAL_SURVEY_QUESTIONS`` to a rating on a + Likert scale from 1 to 5. Raises ValueError if questions are missing or unknown, + or if a rating is out of range, and FileExistsError if the survey file already + exists and ``overwrite`` is False. + """ + if set(answers) != set(INITIAL_SURVEY_QUESTIONS): + raise ValueError( + "Answers must be given for exactly these questions:" + f" {INITIAL_SURVEY_QUESTIONS}" + ) + for question, answer in answers.items(): + if str(answer) not in LIKERT_CHOICES: + raise ValueError( + f"Invalid answer {answer!r} to {question!r}:" + f" must be one of {', '.join(LIKERT_CHOICES)}" + ) + + fpath = check_initial_survey_path(root, overwrite=overwrite) + + rows = [ + { + COL_QUESTION: question, + COL_ANSWER: label_likert_answer(answers[question], FAMILIARITY_LABELS), + } + for question in INITIAL_SURVEY_QUESTIONS + ] + pd.DataFrame(rows, columns=SURVEY_COLUMNS).to_csv(fpath, sep="\t", index=False) diff --git a/src/brainflex/core/record.py b/src/brainflex/core/record.py index 7b3e384..861e35a 100644 --- a/src/brainflex/core/record.py +++ b/src/brainflex/core/record.py @@ -1,28 +1,39 @@ """Record the start and end of experiment steps.""" import warnings +from collections.abc import Mapping from datetime import datetime from enum import StrEnum from pathlib import Path -import pandas as pd - from brainflex.utils.constants import ( - BACKUP_TIMESTAMP_FORMAT, BRAINFLEX_DIRNAME, - COL_NOTES, + COL_ANSWER, + COL_QUESTION, COL_STATUS, COL_STEP, COL_TIMESTAMP, - STATUS_LOG_COLUMNS, - STATUS_LOG_FILENAME, + QUESTION_NOTES, + QUESTION_TIME_WITHOUT_TOOLS, + QUESTION_TOOLS_EASIER, + STEP_SURVEY_COLUMNS, + STEP_SURVEY_FILENAME, + TIME_WITHOUT_TOOLS_CHOICES, + TIMESTAMPS_COLUMNS, + TIMESTAMPS_FILENAME, + TOOLS_EASIER_CHOICES, ) +from brainflex.utils.tsv import append_tsv, load_tsv class Step(StrEnum): """Experiment steps.""" - GET_DATA = "get-data" + IDP_EXTRACTION = "idp_extraction" + SEMANTIC_HARMONIZATION = "semantic_harmonization" + FL_READY_DATA = "fl_ready_data" + FEDBIOMED_NODE_PART_1 = "fedbiomed_node_part_1" + FEDBIOMED_NODE_PART_2 = "fedbiomed_node_part_2" class Status(StrEnum): @@ -32,51 +43,131 @@ class Status(StrEnum): END = "end" -def _load_status_log(fpath: Path, now: datetime) -> pd.DataFrame: - """Load the status log, or return an empty one if it does not exist. - - If the file cannot be parsed or is missing expected columns, it is moved to a - timestamped backup and an empty log is returned. - """ - if not fpath.exists(): - return pd.DataFrame(columns=STATUS_LOG_COLUMNS) - - try: - df = pd.read_csv(fpath, sep="\t", dtype=str, keep_default_na=False) - except (pd.errors.EmptyDataError, pd.errors.ParserError): - df = None - - if df is not None and set(STATUS_LOG_COLUMNS).issubset(df.columns): - return df +DATA_PREP_STEPS = frozenset( + {Step.IDP_EXTRACTION, Step.SEMANTIC_HARMONIZATION, Step.FL_READY_DATA} +) - fpath_backup = fpath.with_name( - f"{fpath.stem}-{now.strftime(BACKUP_TIMESTAMP_FORMAT)}{fpath.suffix}" +# questions asked at the end of a step, mapped to their allowed answers +# (None for free text), in the order they are asked +DATA_PREP_QUESTIONS: dict[str, tuple[str, ...] | None] = { + QUESTION_TOOLS_EASIER: TOOLS_EASIER_CHOICES, + QUESTION_TIME_WITHOUT_TOOLS: TIME_WITHOUT_TOOLS_CHOICES, +} +COMMON_QUESTIONS: dict[str, tuple[str, ...] | None] = {QUESTION_NOTES: None} + +# high-level summaries of data preparation steps +STEP_SUMMARIES: dict[Step, str] = { + Step.IDP_EXTRACTION: ( + "In this step, you used a script that wraps FreeSurfer tools to:\n- extract " + "imaging-derived phenotypes (IDPs) (cortical thickness, subcortical volumes, " + "etc.) from existing FreeSurfer subject directories\n- combine them into " + "tabular files with standardized structure and column names (shared with every " + "other site participating in the FL experiment)" + ), + Step.SEMANTIC_HARMONIZATION: ( + "In this step, you created a file containing participant IDs, session IDs, and " + "demographic information (age, sex) that use standardized column names and " + "values (shared with every other site participating in the FL experiment), " + "despite the fact that the original demographic data may have used " + "different names or values." + ), + Step.FL_READY_DATA: ( + "In this step, you merged the previously extracted FreeSurfer IDPs and " + "harmonized phenotypic data into datasets formatted in a way that can be used " + "directly by the Fed-BioMed federated learning framework. All sites " + "participating in the FL experiment will have datasets with the same structure " + "and column names, so that they can be used together to train models." + ), +} + + +def get_step_questions(step: Step) -> dict[str, tuple[str, ...] | None]: + """Return the questions asked at the end of a step, mapped to allowed answers.""" + if step in DATA_PREP_STEPS: + return DATA_PREP_QUESTIONS | COMMON_QUESTIONS + return dict(COMMON_QUESTIONS) + + +def get_step_summary(step: Step) -> str | None: + """Return a high-level summary of a data preparation step (None otherwise).""" + return STEP_SUMMARIES.get(step) + + +def _check_answers( + step: Step, status: Status, answers: Mapping[str, str] +) -> dict[str, tuple[str, ...] | None]: + """Raise ValueError if the answers don't match the questions for the step.""" + questions = get_step_questions(step) if status == Status.END else {} + if set(answers) != set(questions): + raise ValueError( + f"Answers for {step} {status} must be given for exactly these" + f" questions: {list(questions)}" + ) + for question, answer in answers.items(): + choices = questions[question] + if choices is not None and answer not in choices: + raise ValueError( + f"Invalid answer {answer!r} to {question!r}:" + f" must be one of {', '.join(choices)}" + ) + return questions + + +def check_status(root: Path, step: Step, status: Status) -> None: + """Warn if a step is started twice or ended without being started.""" + df_timestamps = load_tsv( + root / BRAINFLEX_DIRNAME / TIMESTAMPS_FILENAME, TIMESTAMPS_COLUMNS ) - fpath.rename(fpath_backup) - warnings.warn(f"invalid status log moved to {fpath_backup}", stacklevel=3) - return pd.DataFrame(columns=STATUS_LOG_COLUMNS) - - -def record(root: Path, step: Step, status: Status, notes: str = "") -> None: - """Add a timestamped row to the status log, creating it if needed. - - Warns if a step is started twice or ended without being started. - """ - now = datetime.now().astimezone() - df = _load_status_log(fpath := root / BRAINFLEX_DIRNAME / STATUS_LOG_FILENAME, now) - - statuses = df.query(f"{COL_STEP} == @step")[COL_STATUS] + statuses = df_timestamps.query(f"{COL_STEP} == @step")[COL_STATUS] last_status = statuses.iloc[-1] if not statuses.empty else None if status == Status.START and last_status == Status.START: warnings.warn(f"step {step} was already started", stacklevel=2) elif status == Status.END and last_status != Status.START: warnings.warn(f"step {step} was not started", stacklevel=2) + +def record( + root: Path, + step: Step, + status: Status, + answers: Mapping[str, str] | None = None, + timestamp: datetime | None = None, +) -> tuple[Path, Path | None]: + """Record the start or end of a step + some survey answers. + + Call ``check_status`` first to warn about steps started twice or ended without + being started. Returns the paths to the timestamps file and to the survey file + (if applicable). + """ + answers = answers or {} + questions = _check_answers(step, status, answers) + + dpath = root / BRAINFLEX_DIRNAME + fpath_timestamps = dpath / TIMESTAMPS_FILENAME + timestamp = timestamp or datetime.now() + if timestamp.tzinfo is None: + timestamp = timestamp.astimezone() # naive -> interpret as local time row = { - COL_TIMESTAMP: now.isoformat(timespec="seconds"), + COL_TIMESTAMP: timestamp.isoformat(timespec="seconds"), COL_STEP: str(step), COL_STATUS: str(status), - COL_NOTES: notes, } - df = pd.concat([df, pd.DataFrame([row], columns=STATUS_LOG_COLUMNS)]) - df.to_csv(fpath, sep="\t", index=False) + append_tsv(fpath_timestamps, TIMESTAMPS_COLUMNS, [row]) + + if not questions: + return fpath_timestamps, None + + fpath_survey = dpath / STEP_SURVEY_FILENAME + append_tsv( + fpath_survey, + STEP_SURVEY_COLUMNS, + ( + { + COL_STEP: str(step), + COL_QUESTION: q, + COL_ANSWER: answers[q], + } + for q in questions + ), + ) + return fpath_timestamps, fpath_survey diff --git a/src/brainflex/utils/constants.py b/src/brainflex/utils/constants.py index 567af0d..d15b033 100644 --- a/src/brainflex/utils/constants.py +++ b/src/brainflex/utils/constants.py @@ -4,14 +4,44 @@ NIPOPPY_STUDY_DIRNAME = "nipoppy_study" BRAINFLEX_DIRNAME = ".brainflex" -# record -STATUS_LOG_FILENAME = "status_log.tsv" +# surveys +COL_QUESTION = "question" +COL_ANSWER = "answer" +SURVEY_COLUMNS = (COL_QUESTION, COL_ANSWER) +LIKERT_CHOICES = ("1", "2", "3", "4", "5") +FAMILIARITY_LABELS = ( + "I have never heard of it before this project", + "I have heard of it but never used", + "I have used but only with example/toy data", + "I have used with real data but not regularly", + "I use it regularly / It is one of my go-to tools", +) BACKUP_TIMESTAMP_FORMAT = "%Y%m%dT%H%M%S" + +# initial-survey +INITIAL_SURVEY_FILENAME = "initial_survey.tsv" +INITIAL_SURVEY_QUESTIONS = tuple( + f"How familiar are you with {tool}?" + for tool in ("Nipoppy", "Neurobagel", "Fed-BioMed") +) + +# record +TIMESTAMPS_FILENAME = "timestamps.tsv" COL_TIMESTAMP = "timestamp" COL_STEP = "step" COL_STATUS = "status" -COL_NOTES = "notes" -STATUS_LOG_COLUMNS = (COL_TIMESTAMP, COL_STEP, COL_STATUS, COL_NOTES) +TIMESTAMPS_COLUMNS = (COL_TIMESTAMP, COL_STEP, COL_STATUS) +STEP_SURVEY_FILENAME = "step_survey.tsv" +STEP_SURVEY_COLUMNS = (COL_STEP, *SURVEY_COLUMNS) +QUESTION_TOOLS_EASIER = ( + "Did the Nipoppy/Neurobagel tools make things easier to do this step?" +) +TOOLS_EASIER_CHOICES = ("yes", "no", "unsure") +QUESTION_TIME_WITHOUT_TOOLS = ( + "How long do you think this would have taken without the Nipoppy/Neurobagel tools?" +) +TIME_WITHOUT_TOOLS_CHOICES = ("<1h", "1-4h", "4-8h", "1-3d", "3-7d", ">7d") +QUESTION_NOTES = "Notes/feedback" # get-data OPENNEURO_BUCKET = "openneuro.org" diff --git a/src/brainflex/utils/likert.py b/src/brainflex/utils/likert.py new file mode 100644 index 0000000..03076a5 --- /dev/null +++ b/src/brainflex/utils/likert.py @@ -0,0 +1,23 @@ +"""Likert scale utilities.""" + +from collections.abc import Sequence + +from brainflex.utils.constants import LIKERT_CHOICES + + +def label_likert_answer(answer: str | int, labels: Sequence[str]) -> str: + """Return a Likert scale answer with its label. + + Raises ValueError if the answer is not a valid choice or if there is not + exactly one label per choice. + """ + if len(labels) != len(LIKERT_CHOICES): + raise ValueError( + f"Expected {len(LIKERT_CHOICES)} labels, got {len(labels)}: {labels}" + ) + answer = str(answer) + if answer not in LIKERT_CHOICES: + raise ValueError( + f"Invalid answer {answer!r}: must be one of {', '.join(LIKERT_CHOICES)}" + ) + return f"{answer} ({labels[LIKERT_CHOICES.index(answer)]})" diff --git a/src/brainflex/utils/tsv.py b/src/brainflex/utils/tsv.py new file mode 100644 index 0000000..7ba20f6 --- /dev/null +++ b/src/brainflex/utils/tsv.py @@ -0,0 +1,49 @@ +"""TSV file utilities.""" + +import warnings +from collections.abc import Iterable, Mapping, Sequence +from datetime import datetime +from pathlib import Path + +import pandas as pd + +from brainflex.utils.constants import BACKUP_TIMESTAMP_FORMAT + + +def load_tsv(fpath: Path, columns: Sequence[str]) -> pd.DataFrame: + """Load a TSV file, or return an empty one if it does not exist. + + If the file cannot be parsed or is missing expected columns, it is moved to a + timestamped backup and an empty dataframe is returned. + """ + if not fpath.exists(): + return pd.DataFrame(columns=columns) + + try: + df = pd.read_csv(fpath, sep="\t", dtype=str, keep_default_na=False) + except (pd.errors.EmptyDataError, pd.errors.ParserError): + df = None + + if df is not None and set(columns).issubset(df.columns): + return df + + now = datetime.now().astimezone() + fpath_backup = fpath.with_name( + f"{fpath.stem}-{now.strftime(BACKUP_TIMESTAMP_FORMAT)}{fpath.suffix}" + ) + fpath.rename(fpath_backup) + warnings.warn(f"invalid file {fpath} moved to {fpath_backup}", stacklevel=3) + return pd.DataFrame(columns=columns) + + +def append_tsv( + fpath: Path, + columns: Sequence[str], + rows: Iterable[Mapping[str, str]], + df: pd.DataFrame | None = None, +) -> None: + """Append rows to a TSV file, creating it if needed.""" + if df is None: + df = load_tsv(fpath, columns) + df = pd.concat([df, pd.DataFrame(list(rows), columns=columns)]) + df.to_csv(fpath, sep="\t", index=False) diff --git a/tests/cli/test_console.py b/tests/cli/test_console.py index dd89c33..2d9fba5 100644 --- a/tests/cli/test_console.py +++ b/tests/cli/test_console.py @@ -2,7 +2,32 @@ import pytest -from brainflex.cli.console import print_error, render_warnings +from brainflex.cli.console import format_scale, print_error, render_warnings + + +class TestFormatScale: + def test_format_scale(self): + assert format_scale(["a", "b", "c", "d", "e"]).plain.splitlines() == [ + " 1 = a", + " 2 = b", + " 3 = c", + " 4 = d", + " 5 = e", + ] + + def test_choices_styled_like_prompt(self): + text = format_scale(["a", "b", "c", "d", "e"]) + styled = [ + text.plain[span.start : span.end] + for span in text.spans + if span.style == "prompt.choices" + ] + assert styled == ["1", "2", "3", "4", "5"] + + @pytest.mark.parametrize("labels", [["a", "b", "c", "d"], list("abcdef")]) + def test_wrong_number_of_labels(self, labels: list[str]): + with pytest.raises(ValueError, match="zip"): + format_scale(labels) class TestPrintError: diff --git a/tests/cli/test_initial_survey.py b/tests/cli/test_initial_survey.py new file mode 100644 index 0000000..569f161 --- /dev/null +++ b/tests/cli/test_initial_survey.py @@ -0,0 +1,125 @@ +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from pytest_mock import MockerFixture +from typer.testing import CliRunner + +from brainflex.cli.main import app +from brainflex.utils.constants import ( + BRAINFLEX_DIRNAME, + FAMILIARITY_LABELS, + INITIAL_SURVEY_FILENAME, + INITIAL_SURVEY_QUESTIONS, +) + + +@pytest.fixture +def mock_initial_survey(mocker: MockerFixture) -> MagicMock: + return mocker.patch("brainflex.cli.initial_survey.initial_survey") + + +def _answers(*ratings: int) -> dict[str, int]: + return dict(zip(INITIAL_SURVEY_QUESTIONS, ratings, strict=True)) + + +@pytest.fixture +def fpath_survey(dpath_root: Path) -> Path: + return dpath_root / BRAINFLEX_DIRNAME / INITIAL_SURVEY_FILENAME + + +class TestInitialSurvey: + def test_initial_survey( + self, runner: CliRunner, mock_initial_survey: MagicMock, dpath_root: Path + ): + result = runner.invoke( + app, + ["initial-survey", "--project", str(dpath_root)], + input="1\n3\n5\n", + ) + assert result.exit_code == 0 + for tool in ("Nipoppy", "Neurobagel", "Fed-BioMed"): + assert f"How familiar are you with {tool}?" in result.stdout + # legend is printed once, before the questions + for i, label in enumerate(FAMILIARITY_LABELS, start=1): + assert result.stdout.count(f"{i} = {label}") == 1 + assert result.stdout.index(label) < result.stdout.index("How familiar") + assert "Saved answers to" in result.stdout + mock_initial_survey.assert_called_once_with( + dpath_root.resolve(), _answers(1, 3, 5), overwrite=False + ) + + def test_invalid_input_reprompts( + self, runner: CliRunner, mock_initial_survey: MagicMock, dpath_root: Path + ): + result = runner.invoke( + app, + ["initial-survey", "--project", str(dpath_root)], + input="0\nabc\n2\n3\n4\n", + ) + assert result.exit_code == 0 + mock_initial_survey.assert_called_once_with( + dpath_root.resolve(), _answers(2, 3, 4), overwrite=False + ) + + def test_default_path( + self, + runner: CliRunner, + mock_initial_survey: MagicMock, + dpath_subdir: Path, + dpath_root: Path, + monkeypatch: pytest.MonkeyPatch, + ): + monkeypatch.chdir(dpath_subdir) + result = runner.invoke(app, ["initial-survey"], input="1\n1\n1\n") + assert result.exit_code == 0 + mock_initial_survey.assert_called_once_with( + dpath_root.resolve(), _answers(1, 1, 1), overwrite=False + ) + + def test_exists( + self, + runner: CliRunner, + mock_initial_survey: MagicMock, + dpath_root: Path, + fpath_survey: Path, + ): + fpath_survey.write_text("existing") + result = runner.invoke(app, ["initial-survey", "--project", str(dpath_root)]) + assert result.exit_code == 1 + assert "How familiar" not in result.stdout + assert "ERROR" in result.stderr + assert INITIAL_SURVEY_FILENAME in result.stderr + mock_initial_survey.assert_not_called() + + def test_exists_overwrite( + self, + runner: CliRunner, + mock_initial_survey: MagicMock, + dpath_root: Path, + fpath_survey: Path, + ): + fpath_survey.write_text("existing") + result = runner.invoke( + app, + ["initial-survey", "--overwrite", "--project", str(dpath_root)], + input="1\n1\n1\n", + ) + assert result.exit_code == 0 + mock_initial_survey.assert_called_once_with( + dpath_root.resolve(), _answers(1, 1, 1), overwrite=True + ) + + def test_end_to_end(self, runner: CliRunner, dpath_root: Path, fpath_survey: Path): + result = runner.invoke( + app, + ["initial-survey", "--project", str(dpath_root)], + input="1\n3\n5\n", + ) + assert result.exit_code == 0 + assert fpath_survey.read_text().splitlines() == [ + "question\tanswer", + "How familiar are you with Nipoppy?\t1 (I have never heard of it before this project)", + "How familiar are you with Neurobagel?\t3 (I have used but only with example/toy data)", + "How familiar are you with Fed-BioMed?\t5 (I use it regularly / It is one of my go-to tools)", + ] diff --git a/tests/cli/test_options.py b/tests/cli/test_options.py index bf54595..26d7e90 100644 --- a/tests/cli/test_options.py +++ b/tests/cli/test_options.py @@ -12,7 +12,8 @@ class TestRootOption: ("args", "target"), [ (["get-data", "ds000001"], "brainflex.cli.get_data.get_data"), - (["record", "get-data", "start"], "brainflex.cli.record.record"), + (["initial-survey"], "brainflex.cli.initial_survey.initial_survey"), + (["record", "idp_extraction", "start"], "brainflex.cli.record.record"), ], ) @pytest.mark.parametrize("explicit", [True, False]) diff --git a/tests/cli/test_record.py b/tests/cli/test_record.py index b55187d..d5cb019 100644 --- a/tests/cli/test_record.py +++ b/tests/cli/test_record.py @@ -1,36 +1,65 @@ -import warnings from pathlib import Path -from unittest.mock import MagicMock +from unittest.mock import ANY, MagicMock +import pandas as pd import pytest from pytest_mock import MockerFixture from typer.testing import CliRunner from brainflex.cli.main import app from brainflex.core.record import Status, Step +from brainflex.utils.constants import ( + BRAINFLEX_DIRNAME, + QUESTION_NOTES, + QUESTION_TIME_WITHOUT_TOOLS, + QUESTION_TOOLS_EASIER, + STEP_SURVEY_FILENAME, + TIMESTAMPS_FILENAME, +) + + +def _unwrap(text: str) -> str: + """Collapse whitespace so that prompts wrapped by rich can be matched.""" + return " ".join(text.split()) + + +FPATH_TIMESTAMPS = Path("timestamps.tsv") +FPATH_SURVEY = Path("step_survey.tsv") + + +def _fake_record(*args, answers: dict[str, str], **kwargs) -> tuple[Path, Path | None]: + return FPATH_TIMESTAMPS, FPATH_SURVEY if answers else None @pytest.fixture def mock_record(mocker: MockerFixture) -> MagicMock: - return mocker.patch("brainflex.cli.record.record") + return mocker.patch("brainflex.cli.record.record", side_effect=_fake_record) class TestRecord: def test_start(self, runner: CliRunner, mock_record: MagicMock, dpath_root: Path): result = runner.invoke( - app, ["record", "get-data", "start", "--project", str(dpath_root)] + app, ["record", "idp_extraction", "start", "--project", str(dpath_root)] ) assert result.exit_code == 0 - assert "Notes" not in result.stdout - assert "Recorded get-data start" in result.stdout + assert "?" not in result.stdout + assert ( + f"Saved timestamp for (idp_extraction, start) to {FPATH_TIMESTAMPS}" + in result.stdout + ) + assert "Saved answers" not in result.stdout mock_record.assert_called_once_with( - dpath_root.resolve(), Step.GET_DATA, Status.START, notes="" + dpath_root.resolve(), + Step.IDP_EXTRACTION, + Status.START, + answers={}, + timestamp=ANY, ) @pytest.mark.parametrize( ("user_input", "notes"), [("went fine\n", "went fine"), ("\n", "")] ) - def test_end( + def test_end_other( self, runner: CliRunner, mock_record: MagicMock, @@ -40,13 +69,106 @@ def test_end( ): result = runner.invoke( app, - ["record", "get-data", "end", "--project", str(dpath_root)], + ["record", "fedbiomed_node_part_1", "end", "--project", str(dpath_root)], input=user_input, ) assert result.exit_code == 0 - assert "Notes" in result.stdout + stdout = _unwrap(result.stdout) + assert QUESTION_NOTES in stdout + assert QUESTION_TOOLS_EASIER not in stdout + assert QUESTION_TIME_WITHOUT_TOOLS not in stdout + assert ( + f"Saved timestamp for (fedbiomed_node_part_1, end) to {FPATH_TIMESTAMPS}" + in result.stdout + ) + assert f"Saved answers to {FPATH_SURVEY}" in result.stdout mock_record.assert_called_once_with( - dpath_root.resolve(), Step.GET_DATA, Status.END, notes=notes + dpath_root.resolve(), + Step.FEDBIOMED_NODE_PART_1, + Status.END, + answers={QUESTION_NOTES: notes}, + timestamp=ANY, + ) + + @pytest.mark.parametrize( + "step", [Step.IDP_EXTRACTION, Step.SEMANTIC_HARMONIZATION, Step.FL_READY_DATA] + ) + def test_end_data_prep( + self, runner: CliRunner, mock_record: MagicMock, dpath_root: Path, step: Step + ): + result = runner.invoke( + app, + ["record", str(step), "end", "--project", str(dpath_root)], + input="yes\n1-4h\nwent fine\n", + ) + assert result.exit_code == 0 + stdout = _unwrap(result.stdout) + # questions are asked in order, with the free-form question last + assert ( + stdout.index(QUESTION_TOOLS_EASIER) + < stdout.index(QUESTION_TIME_WITHOUT_TOOLS) + < stdout.index(QUESTION_NOTES) + ) + assert "[yes/no/unsure]" in stdout + assert ( + f"Saved timestamp for ({step}, end) to {FPATH_TIMESTAMPS}" in result.stdout + ) + assert f"Saved answers to {FPATH_SURVEY}" in result.stdout + mock_record.assert_called_once_with( + dpath_root.resolve(), + step, + Status.END, + answers={ + QUESTION_TOOLS_EASIER: "yes", + QUESTION_TIME_WITHOUT_TOOLS: "1-4h", + QUESTION_NOTES: "went fine", + }, + timestamp=ANY, + ) + + def test_end_data_prep_invalid_input_reprompts( + self, runner: CliRunner, mock_record: MagicMock, dpath_root: Path + ): + result = runner.invoke( + app, + ["record", "idp_extraction", "end", "--project", str(dpath_root)], + input="maybe\nunsure\n2 days\n>7d\n\n", + ) + assert result.exit_code == 0 + mock_record.assert_called_once_with( + dpath_root.resolve(), + Step.IDP_EXTRACTION, + Status.END, + answers={ + QUESTION_TOOLS_EASIER: "unsure", + QUESTION_TIME_WITHOUT_TOOLS: ">7d", + QUESTION_NOTES: "", + }, + timestamp=ANY, + ) + + def test_timestamp_before_prompts( + self, + runner: CliRunner, + mock_record: MagicMock, + dpath_root: Path, + mocker: MockerFixture, + ): + manager = MagicMock() + mock_datetime = mocker.patch("brainflex.cli.record.datetime") + mock_ask = mocker.patch("brainflex.cli.record.Prompt.ask", return_value="") + manager.attach_mock(mock_datetime.now, "now") + manager.attach_mock(mock_ask, "ask") + result = runner.invoke( + app, + ["record", "fedbiomed_node_part_1", "end", "--project", str(dpath_root)], + ) + assert result.exit_code == 0 + names = [name for name, *_ in manager.mock_calls] + assert names.index("now") < names.index("ask") + assert ( + mock_record.call_args.kwargs["timestamp"] + is mock_datetime.now.return_value.astimezone.return_value ) def test_default_path( @@ -57,10 +179,14 @@ def test_default_path( monkeypatch: pytest.MonkeyPatch, ): monkeypatch.chdir(dpath_root) - result = runner.invoke(app, ["record", "get-data", "start"]) + result = runner.invoke(app, ["record", "idp_extraction", "start"]) assert result.exit_code == 0 mock_record.assert_called_once_with( - dpath_root.resolve(), Step.GET_DATA, Status.START, notes="" + dpath_root.resolve(), + Step.IDP_EXTRACTION, + Status.START, + answers={}, + timestamp=ANY, ) @pytest.mark.parametrize("explicit", [True, False]) @@ -78,18 +204,23 @@ def test_subdirectory( app, [ "record", - "get-data", + "idp_extraction", "start", *(["--project", str(dpath_subdir)] if explicit else []), ], ) assert result.exit_code == 0 mock_record.assert_called_once_with( - dpath_root.resolve(), Step.GET_DATA, Status.START, notes="" + dpath_root.resolve(), + Step.IDP_EXTRACTION, + Status.START, + answers={}, + timestamp=ANY, ) @pytest.mark.parametrize( - "args", [["bad-step", "start"], ["get-data", "bad-status"]] + "args", + [["get-data", "start"], ["bad-step", "start"], ["idp_extraction", "bad"]], ) def test_invalid_choice( self, runner: CliRunner, mock_record: MagicMock, args: list[str] @@ -98,12 +229,59 @@ def test_invalid_choice( assert result.exit_code == 2 mock_record.assert_not_called() - def test_warning(self, runner: CliRunner, mock_record: MagicMock, dpath_root: Path): - mock_record.side_effect = lambda *args, **kwargs: warnings.warn( - "step get-data was already started", stacklevel=2 + def test_warning_already_started( + self, runner: CliRunner, mock_record: MagicMock, dpath_root: Path + ): + args = ["record", "idp_extraction", "start", "--project", str(dpath_root)] + mock_record.side_effect = None + (dpath_root / BRAINFLEX_DIRNAME / TIMESTAMPS_FILENAME).write_text( + "timestamp\tstep\tstatus\nt0\tidp_extraction\tstart\n" ) + mock_record.return_value = (FPATH_TIMESTAMPS, None) + result = runner.invoke(app, args) + assert result.exit_code == 0 + assert "WARNING: step idp_extraction was already started" in result.stderr + + @pytest.mark.parametrize("step", ["idp_extraction", "fedbiomed_node_part_1"]) + def test_warning_not_started_printed_first( + self, runner: CliRunner, mock_record: MagicMock, dpath_root: Path, step: str + ): result = runner.invoke( - app, ["record", "get-data", "start", "--project", str(dpath_root)] + app, + ["record", step, "end", "--project", str(dpath_root)], + input="yes\n1-4h\n\n", + ) + assert result.exit_code == 0 + assert result.output.startswith(f"WARNING: step {step} was not started\n") + mock_record.assert_called_once() + + def test_end_to_end(self, runner: CliRunner, dpath_root: Path): + args = ["record", "fl_ready_data", "--project", str(dpath_root)] + args_start = [*args[:2], "start", *args[2:]] + args_end = [*args[:2], "end", *args[2:]] + dpath = dpath_root.resolve() / BRAINFLEX_DIRNAME + result = runner.invoke(app, args_start) + assert result.exit_code == 0 + fpath_timestamps = dpath / TIMESTAMPS_FILENAME + assert ( + f"Saved timestamp for (fl_ready_data, start) to {fpath_timestamps}" + in result.stdout ) + assert "Saved answers" not in result.stdout + result = runner.invoke(app, args_end, input="yes\n<1h\nok\n") + assert "WARNING" not in result.stderr assert result.exit_code == 0 - assert "WARNING: step get-data was already started" in result.stderr + assert ( + f"Saved timestamp for (fl_ready_data, end) to {fpath_timestamps}" + in result.stdout + ) + assert f"Saved answers to {dpath / STEP_SURVEY_FILENAME}" in result.stdout + df_timestamps = pd.read_csv(dpath / TIMESTAMPS_FILENAME, sep="\t", dtype=str) + assert list(df_timestamps.columns) == ["timestamp", "step", "status"] + assert list(df_timestamps["status"]) == ["start", "end"] + assert (dpath / STEP_SURVEY_FILENAME).read_text().splitlines() == [ + "step\tquestion\tanswer", + f"fl_ready_data\t{QUESTION_TOOLS_EASIER}\tyes", + f"fl_ready_data\t{QUESTION_TIME_WITHOUT_TOOLS}\t<1h", + f"fl_ready_data\t{QUESTION_NOTES}\tok", + ] diff --git a/tests/core/test_initial_survey.py b/tests/core/test_initial_survey.py new file mode 100644 index 0000000..06e91e3 --- /dev/null +++ b/tests/core/test_initial_survey.py @@ -0,0 +1,108 @@ +from pathlib import Path + +import pandas as pd +import pytest + +from brainflex.core.initial_survey import check_initial_survey_path, initial_survey +from brainflex.utils.constants import ( + BRAINFLEX_DIRNAME, + INITIAL_SURVEY_FILENAME, + INITIAL_SURVEY_QUESTIONS, + SURVEY_COLUMNS, +) + + +@pytest.fixture +def root(tmp_path: Path) -> Path: + (tmp_path / BRAINFLEX_DIRNAME).mkdir() + return tmp_path + + +@pytest.fixture +def fpath_survey(root: Path) -> Path: + return root / BRAINFLEX_DIRNAME / INITIAL_SURVEY_FILENAME + + +def _read(fpath: Path) -> pd.DataFrame: + return pd.read_csv(fpath, sep="\t", dtype=str, keep_default_na=False) + + +def _answers(*ratings: int) -> dict[str, int]: + return dict(zip(INITIAL_SURVEY_QUESTIONS, ratings, strict=True)) + + +class TestCheckInitialSurveyPath: + def test_not_exists(self, root: Path, fpath_survey: Path): + assert check_initial_survey_path(root) == fpath_survey + + def test_exists(self, root: Path, fpath_survey: Path): + fpath_survey.write_text("existing") + with pytest.raises(FileExistsError, match=str(fpath_survey)): + check_initial_survey_path(root) + + def test_exists_overwrite(self, root: Path, fpath_survey: Path): + fpath_survey.write_text("existing") + assert check_initial_survey_path(root, overwrite=True) == fpath_survey + + +class TestInitialSurvey: + def test_creates_file(self, root: Path, fpath_survey: Path): + initial_survey(root, _answers(1, 3, 5)) + df = _read(fpath_survey) + assert tuple(df.columns) == SURVEY_COLUMNS + assert list(df["question"]) == [ + "How familiar are you with Nipoppy?", + "How familiar are you with Neurobagel?", + "How familiar are you with Fed-BioMed?", + ] + assert list(df["answer"]) == [ + "1 (I have never heard of it before this project)", + "3 (I have used but only with example/toy data)", + "5 (I use it regularly / It is one of my go-to tools)", + ] + + def test_question_order(self, root: Path, fpath_survey: Path): + answers = dict(reversed(_answers(1, 3, 5).items())) + initial_survey(root, answers) + df = _read(fpath_survey) + assert list(df["question"]) == list(INITIAL_SURVEY_QUESTIONS) + assert list(df["answer"]) == [ + "1 (I have never heard of it before this project)", + "3 (I have used but only with example/toy data)", + "5 (I use it regularly / It is one of my go-to tools)", + ] + + def test_exists(self, root: Path, fpath_survey: Path): + fpath_survey.write_text("existing") + with pytest.raises(FileExistsError, match="--overwrite"): + initial_survey(root, _answers(1, 1, 1)) + assert fpath_survey.read_text() == "existing" + + def test_overwrite(self, root: Path, fpath_survey: Path): + initial_survey(root, _answers(1, 1, 1)) + initial_survey(root, _answers(2, 2, 2), overwrite=True) + df = _read(fpath_survey) + assert len(df) == len(INITIAL_SURVEY_QUESTIONS) + assert list(df["answer"]) == ["2 (I have heard of it but never used)"] * 3 + + @pytest.mark.parametrize("ratings", [(0, 1, 1), (1, 6, 1), (1, 1, -1)]) + def test_invalid_rating( + self, root: Path, fpath_survey: Path, ratings: tuple[int, int, int] + ): + with pytest.raises(ValueError, match="must be one of"): + initial_survey(root, _answers(*ratings)) + assert not fpath_survey.exists() + + @pytest.mark.parametrize( + "answers", + [ + dict(list(_answers(1, 1, 1).items())[:2]), + {**_answers(1, 1, 1), "Unknown question?": 1}, + ], + ) + def test_invalid_questions( + self, root: Path, fpath_survey: Path, answers: dict[str, int] + ): + with pytest.raises(ValueError, match="exactly these questions"): + initial_survey(root, answers) + assert not fpath_survey.exists() diff --git a/tests/core/test_record.py b/tests/core/test_record.py index cda572c..e5f9a60 100644 --- a/tests/core/test_record.py +++ b/tests/core/test_record.py @@ -1,17 +1,35 @@ import warnings -from datetime import datetime +from datetime import datetime, timedelta, timezone from pathlib import Path import pandas as pd import pytest -from brainflex.core.record import Status, Step, record +from brainflex.core.record import ( + DATA_PREP_STEPS, + Status, + Step, + check_status, + get_step_questions, + record, +) from brainflex.utils.constants import ( BRAINFLEX_DIRNAME, - STATUS_LOG_COLUMNS, - STATUS_LOG_FILENAME, + QUESTION_NOTES, + QUESTION_TIME_WITHOUT_TOOLS, + QUESTION_TOOLS_EASIER, + STEP_SURVEY_COLUMNS, + STEP_SURVEY_FILENAME, + TIMESTAMPS_COLUMNS, + TIMESTAMPS_FILENAME, ) +DATA_PREP_ANSWERS = { + QUESTION_TOOLS_EASIER: "yes", + QUESTION_TIME_WITHOUT_TOOLS: "1-4h", + QUESTION_NOTES: "went fine", +} + @pytest.fixture def root(tmp_path: Path) -> Path: @@ -20,99 +38,241 @@ def root(tmp_path: Path) -> Path: @pytest.fixture -def fpath_log(root: Path) -> Path: - return root / BRAINFLEX_DIRNAME / STATUS_LOG_FILENAME +def fpath_timestamps(root: Path) -> Path: + return root / BRAINFLEX_DIRNAME / TIMESTAMPS_FILENAME + + +@pytest.fixture +def fpath_survey(root: Path) -> Path: + return root / BRAINFLEX_DIRNAME / STEP_SURVEY_FILENAME def _read(fpath: Path) -> pd.DataFrame: return pd.read_csv(fpath, sep="\t", dtype=str, keep_default_na=False) +class TestGetStepQuestions: + @pytest.mark.parametrize("step", sorted(DATA_PREP_STEPS)) + def test_data_prep(self, step: Step): + assert list(get_step_questions(step)) == [ + QUESTION_TOOLS_EASIER, + QUESTION_TIME_WITHOUT_TOOLS, + QUESTION_NOTES, + ] + + @pytest.mark.parametrize("step", sorted(set(Step) - DATA_PREP_STEPS)) + def test_other(self, step: Step): + assert list(get_step_questions(step)) == [QUESTION_NOTES] + + class TestRecord: - def test_creates_file(self, root: Path, fpath_log: Path): - record(root, Step.GET_DATA, Status.START) - df = _read(fpath_log) - assert tuple(df.columns) == STATUS_LOG_COLUMNS + def test_start(self, root: Path, fpath_timestamps: Path, fpath_survey: Path): + assert record(root, Step.IDP_EXTRACTION, Status.START) == ( + fpath_timestamps, + None, + ) + df = _read(fpath_timestamps) + assert tuple(df.columns) == TIMESTAMPS_COLUMNS assert len(df) == 1 row = df.iloc[0] - assert row["step"] == "get-data" + assert row["step"] == "idp_extraction" assert row["status"] == "start" - assert row["notes"] == "" assert datetime.fromisoformat(row["timestamp"]).tzinfo is not None + assert not fpath_survey.exists() + + def test_timestamp(self, root: Path, fpath_timestamps: Path): + timestamp = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone(timedelta(hours=-5))) + record(root, Step.IDP_EXTRACTION, Status.START, timestamp=timestamp) + assert _read(fpath_timestamps)["timestamp"].iloc[0] == ( + "2026-01-02T03:04:05-05:00" + ) + + def test_end_data_prep( + self, root: Path, fpath_timestamps: Path, fpath_survey: Path + ): + record(root, Step.IDP_EXTRACTION, Status.START) + assert record( + root, Step.IDP_EXTRACTION, Status.END, answers=DATA_PREP_ANSWERS + ) == (fpath_timestamps, fpath_survey) + assert list(_read(fpath_timestamps)["status"]) == ["start", "end"] + assert fpath_timestamps.read_text().count("timestamp") == 1 + df = _read(fpath_survey) + assert tuple(df.columns) == STEP_SURVEY_COLUMNS + assert df.to_dict("records") == [ + { + "step": "idp_extraction", + "question": QUESTION_TOOLS_EASIER, + "answer": "yes", + }, + { + "step": "idp_extraction", + "question": QUESTION_TIME_WITHOUT_TOOLS, + "answer": "1-4h", + }, + { + "step": "idp_extraction", + "question": QUESTION_NOTES, + "answer": "went fine", + }, + ] - def test_appends(self, root: Path, fpath_log: Path): - record(root, Step.GET_DATA, Status.START) - record(root, Step.GET_DATA, Status.END, notes="went fine") - df = _read(fpath_log) - assert list(df["status"]) == ["start", "end"] - assert list(df["notes"]) == ["", "went fine"] - assert fpath_log.read_text().count("timestamp") == 1 + def test_end_other(self, root: Path, fpath_timestamps: Path, fpath_survey: Path): + record(root, Step.FEDBIOMED_NODE_PART_1, Status.START) + assert record( + root, Step.FEDBIOMED_NODE_PART_1, Status.END, answers={QUESTION_NOTES: ""} + ) == (fpath_timestamps, fpath_survey) + assert _read(fpath_survey).to_dict("records") == [ + {"step": "fedbiomed_node_part_1", "question": QUESTION_NOTES, "answer": ""} + ] - def test_notes_with_special_characters(self, root: Path, fpath_log: Path): + def test_survey_appends_in_question_order(self, root: Path, fpath_survey: Path): + record(root, Step.FL_READY_DATA, Status.START) + record( + root, + Step.FL_READY_DATA, + Status.END, + answers=dict(reversed(DATA_PREP_ANSWERS.items())), + ) + record(root, Step.FEDBIOMED_NODE_PART_2, Status.START) + record( + root, Step.FEDBIOMED_NODE_PART_2, Status.END, answers={QUESTION_NOTES: "x"} + ) + df = _read(fpath_survey) + assert list(df["question"]) == [*DATA_PREP_ANSWERS, QUESTION_NOTES] + assert fpath_survey.read_text().count("question") == 1 + + def test_notes_with_special_characters(self, root: Path, fpath_survey: Path): notes = 'tab\there\nnew line and "quotes"' - record(root, Step.GET_DATA, Status.START) - record(root, Step.GET_DATA, Status.END, notes=notes) - assert _read(fpath_log).iloc[-1]["notes"] == notes + record(root, Step.FEDBIOMED_NODE_PART_1, Status.START) + record( + root, + Step.FEDBIOMED_NODE_PART_1, + Status.END, + answers={QUESTION_NOTES: notes}, + ) + assert _read(fpath_survey).iloc[-1]["answer"] == notes + + @pytest.mark.parametrize( + ("step", "status", "answers"), + [ + (Step.IDP_EXTRACTION, Status.START, {QUESTION_NOTES: ""}), + (Step.IDP_EXTRACTION, Status.END, None), + (Step.IDP_EXTRACTION, Status.END, {QUESTION_NOTES: ""}), + (Step.FEDBIOMED_NODE_PART_1, Status.END, DATA_PREP_ANSWERS), + (Step.FEDBIOMED_NODE_PART_1, Status.END, {"Unknown?": ""}), + ], + ) + def test_invalid_questions( + self, + root: Path, + fpath_timestamps: Path, + fpath_survey: Path, + step: Step, + status: Status, + answers: dict[str, str] | None, + ): + with pytest.raises(ValueError, match="exactly these questions"): + record(root, step, status, answers=answers) + assert not fpath_timestamps.exists() + assert not fpath_survey.exists() + + @pytest.mark.parametrize( + ("question", "answer"), + [ + (QUESTION_TOOLS_EASIER, "maybe"), + (QUESTION_TOOLS_EASIER, "4"), + (QUESTION_TIME_WITHOUT_TOOLS, "2 days"), + ], + ) + def test_invalid_answer( + self, root: Path, fpath_timestamps: Path, question: str, answer: str + ): + with pytest.raises(ValueError, match="must be one of"): + record( + root, + Step.SEMANTIC_HARMONIZATION, + Status.END, + answers={**DATA_PREP_ANSWERS, question: answer}, + ) + assert not fpath_timestamps.exists() def test_no_warning(self, root: Path): + # warnings are emitted by check_status with warnings.catch_warnings(): warnings.simplefilter("error") - record(root, Step.GET_DATA, Status.START) - record(root, Step.GET_DATA, Status.END) + record( + root, + Step.FEDBIOMED_NODE_PART_1, + Status.END, + answers={QUESTION_NOTES: ""}, + ) + record(root, Step.FEDBIOMED_NODE_PART_1, Status.START) + record(root, Step.FEDBIOMED_NODE_PART_1, Status.START) + + def test_invalid_timestamps_file(self, root: Path, fpath_timestamps: Path): + fpath_timestamps.write_text("timestamp\tstep\nt0\tidp_extraction\n") + with pytest.warns(UserWarning, match="invalid file"): + record(root, Step.IDP_EXTRACTION, Status.START) + assert len(list(fpath_timestamps.parent.glob("timestamps-*.tsv"))) == 1 + assert len(_read(fpath_timestamps)) == 1 + + +class TestCheckStatus: + def _record(self, root: Path, step: Step, status: Status) -> None: + answers = {QUESTION_NOTES: ""} if status == Status.END else None + record(root, step, status, answers=answers) @pytest.mark.parametrize( - ("statuses", "message"), + ("statuses", "status"), [ - ([Status.END], "was not started"), - ([Status.START, Status.START], "was already started"), - ([Status.START, Status.END, Status.END], "was not started"), + ([], Status.START), + ([Status.START], Status.END), + ([Status.START, Status.END], Status.START), + ], + ) + def test_no_warning(self, root: Path, statuses: list[Status], status: Status): + for previous in statuses: + self._record(root, Step.FEDBIOMED_NODE_PART_1, previous) + with warnings.catch_warnings(): + warnings.simplefilter("error") + check_status(root, Step.FEDBIOMED_NODE_PART_1, status) + + @pytest.mark.parametrize( + ("statuses", "status", "message"), + [ + ([], Status.END, "was not started"), + ([Status.START], Status.START, "was already started"), + ([Status.START, Status.END], Status.END, "was not started"), ], ) def test_warning( self, root: Path, - fpath_log: Path, + fpath_timestamps: Path, statuses: list[Status], + status: Status, message: str, ): - for status in statuses[:-1]: - record(root, Step.GET_DATA, status) + for previous in statuses: + self._record(root, Step.FEDBIOMED_NODE_PART_1, previous) + content = fpath_timestamps.read_text() if statuses else None with pytest.warns(UserWarning, match=message): - record(root, Step.GET_DATA, statuses[-1]) - assert len(_read(fpath_log)) == len(statuses) + check_status(root, Step.FEDBIOMED_NODE_PART_1, status) + # nothing is written + if content is None: + assert not fpath_timestamps.exists() + else: + assert fpath_timestamps.read_text() == content - def test_extra_columns_kept(self, root: Path, fpath_log: Path): - fpath_log.write_text( - "timestamp\tstep\tstatus\tnotes\textra\nt0\tget-data\tstart\t\tx\n" - ) - record(root, Step.GET_DATA, Status.END) - df = _read(fpath_log) - assert list(df["status"]) == ["start", "end"] - assert list(df["extra"]) == ["x", ""] + def test_other_step(self, root: Path): + self._record(root, Step.IDP_EXTRACTION, Status.START) + with pytest.warns(UserWarning, match="fl_ready_data was not started"): + check_status(root, Step.FL_READY_DATA, Status.END) - @pytest.mark.parametrize( - "content", - [ - "", - "timestamp\tstep\tstatus\nt0\tget-data\tstart\n", - 'timestamp\tstep\tstatus\tnotes\n"unclosed\n', - ], - ) - def test_invalid_log( - self, - root: Path, - fpath_log: Path, - content: str, - ): - fpath_log.write_text(content) - with pytest.warns(UserWarning, match="invalid status log") as record_warnings: - record(root, Step.GET_DATA, Status.START) - - fpaths_backup = list(fpath_log.parent.glob("status_log-*.tsv")) - assert len(fpaths_backup) == 1 - assert fpaths_backup[0].read_text() == content - df = _read(fpath_log) - assert tuple(df.columns) == STATUS_LOG_COLUMNS - assert len(df) == 1 + def test_invalid_timestamps_file(self, root: Path, fpath_timestamps: Path): + fpath_timestamps.write_text("timestamp\tstep\nt0\tidp_extraction\n") + with pytest.warns(UserWarning) as record_warnings: + check_status(root, Step.IDP_EXTRACTION, Status.START) # previous records are discarded, so no start/end warning + assert str(record_warnings[0].message).startswith("invalid file") assert len(record_warnings) == 1 diff --git a/tests/utils/test_likert.py b/tests/utils/test_likert.py new file mode 100644 index 0000000..eb005db --- /dev/null +++ b/tests/utils/test_likert.py @@ -0,0 +1,23 @@ +import pytest + +from brainflex.utils.likert import label_likert_answer + +LABELS = ("a", "b", "c", "d", "e") + + +class TestLabelLikertAnswer: + @pytest.mark.parametrize( + ("answer", "expected"), [("1", "1 (a)"), (3, "3 (c)"), ("5", "5 (e)")] + ) + def test_label(self, answer: str | int, expected: str): + assert label_likert_answer(answer, LABELS) == expected + + @pytest.mark.parametrize("answer", ["0", "6", 6, "", "a"]) + def test_invalid_answer(self, answer: str | int): + with pytest.raises(ValueError, match="must be one of"): + label_likert_answer(answer, LABELS) + + @pytest.mark.parametrize("labels", [LABELS[:4], (*LABELS, "f")]) + def test_wrong_number_of_labels(self, labels: tuple[str, ...]): + with pytest.raises(ValueError, match="Expected 5 labels"): + label_likert_answer("1", labels) diff --git a/tests/utils/test_tsv.py b/tests/utils/test_tsv.py new file mode 100644 index 0000000..af68751 --- /dev/null +++ b/tests/utils/test_tsv.py @@ -0,0 +1,68 @@ +from pathlib import Path + +import pandas as pd +import pytest + +from brainflex.utils.tsv import append_tsv, load_tsv + +COLUMNS = ("a", "b") + + +@pytest.fixture +def fpath(tmp_path: Path) -> Path: + return tmp_path / "file.tsv" + + +def _read(fpath: Path) -> pd.DataFrame: + return pd.read_csv(fpath, sep="\t", dtype=str, keep_default_na=False) + + +class TestLoadTsv: + def test_missing(self, fpath: Path): + df = load_tsv(fpath, COLUMNS) + assert tuple(df.columns) == COLUMNS + assert df.empty + assert not fpath.exists() + + def test_valid(self, fpath: Path): + fpath.write_text("a\tb\textra\n1\t2\tx\n") + df = load_tsv(fpath, COLUMNS) + assert df.to_dict("records") == [{"a": "1", "b": "2", "extra": "x"}] + + @pytest.mark.parametrize("content", ["", "a\n1\n", 'a\tb\n"unclosed\n']) + def test_invalid(self, fpath: Path, content: str): + fpath.write_text(content) + with pytest.warns(UserWarning, match="invalid file"): + df = load_tsv(fpath, COLUMNS) + assert tuple(df.columns) == COLUMNS + assert df.empty + assert not fpath.exists() + fpaths_backup = list(fpath.parent.glob("file-*.tsv")) + assert len(fpaths_backup) == 1 + assert fpaths_backup[0].read_text() == content + + +class TestAppendTsv: + def test_creates_file(self, fpath: Path): + append_tsv(fpath, COLUMNS, [{"a": "1", "b": "2"}]) + assert fpath.read_text() == "a\tb\n1\t2\n" + + def test_appends(self, fpath: Path): + append_tsv(fpath, COLUMNS, [{"a": "1", "b": "2"}]) + append_tsv(fpath, COLUMNS, ({"a": str(i), "b": ""} for i in range(2))) + assert list(_read(fpath)["a"]) == ["1", "0", "1"] + + def test_extra_columns_kept(self, fpath: Path): + fpath.write_text("a\tb\textra\n1\t2\tx\n") + append_tsv(fpath, COLUMNS, [{"a": "3", "b": "4"}]) + assert list(_read(fpath)["extra"]) == ["x", ""] + + def test_special_characters(self, fpath: Path): + value = 'tab\there\nnew line and "quotes"' + append_tsv(fpath, COLUMNS, [{"a": value, "b": ""}]) + assert _read(fpath).iloc[0]["a"] == value + + def test_given_df(self, fpath: Path): + df = pd.DataFrame([{"a": "0", "b": "0"}], columns=COLUMNS) + append_tsv(fpath, COLUMNS, [{"a": "1", "b": "2"}], df=df) + assert list(_read(fpath)["a"]) == ["0", "1"]