From da46dc1660d0e6a7630e82f644803710227fd4cb Mon Sep 17 00:00:00 2001 From: AditiAdhikari05 Date: Thu, 9 Jul 2026 15:38:44 +0545 Subject: [PATCH 1/2] add parameters --- src/ariadnepy/core/_versions.py | 3 + src/ariadnepy/graph/_weave.py | 153 +++++++++++++++++++--- src/ariadnepy/io/_batching.py | 56 ++++++++ src/ariadnepy/io/_ott.py | 26 ++-- src/ariadnepy/io/_sparql.py | 18 ++- src/ariadnepy/plot/_draw.py | 112 ++++++++++------ src/ariadnepy/resources/__init__.py | 4 +- src/ariadnepy/resources/_data.py | 28 ++++ src/ariadnepy/resources/_parsers.py | 18 ++- src/ariadnepy/resources/data/pathmeta.csv | 5 + tests/test_core/test_graph/test_plot.py | 84 ++++++++++++ tests/test_core/test_graph/test_weave.py | 88 +++++++++++++ 12 files changed, 516 insertions(+), 79 deletions(-) create mode 100644 src/ariadnepy/io/_batching.py create mode 100644 src/ariadnepy/resources/data/pathmeta.csv diff --git a/src/ariadnepy/core/_versions.py b/src/ariadnepy/core/_versions.py index acbd37d..8034881 100644 --- a/src/ariadnepy/core/_versions.py +++ b/src/ariadnepy/core/_versions.py @@ -171,6 +171,9 @@ def list_resource_versions(default: bool = False) -> pd.DataFrame: >>> list_resource_versions() >>> list_resource_versions(default=True) """ + if not isinstance(default, bool): + raise AriadneVersionError("'default' must be True or False.") + metadata = load_version_metadata() if default: metadata = metadata[metadata["default"]].copy() diff --git a/src/ariadnepy/graph/_weave.py b/src/ariadnepy/graph/_weave.py index 536b080..6a3fa04 100644 --- a/src/ariadnepy/graph/_weave.py +++ b/src/ariadnepy/graph/_weave.py @@ -63,6 +63,42 @@ def _get_edge_key(from_: str, to: str) -> str: return f"{from_}--{to}" +def _get_sorted_edge_key(from_: str, to: str, source: str) -> str: + """Direction-agnostic edge key, matching R's ``.get_edge_keys``.""" + return "_".join([*sorted([str(from_), str(to)]), str(source)]) + + +def _graph_from_path_df(path_df: pd.DataFrame) -> ig.Graph: + """Rebuild the minimal igraph subgraph covering exactly the steps in path_df. + + Mirrors R's ``.graph_from_path_df``, used by the ``weavePath``/ + ``weaveComplex`` data.frame methods (e.g. for the bundled ``pathMeta`` + example, or output from ``draw_path()``). + """ + if "version" in path_df.columns: + res_df = path_df[["source", "version"]].drop_duplicates().dropna() + versions: dict[str, str] | None = dict( + zip(res_df["source"], res_df["version"], strict=False) + ) + else: + versions = None + + from ariadnepy.core._graph import ariadne + graph = ariadne(versions=versions) + + _, edge_df = _get_graph_dataframes(graph) + graph_keys = [ + _get_sorted_edge_key(r["from"], r["to"], r.get("source", "")) + for _, r in edge_df.iterrows() + ] + path_keys = { + _get_sorted_edge_key(r["from"], r["to"], r.get("source", "")) + for _, r in path_df.iterrows() + } + keep_idx = [i for i, key in enumerate(graph_keys) if key in path_keys] + return graph.subgraph_edges(keep_idx, delete_vertices=True) + + def _generic2specific( path_df: pd.DataFrame, node_df: pd.DataFrame, col: str ) -> list[str]: @@ -326,7 +362,10 @@ def _fetch_kegg_edge( def _fetch_sparql_edge( - step: pd.Series, init: list[str] | None, timeout: float + step: pd.Series, + init: list[str] | None, + timeout: float, + batch_kwargs: dict | None = None, ) -> pd.DataFrame: """Fetch one SPARQL edge, delegating to io._sparql when available.""" try: @@ -337,6 +376,7 @@ def _fetch_sparql_edge( endpoint=step["source"], init=init, timeout=timeout, + **(batch_kwargs or {}), ) except (ImportError, AttributeError): pass @@ -383,7 +423,10 @@ def _fetch_sparql_edge( def _fetch_ott_edge( - step: pd.Series, init: list[str] | None, timeout: float + step: pd.Series, + init: list[str] | None, + timeout: float, + batch_kwargs: dict | None = None, ) -> pd.DataFrame: """Fetch OTT taxonomy mappings, delegating to io._ott when available.""" try: @@ -393,6 +436,7 @@ def _fetch_ott_edge( to=step["specTo"], init=init, timeout=timeout, + **(batch_kwargs or {}), ) except (ImportError, AttributeError): pass @@ -478,7 +522,10 @@ def _fetch_file_edge( # ── Edge dispatcher ─────────────────────────────────────────────────────────── def _fetch_edge( - step: pd.Series, init: list[str] | None, timeout: float + step: pd.Series, + init: list[str] | None, + timeout: float, + batch_kwargs: dict | None = None, ) -> pd.DataFrame: """Route one path step to the correct backend and return a linkmap.""" source = step["source"] @@ -492,10 +539,10 @@ def _fetch_edge( elif source == "OTT": if not is_init: raise AriadneError("'init' must be provided for OTT queries.") - df = _fetch_ott_edge(step, init, timeout) + df = _fetch_ott_edge(step, init, timeout, batch_kwargs) elif source in ("Rhea", "UniProt"): - df = _fetch_sparql_edge(step, init, timeout) + df = _fetch_sparql_edge(step, init, timeout, batch_kwargs) if step.get("specTo") == "BioCyc": key = re.sub(r"_.+$", "", step["initTo"]) df = df[ @@ -773,6 +820,7 @@ def _build_path_mf( prune_last: bool, verbose: bool, timeout: float, + batch_kwargs: dict | None = None, ) -> dict[str, pd.DataFrame]: """Build the ordered chain of linkmaps for the path from_ → to.""" if not isinstance(timeout, (int, float)) or timeout <= 0: @@ -807,7 +855,7 @@ def _build_path_mf( for i, (_, step) in enumerate(path_df.iterrows()): if verbose: print(f" {step['initFrom']} -({step['source']})-> {step['initTo']}") - linkmap = _fetch_edge(step, init_list, timeout) + linkmap = _fetch_edge(step, init_list, timeout, batch_kwargs) key = f"{step['from']}2{step['to']}" linkmaps[key] = linkmap if prune_vec[i]: @@ -821,8 +869,8 @@ def _build_path_mf( # ── Public API ──────────────────────────────────────────────────────────────── def weave_path( - graph: ig.Graph, - by: str, + graph: ig.Graph | pd.DataFrame, + by: str | None = None, k: int = 1, include: list[str] | None = None, exclude: list[str] | None = None, @@ -832,25 +880,36 @@ def weave_path( use_names: bool = True, verbose: bool = True, timeout: float = 1e6, + batch_size: int | None = None, + workers: int | None = None, + factor: int = 3, ) -> pd.DataFrame: """Build a linkmap by traversing the resource graph from origin to target. Equivalent to R's ``weavePath(graph, taxname ~ bugsig, init = tax_labs)``. + Accepts either an igraph resource graph (from ``ariadne()``) or a fixed + path DataFrame (from ``draw_path()``, or the bundled ``pathMeta`` example). + When ``graph`` is a DataFrame, the path is already determined by its rows, + so ``by``, ``k``, ``include``, ``exclude``, and ``res_name`` don't apply — + same restriction as R's ``data.frame`` method for ``weavePath``. + Parameters ---------- graph: - NetworkX MultiDiGraph returned by ``ariadne()``. + igraph resource graph returned by ``ariadne()``, or a path DataFrame + (columns ``from``, ``to``, ``source``, optionally ``version``). by: Path formula string, e.g. ``"taxname ~ bugsig"`` or ``"ko ~ ec"``. + Required when ``graph`` is an igraph; not accepted for a DataFrame. k: - Use the k-th shortest path (1 = shortest). + Use the k-th shortest path (1 = shortest). igraph input only. include: - Node names that the chosen path must pass through. + Node names that the chosen path must pass through. igraph input only. exclude: - Node names the chosen path must avoid. + Node names the chosen path must avoid. igraph input only. res_name: - Restrict graph edges to these resource names only. + Restrict graph edges to these resource names only. igraph input only. init: Seed IDs for the first step (list), or a 2-column DataFrame for stratified input (first column = strata, second = IDs). @@ -862,6 +921,12 @@ def weave_path( Print step-by-step progress messages. timeout: HTTP request timeout in seconds. + batch_size: + Maximum IDs per SPARQL/OTT request. Backend-specific default when None. + workers: + Parallel workers for batched requests. Auto-detected when None. + factor: + Number of jobs per worker, used to cap the batch count. Returns ------- @@ -876,12 +941,33 @@ def weave_path( >>> tax2bugsig_via_taxid = weave_path( ... graph, "taxname ~ bugsig", include=["taxid"], init=tax_labs ... ) + >>> chebi2gmm = weave_path(pathmeta_df, init=[15377, 30616, 4167]) """ + if isinstance(graph, pd.DataFrame): + if ( + by is not None or k != 1 or include is not None + or exclude is not None or res_name is not None + ): + raise AriadneError( + "'by', 'k', 'include', 'exclude', and 'res_name' are not " + "supported when 'graph' is a DataFrame — the path is already " + "fixed by its rows." + ) + by = f"{graph['from'].iloc[0]} ~ {graph['to'].iloc[-1]}" + graph = _graph_from_path_df(graph) + elif by is None: + raise AriadneError("'by' must be provided when 'graph' is an igraph object.") + from_, to = _parse_by(by) + batch_kwargs = { + key: val for key, val in {"batch_size": batch_size, "workers": workers}.items() + if val is not None + } + batch_kwargs["factor"] = factor linkmaps = _build_path_mf( graph, from_, to, k, include, exclude, res_name, - init, prune, prune, verbose, timeout, + init, prune, prune, verbose, timeout, batch_kwargs, ) result = _weave_linkmaps(linkmaps, from_, to) @@ -900,8 +986,8 @@ def weave_path( def weave_complex( - graph: ig.Graph, - by: str, + graph: ig.Graph | pd.DataFrame, + by: str | None = None, k: int = 1, include: list[str] | None = None, exclude: list[str] | None = None, @@ -912,6 +998,9 @@ def weave_complex( threshold: float | None = None, verbose: bool = True, timeout: float = 1e6, + batch_size: int | None = None, + workers: int | None = None, + factor: int = 3, ) -> pd.DataFrame: """Like ``weave_path`` but returns module coverage scores for complex modules. @@ -921,10 +1010,20 @@ def weave_complex( Equivalent to R's ``weaveComplex(graph, kegg_disease ~ gmm, threshold=0.8)``. + Accepts either an igraph resource graph or a fixed path DataFrame, same + as ``weave_path`` — see its docstring for the DataFrame-input restrictions + on ``by``/``k``/``include``/``exclude``/``res_name``. + Parameters ---------- threshold: Only return rows where ``cov >= threshold``. Must be in (0, 1]. + batch_size: + Maximum IDs per SPARQL/OTT request. Backend-specific default when None. + workers: + Parallel workers for batched requests. Auto-detected when None. + factor: + Number of jobs per worker, used to cap the batch count. Returns ------- @@ -940,7 +1039,27 @@ def weave_complex( if threshold is not None and not (0 < threshold <= 1): raise AriadneError("'threshold' must be a number between 0 and 1.") + if isinstance(graph, pd.DataFrame): + if ( + by is not None or k != 1 or include is not None + or exclude is not None or res_name is not None + ): + raise AriadneError( + "'by', 'k', 'include', 'exclude', and 'res_name' are not " + "supported when 'graph' is a DataFrame — the path is already " + "fixed by its rows." + ) + by = f"{graph['from'].iloc[0]} ~ {graph['to'].iloc[-1]}" + graph = _graph_from_path_df(graph) + elif by is None: + raise AriadneError("'by' must be provided when 'graph' is an igraph object.") + from_, to = _parse_by(by) + batch_kwargs = { + key: val for key, val in {"batch_size": batch_size, "workers": workers}.items() + if val is not None + } + batch_kwargs["factor"] = factor complex_modules = {"gbm", "gmm"} _, edge_df = _get_graph_dataframes(graph) @@ -962,7 +1081,7 @@ def weave_complex( inner_from, inner_to = _parse_by(inner_by) linkmaps = _build_path_mf( graph, inner_from, inner_to, k, include, exclude, res_name, - init, prune, True, verbose, timeout, + init, prune, True, verbose, timeout, batch_kwargs, ) if to in complex_modules: diff --git a/src/ariadnepy/io/_batching.py b/src/ariadnepy/io/_batching.py new file mode 100644 index 0000000..661d852 --- /dev/null +++ b/src/ariadnepy/io/_batching.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import math +import os + +from ariadnepy.exceptions import AriadneError + + +def get_batches( + x_len: int | None, + batch_size: int, + workers: int | None, + factor: int = 3, +) -> list[tuple[int, int]]: + """Split a sequence of length ``x_len`` into (start, end) index ranges. + + Mirrors R ariadne's internal ``.get_batches``: the number of batches is + capped at ``factor * workers`` so that batching never spawns more jobs + than the worker pool can use, then raises if the resulting per-batch size + still exceeds ``batch_size``. + + Parameters + ---------- + x_len: + Length of the sequence to batch, or None for a single empty batch. + batch_size: + Maximum number of items per batch. + workers: + Number of parallel workers. Auto-detected from CPU count when None. + factor: + Number of jobs per worker. + + Returns + ------- + list of (start, end) + 0-based, end-exclusive index ranges suitable for slicing. + """ + if x_len is None: + return [(0, 0)] + + if workers is None: + workers = os.cpu_count() or 1 + + batch_num = max(min(math.ceil(x_len / batch_size), factor * workers), 1) + adapted_size = math.ceil(x_len / batch_num) + + if adapted_size > batch_size: + raise AriadneError( + f"Query limit was reached ({adapted_size} > {batch_size}). " + "Increase 'factor', 'batch_size' or 'workers' and try again." + ) + + return [ + (i * adapted_size, min((i + 1) * adapted_size, x_len)) + for i in range(batch_num) + ] diff --git a/src/ariadnepy/io/_ott.py b/src/ariadnepy/io/_ott.py index a52d058..7ecd65a 100644 --- a/src/ariadnepy/io/_ott.py +++ b/src/ariadnepy/io/_ott.py @@ -1,11 +1,13 @@ from __future__ import annotations +import os import re from concurrent.futures import ThreadPoolExecutor import pandas as pd from ariadnepy.exceptions import AriadneDownloadError, AriadneError +from ariadnepy.io._batching import get_batches try: import requests as _requests @@ -27,13 +29,16 @@ def _query_tnrs( to: str, timeout: float, batch_size: int = 1000, - workers: int = 4, + workers: int | None = None, + factor: int = 3, ) -> list[str | None]: """Match taxonomy names via TNRS API, return OTT id (or other target) per name.""" if _requests is None: raise AriadneDownloadError("'requests' is required for OTT queries.") - batches = [names[i : i + batch_size] for i in range(0, len(names), batch_size)] + ranges = get_batches(len(names), batch_size, workers, factor) + batches = [names[start:end] for start, end in ranges] + resolved_workers = workers if workers is not None else (os.cpu_count() or 1) def _run(batch: list[str]) -> list[str | None]: resp = _requests.post( @@ -62,7 +67,7 @@ def _run(batch: list[str]) -> list[str | None]: out.append(val) return out - with ThreadPoolExecutor(max_workers=min(workers, len(batches))) as pool: + with ThreadPoolExecutor(max_workers=min(resolved_workers, len(batches))) as pool: nested = list(pool.map(_run, batches)) return [item for batch in nested for item in batch] @@ -106,7 +111,8 @@ def query_ott( init: list[str] | None, timeout: float = 1e6, batch_size: int = 1000, - workers: int = 4, + workers: int | None = None, + factor: int = 3, ) -> pd.DataFrame: """Fetch taxonomy mappings from the Open Tree of Life API. @@ -126,7 +132,10 @@ def query_ott( batch_size: Number of names per TNRS batch (for taxname queries). workers: - Parallel workers for batched requests. + Parallel workers for batched requests. Auto-detected from CPU count + when None. + factor: + Number of jobs per worker, used to cap the batch count. Returns ------- @@ -142,9 +151,10 @@ def query_ott( raise AriadneError("'init' must be provided for OTT queries.") clean = _strip_rank_prefix(init) + resolved_workers = workers if workers is not None else (os.cpu_count() or 1) if from_ == "taxname": - values = _query_tnrs(clean, to, timeout, batch_size, workers) + values = _query_tnrs(clean, to, timeout, batch_size, workers, factor) rows = [ (orig, val) for orig, val in zip(init, values, strict=False) @@ -156,7 +166,7 @@ def query_ott( def _fetch(raw_id: str) -> str | None: return _query_taxon_info(raw_id, to, "ott_id", timeout) - with ThreadPoolExecutor(max_workers=min(workers, len(clean))) as pool: + with ThreadPoolExecutor(max_workers=min(resolved_workers, len(clean))) as pool: values = list(pool.map(_fetch, clean)) rows = [(orig, val) for orig, val in zip(init, values, strict=False) if val is not None] df = pd.DataFrame(rows, columns=[from_, to]) @@ -165,7 +175,7 @@ def _fetch(raw_id: str) -> str | None: def _fetch_ext(raw_id: str) -> str | None: return _query_taxon_info(f"{from_}:{raw_id}", to, "source_id", timeout) - with ThreadPoolExecutor(max_workers=min(workers, len(clean))) as pool: + with ThreadPoolExecutor(max_workers=min(resolved_workers, len(clean))) as pool: values = list(pool.map(_fetch_ext, clean)) rows = [(orig, val) for orig, val in zip(init, values, strict=False) if val is not None] df = pd.DataFrame(rows, columns=[from_, to]) diff --git a/src/ariadnepy/io/_sparql.py b/src/ariadnepy/io/_sparql.py index c10a563..be66916 100644 --- a/src/ariadnepy/io/_sparql.py +++ b/src/ariadnepy/io/_sparql.py @@ -1,10 +1,12 @@ from __future__ import annotations +import os from concurrent.futures import ThreadPoolExecutor import pandas as pd from ariadnepy.exceptions import AriadneDownloadError, AriadneError +from ariadnepy.io._batching import get_batches try: import requests as _requests @@ -173,8 +175,9 @@ def query_sparql( endpoint: str, init: list[str] | None = None, timeout: float = 1e6, - batch_size: int = 500, - workers: int = 4, + batch_size: int = 25000, + workers: int | None = None, + factor: int = 3, ) -> pd.DataFrame: """Run a SPARQL query against UniProt or Rhea and return a linkmap DataFrame. @@ -193,7 +196,10 @@ def query_sparql( batch_size: Maximum number of IDs per SPARQL request. workers: - Number of parallel workers for batched requests. + Number of parallel workers for batched requests. Auto-detected from + CPU count when None. + factor: + Number of jobs per worker, used to cap the batch count. Returns ------- @@ -220,13 +226,15 @@ def query_sparql( return df # Split init into batches - batches = [init[i : i + batch_size] for i in range(0, len(init), batch_size)] + ranges = get_batches(len(init), batch_size, workers, factor) + batches = [init[start:end] for start, end in ranges] + resolved_workers = workers if workers is not None else (os.cpu_count() or 1) def _run_batch(batch: list[str]) -> pd.DataFrame: query = _build_query(from_, to, clause, batch, uniref_identity) return _send_sparql(query, endpoint, timeout) - with ThreadPoolExecutor(max_workers=min(workers, len(batches))) as pool: + with ThreadPoolExecutor(max_workers=min(resolved_workers, len(batches))) as pool: results = list(pool.map(_run_batch, batches)) df = pd.concat(results, ignore_index=True) diff --git a/src/ariadnepy/plot/_draw.py b/src/ariadnepy/plot/_draw.py index 44c899c..7ff761b 100644 --- a/src/ariadnepy/plot/_draw.py +++ b/src/ariadnepy/plot/_draw.py @@ -4,11 +4,12 @@ import matplotlib.pyplot as plt from matplotlib.figure import Figure -from ariadnepy.graph._weave import _draw_path, _parse_by +from ariadnepy.graph._weave import _draw_path, _get_sorted_edge_key, _parse_by _C_NODE = "#FF8C00" # darkorange — matches R ariadne _C_EDGE_PATH = "#E74C3C" # red _C_EDGE_BG = "#CCCCCC" # grey80 +_FADE_ALPHA = 0.15 # opacity for edges/nodes faded out by prune/res_name def plot_path( @@ -39,13 +40,18 @@ def plot_path( exclude : list of str, optional Nodes the path must avoid. res_name : list of str, optional - Restrict path edges to these resource names. + Resource names to highlight. With ``by`` set, restricts path-finding + to these resources; with ``by=None``, fades every edge/node not + belonging to these resources across the whole graph (R's + ``plotPath(graph, res_name=["KEGG", "WoL"])`` usage). prune : bool - If True, show only path nodes and edges. - Equivalent to R's ``prune=TRUE``. + If True, fade every edge/node not on the chosen path (requires + ``by``). Equivalent to R's ``prune=TRUE``. focus : bool - If True, compute layout on path subgraph only (removes non-path nodes - from the canvas entirely). Equivalent to R's ``focus=TRUE``. + If True, drop every faded (non-highlighted) edge/node from the plot + entirely, instead of just fading them. Equivalent to R's + ``focus=TRUE``. Has no effect unless ``prune`` or ``res_name`` is + also active, since otherwise nothing is faded to begin with. figsize : tuple Matplotlib figure size ``(width, height)`` in inches. @@ -55,73 +61,93 @@ def plot_path( Examples -------- - >>> fig = plot_path(graph, "ec ~ ko", k=3) # full graph, path highlighted - >>> fig = plot_path(graph, "ec ~ ko", k=3, prune=True) # only path shown - >>> fig = plot_path(graph) # full resource graph + >>> fig = plot_path(graph, "ec ~ ko", k=3) # full graph, path highlighted + >>> fig = plot_path(graph, "ec ~ ko", k=3, prune=True) # non-path faded + >>> fig = plot_path(graph, "ec ~ ko", k=3, focus=True) # only path shown + >>> fig = plot_path(graph, res_name=["KEGG", "WoL"]) # resource highlighted + >>> fig = plot_path(graph) # full resource graph >>> fig.savefig("path.png") """ + if not isinstance(prune, bool): + raise ValueError("'prune' must be True or False.") + if not isinstance(focus, bool): + raise ValueError("'focus' must be True or False.") if prune and by is None: raise ValueError("'prune' requires 'by' to be specified.") - path_nodes: list[str] = [] - path_edges: list[tuple] = [] - path_edge_labels: dict = {} + has_source_attr = "source" in graph.edge_attributes() + edges_info = [] + for e in graph.es: + u = graph.vs[e.source]["name"] + v = graph.vs[e.target]["name"] + src = e["source"] if has_source_attr else "" + edges_info.append({"u": u, "v": v, "source": src}) + path_edge_keys: set[str] = set() if by is not None: from_, to = _parse_by(by) path_df = _draw_path(graph, from_, to, k, include, exclude, res_name) - path_nodes = [path_df.iloc[0]["from"]] + list(path_df["to"]) for _, row in path_df.iterrows(): - path_edges.append((row["from"], row["to"])) - path_edge_labels[(row["from"], row["to"])] = row.get("source", "") - - # Use frozensets so edge lookup is direction-agnostic, matching R's sort(c(from,to)) approach. - path_edge_set = {frozenset(e) for e in path_edges} - - # prune/focus both restrict layout to the path subgraph. - if path_nodes and (focus or prune): - path_idx = [ - graph.vs.find(name=n).index - for n in path_nodes - if n in graph.vs["name"] - ] - draw_graph = graph.induced_subgraph(path_idx) + path_edge_keys.add( + _get_sorted_edge_key(row["from"], row["to"], row.get("source", "")) + ) + + for info in edges_info: + info["mark"] = _get_sorted_edge_key(info["u"], info["v"], info["source"]) in path_edge_keys + + # alpha marks what's highlighted vs faded; mark (above) marks what's coloured red. + # These are independent axes, same as R's plotPath. + if prune: + for info in edges_info: + info["alpha"] = info["mark"] + elif res_name is not None: + for info in edges_info: + info["alpha"] = info["source"] in res_name + else: + for info in edges_info: + info["alpha"] = True + + connected_alpha_nodes = { + n for info in edges_info if info["alpha"] for n in (info["u"], info["v"]) + } + node_alpha = { + name: (name in connected_alpha_nodes) if (prune or res_name is not None) else True + for name in graph.vs["name"] + } + + if focus: + keep_edge_idx = [e.index for e, info in zip(graph.es, edges_info, strict=False) if info["alpha"]] + draw_graph = graph.subgraph_edges(keep_edge_idx, delete_vertices=True) + draw_edges_info = [info for info in edges_info if info["alpha"]] else: draw_graph = graph + draw_edges_info = edges_info # Kamada-Kawai layout — closest available equivalent to R ariadne's "stress" layout layout = draw_graph.layout("kk") all_names = draw_graph.vs["name"] pos = {name: tuple(layout[i]) for i, name in enumerate(all_names)} - all_edges_raw = [ - ( - draw_graph.vs[e.source]["name"], - draw_graph.vs[e.target]["name"], - e["source"] if "source" in draw_graph.edge_attributes() else "", - ) - for e in draw_graph.es - ] - # plt.close(fig) before returning removes it from pyplot's display queue, # preventing Jupyter from rendering it twice while still showing it as the # cell's return value via Jupyter's own repr mechanism. fig, ax = plt.subplots(figsize=figsize) # ── Edges ───────────────────────────────────────────────────────────────── - for u, v, src in all_edges_raw: + for info in draw_edges_info: + u, v, src = info["u"], info["v"], info["source"] if u not in pos or v not in pos: continue - is_path_edge = frozenset((u, v)) in path_edge_set - color = _C_EDGE_PATH if is_path_edge else _C_EDGE_BG - lw = 2.0 if is_path_edge else 1.0 + color = _C_EDGE_PATH if info["mark"] else _C_EDGE_BG + lw = 2.0 if info["mark"] else 1.0 + line_alpha = 1.0 if info["alpha"] else _FADE_ALPHA x0, y0 = pos[u] x1, y1 = pos[v] ax.plot( [x0, x1], [y0, y1], - color=color, lw=lw, zorder=1, solid_capstyle="round", + color=color, lw=lw, alpha=line_alpha, zorder=1, solid_capstyle="round", ) - if is_path_edge and src: + if info["mark"] and src: ax.text( (x0 + x1) / 2, (y0 + y1) / 2, src, fontsize=8, ha="center", va="center", @@ -133,7 +159,7 @@ def plot_path( ) # ── Nodes ───────────────────────────────────────────────────────────────── - draw_names = list(all_names) + draw_names = [n for n in all_names if node_alpha.get(n, True)] if draw_names: ax.scatter( [pos[n][0] for n in draw_names], diff --git a/src/ariadnepy/resources/__init__.py b/src/ariadnepy/resources/__init__.py index 5e7e7d3..dacba2f 100644 --- a/src/ariadnepy/resources/__init__.py +++ b/src/ariadnepy/resources/__init__.py @@ -1,4 +1,4 @@ from ariadnepy.resources._cache import cache_resource, init_cache -from ariadnepy.resources._data import load_butyrate +from ariadnepy.resources._data import load_butyrate, load_pathmeta -__all__ = ["cache_resource", "init_cache", "load_butyrate"] +__all__ = ["cache_resource", "init_cache", "load_butyrate", "load_pathmeta"] diff --git a/src/ariadnepy/resources/_data.py b/src/ariadnepy/resources/_data.py index 6c167ed..3e4bc01 100644 --- a/src/ariadnepy/resources/_data.py +++ b/src/ariadnepy/resources/_data.py @@ -28,3 +28,31 @@ def load_butyrate() -> pd.DataFrame: "Ensure the package was installed correctly." ) return pd.read_csv(data_file) + + +def load_pathmeta() -> pd.DataFrame: + """Load the bundled example pathway from chebi to gmm. + + A minimal path DataFrame describing one route through the ariadne graph + (chebi -> rhea -> ec -> ko -> gmm), the kind of table normally produced by + ``draw_path()`` and consumed by ``weave_path()``/``weave_complex()``. + + Returns + ------- + pd.DataFrame + Path steps with columns: from, to, source, version, url. + + Examples + -------- + >>> from ariadnepy.resources import load_pathmeta + >>> pathmeta = load_pathmeta() + >>> from ariadnepy.graph import weave_path + >>> chebi2gmm = weave_path(pathmeta, init=[15377, 30616, 4167]) + """ + data_file = Path(__file__).resolve().parent / "data" / "pathmeta.csv" + if not data_file.exists(): + raise FileNotFoundError( + f"Bundled dataset not found at {data_file}. " + "Ensure the package was installed correctly." + ) + return pd.read_csv(data_file) diff --git a/src/ariadnepy/resources/_parsers.py b/src/ariadnepy/resources/_parsers.py index 2966d34..64da9da 100644 --- a/src/ariadnepy/resources/_parsers.py +++ b/src/ariadnepy/resources/_parsers.py @@ -36,14 +36,24 @@ def process_one2many( path: str | Path, from_col: str, to_col: str, + key_col: int = 0, key_fn: Callable[[str], str] | None = None, val_fn: Callable[[str], str] | None = None, skiprows: int = 0, val_cols: slice | list[int] | None = None, ) -> pd.DataFrame: - """Parse a TSV where the first column maps to multiple values in the rest. + """Parse a TSV where one column maps to multiple values in the rest. Used for: ChocoPhlAn, WoL, BugSigDB. + + Parameters + ---------- + key_col: + Index of the column holding the "one" side of the mapping. + (R equivalent: ``key.col``.) + val_cols: + Indices/slice of columns holding the "many" side. Defaults to every + column except ``key_col``. (R equivalent: ``val.cols``.) """ path = Path(path) rows = [] @@ -57,11 +67,11 @@ def process_one2many( if not line: continue parts = line.split("\t") - if len(parts) < 2: + if len(parts) < 2 or key_col >= len(parts): continue - key = parts[0] + key = parts[key_col] if val_cols is None: - selection = parts[1:] + selection = [p for i, p in enumerate(parts) if i != key_col] elif isinstance(val_cols, slice): selection = parts[val_cols] else: diff --git a/src/ariadnepy/resources/data/pathmeta.csv b/src/ariadnepy/resources/data/pathmeta.csv new file mode 100644 index 0000000..b67a606 --- /dev/null +++ b/src/ariadnepy/resources/data/pathmeta.csv @@ -0,0 +1,5 @@ +from,to,source,version,url +chebi,rhea,Rhea,latest, +rhea,ec,Rhea,latest, +ec,ko,KEGG,latest, +ko,gmm,GM,v1,https://github.com/omixer/omixer-rpmR/raw/refs/heads/main/inst/extdata/GMMs.v1.07.txt diff --git a/tests/test_core/test_graph/test_plot.py b/tests/test_core/test_graph/test_plot.py index 455ac06..dd1c2e6 100644 --- a/tests/test_core/test_graph/test_plot.py +++ b/tests/test_core/test_graph/test_plot.py @@ -106,3 +106,87 @@ def test_plot_path_returns_exactly_one_axes(simple_graph): """The returned Figure must contain exactly one Axes object.""" fig = plot_path(simple_graph) assert len(fig.axes) == 1 + + +# ── R parity: res_name without 'by' (plotPath(graph, res.name = c(...))) ─────── + + +@pytest.fixture +def two_resource_graph() -> ig.Graph: + """A -[DB1]- B -[DB2]- C: two edges from different resources.""" + g = ig.Graph(directed=False) + for n in ("A", "B", "C"): + g.add_vertex(name=n) + g.add_edge(0, 1) + g.es[0]["source"] = "DB1" + g.add_edge(1, 2) + g.es[1]["source"] = "DB2" + return g + + +def test_plot_path_res_name_without_by_fades_other_resources(two_resource_graph): + """R: plotPath(graph, res.name = "DB1") fades the DB2 edge, not just DB1's.""" + fig = plot_path(two_resource_graph, res_name=["DB1"]) + ax = fig.axes[0] + alphas = sorted(line.get_alpha() for line in ax.lines) + assert alphas == pytest.approx([0.15, 1.0]) + + +def test_plot_path_res_name_without_by_focus_crops_to_resource(two_resource_graph): + """focus=True with only res_name (no 'by') must crop to that resource's nodes.""" + fig = plot_path(two_resource_graph, res_name=["DB1"], focus=True) + ax = fig.axes[0] + assert len(ax.collections[0].get_offsets()) == 2 # only A, B + + +def test_plot_path_res_name_without_by_no_focus_keeps_full_graph(two_resource_graph): + """Without focus, all 3 nodes stay on the canvas even though DB2 is faded.""" + fig = plot_path(two_resource_graph, res_name=["DB1"]) + ax = fig.axes[0] + assert len(ax.collections[0].get_offsets()) == 3 + + +# ── R parity: prune (fade) vs focus (crop) are independent axes ─────────────── + + +@pytest.fixture +def branching_graph() -> ig.Graph: + """A -[DB1]-> M1 -[DB1]-> C and A -[DB2]-> M2 -[DB2]-> C: two equal paths.""" + g = ig.Graph(directed=False) + for n in ("A", "M1", "M2", "C"): + g.add_vertex(name=n) + g.add_edge(g.vs.find(name="A").index, g.vs.find(name="M1").index) + g.es[-1]["source"] = "DB1" + g.add_edge(g.vs.find(name="M1").index, g.vs.find(name="C").index) + g.es[-1]["source"] = "DB1" + g.add_edge(g.vs.find(name="A").index, g.vs.find(name="M2").index) + g.es[-1]["source"] = "DB2" + g.add_edge(g.vs.find(name="M2").index, g.vs.find(name="C").index) + g.es[-1]["source"] = "DB2" + return g + + +def test_plot_path_prune_fades_but_keeps_full_graph_without_focus(branching_graph): + """R: prune=TRUE alone fades non-path edges but does not remove them.""" + fig = plot_path(branching_graph, "A ~ C", k=1, prune=True) + ax = fig.axes[0] + assert len(ax.lines) == 4 # all 4 edges still drawn + assert len(ax.collections[0].get_offsets()) == 4 # all 4 nodes still drawn + alphas = sorted(line.get_alpha() for line in ax.lines) + assert alphas == pytest.approx([0.15, 0.15, 1.0, 1.0]) + + +def test_plot_path_focus_alone_is_a_noop_without_prune_or_res_name(branching_graph): + """R quirk: focus=TRUE with prune=FALSE and no res_name fades nothing, so + there is nothing to crop — the full graph is still shown.""" + fig = plot_path(branching_graph, "A ~ C", k=1, focus=True) + ax = fig.axes[0] + assert len(ax.lines) == 4 + assert len(ax.collections[0].get_offsets()) == 4 + + +def test_plot_path_prune_and_focus_together_crop_to_path_only(branching_graph): + fig = plot_path(branching_graph, "A ~ C", k=1, prune=True, focus=True) + ax = fig.axes[0] + assert len(ax.lines) == 2 + assert len(ax.collections[0].get_offsets()) == 2 diff --git a/tests/test_core/test_graph/test_weave.py b/tests/test_core/test_graph/test_weave.py index 0cf1241..24fcb95 100644 --- a/tests/test_core/test_graph/test_weave.py +++ b/tests/test_core/test_graph/test_weave.py @@ -11,6 +11,8 @@ from ariadnepy.exceptions import AriadneError from ariadnepy.graph._weave import ( _draw_path, + _get_sorted_edge_key, + _graph_from_path_df, _map_complex_modules, _parse_by, _process_complex_modules, @@ -348,6 +350,92 @@ def _mock_fetch(_step, init, *_, **__): assert set(result["ko"]) == {"K00001"} +# ── weave_path / weave_complex data.frame method ──────────────────────────── +# R: weavePath(pathMeta, init=...) — no by/k/include/exclude/res.name slot. + +_PATH_META = pd.DataFrame({ + "from": ["ko"], + "to": ["ec"], + "source": ["FileDB"], +}) + + +@pytest.mark.parametrize("kwargs", [ + {"by": "ko ~ ec"}, + {"k": 2}, + {"include": ["x"]}, + {"exclude": ["x"]}, + {"res_name": ["FileDB"]}, +]) +def test_weave_path_dataframe_input_rejects_search_params(kwargs): + with pytest.raises(AriadneError, match="not supported"): + weave_path(_PATH_META, verbose=False, **kwargs) + + +@pytest.mark.parametrize("kwargs", [ + {"by": "ko ~ ec"}, + {"k": 2}, + {"include": ["x"]}, + {"exclude": ["x"]}, + {"res_name": ["FileDB"]}, +]) +def test_weave_complex_dataframe_input_rejects_search_params(kwargs): + with pytest.raises(AriadneError, match="not supported"): + weave_complex(_PATH_META, verbose=False, **kwargs) + + +def test_weave_path_igraph_input_requires_by(): + g = _file_graph() + with pytest.raises(AriadneError, match="'by' must be provided"): + weave_path(g, verbose=False) + + +def test_weave_complex_igraph_input_requires_by(gmm_graph): + with pytest.raises(AriadneError, match="'by' must be provided"): + weave_complex(gmm_graph, verbose=False) + + +def test_weave_path_dataframe_input_derives_by_and_runs(): + """R: chebi2gmm <- weavePath(pathMeta, init=c(15377, 30616, 4167))""" + g = _file_graph() + with ( + patch("ariadnepy.core._graph.ariadne", return_value=g), + patch("ariadnepy.graph._weave._fetch_edge", return_value=_MOCK_LM_KO_EC), + ): + result = weave_path(_PATH_META, use_names=False, verbose=False) + assert list(result.columns) == ["ko", "ec"] + + +def test_get_sorted_edge_key_is_direction_agnostic(): + assert _get_sorted_edge_key("a", "b", "DB1") == _get_sorted_edge_key("b", "a", "DB1") + + +def test_get_sorted_edge_key_distinguishes_source(): + assert _get_sorted_edge_key("a", "b", "DB1") != _get_sorted_edge_key("a", "b", "DB2") + + +def test_graph_from_path_df_keeps_only_matching_edges(): + full_graph = _ig_build( + ["ko", "ec", "other"], + [("ko", "ec", {"source": "FileDB", "url": "dummy"}), + ("ko", "other", {"source": "FileDB", "url": "dummy"})], + ) + with patch("ariadnepy.core._graph.ariadne", return_value=full_graph): + sub = _graph_from_path_df(_PATH_META) + assert set(sub.vs["name"]) == {"ko", "ec"} + assert sub.ecount() == 1 + + +def test_graph_from_path_df_derives_versions_from_version_column(): + path_df = pd.DataFrame({ + "from": ["ko"], "to": ["ec"], "source": ["FileDB"], "version": ["v1"], + }) + g = _file_graph() + with patch("ariadnepy.core._graph.ariadne", return_value=g) as mock_ariadne: + _graph_from_path_df(path_df) + mock_ariadne.assert_called_once_with(versions={"FileDB": "v1"}) + + # ── weave_complex ───────────────────────────────────────────────────────────── _MOCK_LM_EC_KO = pd.DataFrame({ From 554de7e85310b0f0b2d407b92f22737fc0cc1ca0 Mon Sep 17 00:00:00 2001 From: AditiAdhikari05 Date: Thu, 9 Jul 2026 16:20:36 +0545 Subject: [PATCH 2/2] add parameters --- src/ariadnepy/plot/_draw.py | 4 +++- tests/test_core/test_graph/test_plot.py | 27 +++++++++++++++++-------- 2 files changed, 22 insertions(+), 9 deletions(-) diff --git a/src/ariadnepy/plot/_draw.py b/src/ariadnepy/plot/_draw.py index 7ff761b..39a9109 100644 --- a/src/ariadnepy/plot/_draw.py +++ b/src/ariadnepy/plot/_draw.py @@ -116,7 +116,9 @@ def plot_path( } if focus: - keep_edge_idx = [e.index for e, info in zip(graph.es, edges_info, strict=False) if info["alpha"]] + keep_edge_idx = [ + e.index for e, info in zip(graph.es, edges_info, strict=False) if info["alpha"] + ] draw_graph = graph.subgraph_edges(keep_edge_idx, delete_vertices=True) draw_edges_info = [info for info in edges_info if info["alpha"]] else: diff --git a/tests/test_core/test_graph/test_plot.py b/tests/test_core/test_graph/test_plot.py index dd1c2e6..97f654b 100644 --- a/tests/test_core/test_graph/test_plot.py +++ b/tests/test_core/test_graph/test_plot.py @@ -133,17 +133,23 @@ def test_plot_path_res_name_without_by_fades_other_resources(two_resource_graph) def test_plot_path_res_name_without_by_focus_crops_to_resource(two_resource_graph): - """focus=True with only res_name (no 'by') must crop to that resource's nodes.""" + """focus=True with only res_name (no 'by') must crop the graph structure + itself down to just the DB1 edge — only 1 line, only A/B remain.""" fig = plot_path(two_resource_graph, res_name=["DB1"], focus=True) ax = fig.axes[0] + assert len(ax.lines) == 1 assert len(ax.collections[0].get_offsets()) == 2 # only A, B -def test_plot_path_res_name_without_by_no_focus_keeps_full_graph(two_resource_graph): - """Without focus, all 3 nodes stay on the canvas even though DB2 is faded.""" +def test_plot_path_res_name_without_by_no_focus_keeps_edge_structure(two_resource_graph): + """Without focus, both edges stay on the canvas (DB2 merely faded), but a + node marker/label is only drawn for nodes that participate in DB1 — same + as R's geom_node_point(aes(filter = alpha)), which removes non-alpha + nodes from rendering regardless of 'focus'.""" fig = plot_path(two_resource_graph, res_name=["DB1"]) ax = fig.axes[0] - assert len(ax.collections[0].get_offsets()) == 3 + assert len(ax.lines) == 2 # DB1 and DB2 edges both still drawn + assert len(ax.collections[0].get_offsets()) == 2 # only A, B (not C) # ── R parity: prune (fade) vs focus (crop) are independent axes ─────────────── @@ -166,12 +172,15 @@ def branching_graph() -> ig.Graph: return g -def test_plot_path_prune_fades_but_keeps_full_graph_without_focus(branching_graph): - """R: prune=TRUE alone fades non-path edges but does not remove them.""" +def test_plot_path_prune_fades_but_keeps_edges_without_focus(branching_graph): + """R: prune=TRUE alone fades non-path edges but does not remove them. + Node markers are still filtered to path-connected nodes only (A, M1, C) — + M2 has no alpha=TRUE edge, so its marker/label is dropped even though its + faded edges remain drawn, matching geom_node_point's filter aesthetic.""" fig = plot_path(branching_graph, "A ~ C", k=1, prune=True) ax = fig.axes[0] assert len(ax.lines) == 4 # all 4 edges still drawn - assert len(ax.collections[0].get_offsets()) == 4 # all 4 nodes still drawn + assert len(ax.collections[0].get_offsets()) == 3 # A, M1, C (not M2) alphas = sorted(line.get_alpha() for line in ax.lines) assert alphas == pytest.approx([0.15, 0.15, 1.0, 1.0]) @@ -186,7 +195,9 @@ def test_plot_path_focus_alone_is_a_noop_without_prune_or_res_name(branching_gra def test_plot_path_prune_and_focus_together_crop_to_path_only(branching_graph): + """focus=True structurally drops the 2 non-path edges, leaving the 2-edge + path A-M1-C — 3 nodes, not 2, since a 2-edge path has 3 stops.""" fig = plot_path(branching_graph, "A ~ C", k=1, prune=True, focus=True) ax = fig.axes[0] assert len(ax.lines) == 2 - assert len(ax.collections[0].get_offsets()) == 2 + assert len(ax.collections[0].get_offsets()) == 3 # A, M1, C