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
63 changes: 59 additions & 4 deletions src/nomadic/realtime/commands.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import difflib
import glob
import os
from pathlib import Path
from shutil import rmtree
Expand Down Expand Up @@ -246,6 +248,8 @@ def find_minknow_fastq_dirs(
return minknow.resolve_minknow_fastq_dirs(minknow_dir, experiment_name)
except minknow.MinknowPathError as e:
raise BadParameterWithSource(message=str(e), param_hint="-k/--minknow_dir")
except UserInputError as e:
raise click.UsageError(f"Could not find minknow folders: {e}") from e
else:
# If fastq_dir is manually given, we assume there is no minknow dir
return None, fastq_dir
Expand Down Expand Up @@ -297,9 +301,60 @@ def find_metadata_file(experiment_name: str, workspace: Workspace) -> str:
break

if metadata_path is None or not os.path.isfile(metadata_path):
msg = f"Metadata file not found. Did you create your metadata file in `{workspace.get_metadata_dir()}`"
if shared_workspace:
msg += f", or in `{shared_workspace}`"
msg += f", and does the name match `{experiment_name}`?"
msg = "Metadata file not found. "
closest_match = find_closest_metadata_file(
experiment_name,
[workspace.get_metadata_dir()]
+ ([shared_workspace.get_metadata_dir()] if shared_workspace else []),
)
if closest_match is not None:
closest_experiment_name, closest_metadata_path = closest_match
msg += f"Did you mean `{closest_experiment_name}`? The closest metadata file is `{closest_metadata_path}`. Either rename your metadata file to match the experiment name or correct the experiment name in your command."
else:
msg += (
f"Did you create your metadata file in `{workspace.get_metadata_dir()}`"
)
if shared_workspace:
msg += f", or in `{shared_workspace}`"
msg += f", and does the name match `{experiment_name}`?"
raise click.BadParameter(message=msg)
return metadata_path


def find_closest_metadata_file(
experiment_name: str, metadata_folders: list[str]
) -> tuple[str, str] | None:
"""
Find the closest metadata file to the given experiment name within the specified metadata folders.

Args:
experiment_name (str): The name of the experiment to find a metadata file for.
metadata_folders (list[str]): A list of folders to search for metadata files.

Returns:
tuple[str, str] | None: A tuple containing the closest matching experiment name and the path to the metadata file,
or None if no close match is found.
"""
files = []
for folder in metadata_folders:
files.extend(
file
for file in glob.glob(f"{folder}/*")
if os.path.isfile(file)
and os.path.splitext(file)[1].lower() in {".csv", ".xlsx"}
)

close_matches = difflib.get_close_matches(
experiment_name,
[os.path.splitext(os.path.basename(f))[0] for f in files],
cutoff=0.8,
)
if not close_matches:
return None
closest_match = close_matches[0]

for file in files:
if os.path.splitext(os.path.basename(file))[0] == closest_match:
return closest_match, file

return None
35 changes: 34 additions & 1 deletion src/nomadic/util/minknow.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,14 @@
import difflib
import glob
import platform
import warnings
from pathlib import Path
from typing import Optional

from nomadic.util.exceptions import UserInputError

class MinknowPathError(Exception):

class MinknowPathError(UserInputError):
pass


Expand Down Expand Up @@ -74,6 +77,18 @@ def resolve_minknow_fastq_dirs(

if is_minknow_base_dir(minknow_path):
minknow_dir = minknow_path / experiment_name
if not minknow_dir.is_dir():
closest_match = get_most_similar_experiment_name(
experiment_name, minknow_path
)
if closest_match:
raise UserInputError(
f"'{experiment_name}' does not match any existing minknow experiments in '{minknow_path}'. Did you mean '{closest_match}'?",
)
else:
raise UserInputError(
f"'{experiment_name}' does not match any existing minknow experiments in '{minknow_path}', please ensure the experiment name is correct and matches the minknow experiment name.",
)
elif is_minknow_experiment_dir(minknow_path):
minknow_dir = minknow_path
else:
Expand Down Expand Up @@ -114,3 +129,21 @@ def default_data_dir() -> Path:
return integrated_devices_path

return standard_path


def get_most_similar_experiment_name(
experiment_name: str, minknow_path: Path
) -> str | None:
close_matches = difflib.get_close_matches(
experiment_name.lower(),
[
d.name
for d in minknow_path.glob("*")
if d.is_dir()
and d.name not in {"persistence", "reads", "queued_reads", "intermediates"}
],
cutoff=0.8,
)
if close_matches:
return close_matches[0]
return None
Loading