diff --git a/dascore/viz/__init__.py b/dascore/viz/__init__.py index 992b36e2f..8048fd1e8 100644 --- a/dascore/viz/__init__.py +++ b/dascore/viz/__init__.py @@ -2,13 +2,14 @@ Module for static, matplotlib-based visualizations and figure generation. """ from __future__ import annotations -from dascore.utils.namespace import PatchNameSpace +from dascore.utils.namespace import InventoryNameSpace, PatchNameSpace from .spectrogram import spectrogram from .specplot import specplot from .waterfall import waterfall from .wiggle import wiggle from .map_fiber import map_fiber +from .inventory import map_path, path, timeline class VizPatchNameSpace(PatchNameSpace): @@ -21,3 +22,13 @@ class VizPatchNameSpace(PatchNameSpace): specplot = specplot wiggle = wiggle map_fiber = map_fiber + + +class VizInventoryNameSpace(InventoryNameSpace): + """The plots an inventory can draw of itself.""" + + name = "viz" + + path = path + map = map_path + timeline = timeline diff --git a/dascore/viz/_lanes.py b/dascore/viz/_lanes.py new file mode 100644 index 000000000..6ff58dc7c --- /dev/null +++ b/dascore/viz/_lanes.py @@ -0,0 +1,531 @@ +""" +A general renderer for intervals laid out in horizontal lanes. + +The inventory draws its tracks with this, a spool can draw what it +covers, and an annotation set is the same shape over a patch dimension. +So the input is a dataframe of intervals rather than any one of those +objects, and the columns it reads are named by the caller. +""" + +from __future__ import annotations + +import datetime +from collections.abc import Mapping, Sequence + +import matplotlib.dates as mdates +import matplotlib.patheffects as pe +import matplotlib.pyplot as plt +import numpy as np +import pandas as pd +from matplotlib.collections import PatchCollection +from matplotlib.colors import BoundaryNorm, ListedColormap +from matplotlib.patches import Patch as PatchArtist +from matplotlib.patches import Rectangle + +from dascore.exceptions import ParameterError +from dascore.utils.intervals import normalize_value, value_kind +from dascore.utils.plotting import _format_time_axis, _get_ax, _get_cmap + +# Palettes are module level so that two figures of one inventory agree. +STRING_CMAP = "tab20" +# tab20 runs dark, light, dark, light, so consecutive categories come out +# as two shades of one hue and read as one variable. Take the dark half +# first, and skip its two greys, which the uncovered colors already use. +WHEEL_ORDER = (0, 2, 4, 6, 8, 10, 12, 16, 18, 1, 3, 5, 7, 9, 11, 13, 17, 19) +LANE_CMAP = "tab10" +NUMERIC_CMAP = "viridis" +UNCOVERED_COLOR = "0.7" + +# The fraction of the x axis hatched where a bar runs off the end of it. +_OPEN_FRACTION = 0.02 +_MAX_SUB_ROWS = 8 +# Past this many distinct numbers a lane earns a colorbar rather than +# relying on the value printed in each box. +_MAX_DISCRETE = 6 + + +def _as_numeric(values): + """Return values as floats, converting datetimes to matplotlib dates.""" + array = np.asarray(values) + if _is_dated(values): + # Losing nanosecond precision is fine; this is a picture. + stamps = pd.DatetimeIndex(array.ravel()) + if stamps.tz is not None: + stamps = stamps.tz_convert("UTC").tz_localize(None) + return mdates.date2num(stamps.to_numpy()).reshape(array.shape) + return array.astype(float) + + +def _default_label(value) -> str: + """Text for a value which was not given a label of its own.""" + if isinstance(value, str): + return value + if isinstance(value, bool) or value is None: + return "" + # A number states itself; a boolean group is named by its lane instead. + return f"{value:g}" if isinstance(value, float) else str(value) + + +def _is_dated(values) -> bool: + """Whether a column of interval bounds states times rather than numbers.""" + array = np.asarray(values) + return bool( + pd.api.types.is_datetime64_any_dtype(values) + or ( + array.dtype == object + and len(array) + and isinstance(array.flat[0], datetime.datetime | np.datetime64) + ) + ) + + +def _read_frame(intervals, start, end, lane, value, label): + """Pull the named columns out into a frame, and say if it is dated.""" + if not isinstance(intervals, pd.DataFrame): + intervals = pd.DataFrame(intervals) + missing = [x for x in (start, end) if x not in intervals.columns] + if missing: + msg = ( + f"An interval frame needs the columns {sorted(missing)}; this one " + f"has {list(intervals.columns)}. Name the columns holding the " + "interval bounds with the start and end arguments." + ) + raise ParameterError(msg) + for name, kind in ((lane, "lane"), (value, "value"), (label, "label")): + if name is not None and name not in intervals.columns: + msg = ( + f"{kind}={name!r} is not a column of this frame, which has " + f"{list(intervals.columns)}." + ) + raise ParameterError(msg) + out = pd.DataFrame(index=intervals.index) + out["start"] = _as_numeric(intervals[start]) + out["end"] = _as_numeric(intervals[end]) + out["lane"] = intervals[lane].astype(str) if lane else "" + out["value"] = intervals[value] if value else None + if label: + out["label"] = intervals[label].astype(str) + elif value: + out["label"] = [_default_label(x) for x in intervals[value].tolist()] + else: + out["label"] = "" + for flag in ("open_start", "open_end"): + col = intervals[flag] if flag in intervals.columns else False + out[flag] = np.asarray(col, dtype=bool) if flag in intervals.columns else False + dated = _is_dated(intervals[start]) or _is_dated(intervals[end]) + return out, dated + + +def _lane_kind(values) -> str: + """Return the one value kind a lane states, refusing a mixture.""" + kinds = {value_kind(normalize_value(x)) for x in values if x is not None} + kinds.discard(None) + if not kinds: + return "none" + if len(kinds) > 1: + return "mixed" + return kinds.pop() + + +def _pack_rows(frame) -> np.ndarray: + """Assign each interval a sub-row so overlapping ones do not collide.""" + order = np.argsort(frame["start"].to_numpy(), kind="stable") + rows = np.zeros(len(frame), dtype=int) + ends: list[float] = [] + starts = frame["start"].to_numpy() + stops = frame["end"].to_numpy() + for index in order: + for row, last in enumerate(ends): + if starts[index] >= last: + rows[index] = row + ends[row] = stops[index] + break + else: + rows[index] = len(ends) + ends.append(stops[index]) + return np.minimum(rows, _MAX_SUB_ROWS - 1) + + +def _string_colors(frame, vocabulary=None, cmap_name=STRING_CMAP) -> dict: + """Map every string value to a stable color. + + The vocabulary widens the palette beyond what this frame holds, so a + figure of part of a subject colors it as a figure of all of it does. + """ + seen = list(frame["value"].tolist()) + list(vocabulary or []) + values = sorted({x for x in seen if isinstance(x, str) and x != ""}) + cmap = plt.get_cmap(cmap_name) + return { + value: cmap(WHEEL_ORDER[index % len(WHEEL_ORDER)]) + for index, value in enumerate(values) + } + + +def numeric_scale(values, cmap_name=NUMERIC_CMAP): + """Return (cmap, norm, ticks) for a column of numbers. + + A handful of distinct values is a set of categories which happen to + be numbered, so it gets one color each and a stepped bar reading at + the values themselves. Anything more is a quantity, and ramps. + """ + finite = np.asarray(values, dtype=float) + finite = finite[np.isfinite(finite)] + base = _get_cmap(cmap_name) + unique = np.unique(finite) + if len(unique) < 2: + low = float(unique[0]) if len(unique) else 0.0 + return base, plt.Normalize(low, low + 1.0), None + if len(unique) <= _MAX_DISCRETE: + picks = np.linspace(0.12, 0.9, len(unique)) + listed = ListedColormap([base(x) for x in picks]) + middles = (unique[:-1] + unique[1:]) / 2 + edges = np.concatenate( + [ + [unique[0] - (middles[0] - unique[0])], + middles, + [unique[-1] + (unique[-1] - middles[-1])], + ] + ) + return listed, BoundaryNorm(edges, listed.N), unique + return base, plt.Normalize(float(unique.min()), float(unique.max())), None + + +def _resolve_colors(rows, kind, lane_index, string_map, color): + """Return one color per row, and a legend/colorbar description.""" + if isinstance(color, Mapping) and any( + isinstance(x, Mapping) for x in color.values() + ): + # Keyed by lane. A lane the mapping does not name takes the + # default treatment rather than being matched against lane names. + color = color.get(rows["lane"].iloc[0]) + if isinstance(color, Mapping): + colors = [color.get(x, UNCOVERED_COLOR) for x in rows["value"]] + used = {x: color[x] for x in rows["value"] if x in color} + return colors, ("legend", used) + if isinstance(color, str) and kind != "numeric": + return [color] * len(rows), None + if kind == "string": + colors = [string_map.get(x, UNCOVERED_COLOR) for x in rows["value"]] + return colors, ( + "legend", + {x: string_map[x] for x in rows["value"] if x in string_map}, + ) + if kind == "numeric": + values = np.asarray( + [float(normalize_value(x)) for x in rows["value"]], dtype=float + ) + if isinstance(color, str): + try: + cmap = _get_cmap(color) + except (ValueError, KeyError): + # A color name, not a colormap: one color for the lane, as + # a lane of any other kind would take it. + return [color] * len(rows), None + else: + cmap = _get_cmap(NUMERIC_CMAP) + if len(np.unique(values[np.isfinite(values)])) < 2: + # One value is not a scale, so it gets a color and its number + # rather than a colorbar reading from it to a value nothing has. + return [cmap(0.5)] * len(rows), None + cmap, norm, ticks = numeric_scale(values, getattr(cmap, "name", NUMERIC_CMAP)) + # A value nothing states maps to a transparent color unless the + # colormap is told otherwise, and the box would simply vanish. + cmap = cmap.with_extremes(bad=UNCOVERED_COLOR) + colors = [cmap(norm(x)) for x in values] + if ticks is not None: + # Few enough to be read off the boxes they are printed in. + return colors, None + # Each numeric lane is its own scale, so each earns its own bar; + # one bar for two lanes would read from a scale only one of them has. + return colors, ("colorbar", (rows["lane"].iloc[0], cmap, norm)) + # Boolean and unvalued lanes take one color, so the lane reads as one + # variable; a False interval is drawn faintly rather than dropped. + base = plt.get_cmap(LANE_CMAP)(lane_index % 10) + colors = [ + base if normalize_value(x) is not False else (*base[:3], 0.25) + for x in rows["value"] + ] + return colors, ("legend", {rows["lane"].iloc[0]: base}) + + +def _draw_open_edges(ax, rows, y_low, height, colors, span): + """Hatch the outer sliver of any bar which runs off the axis.""" + marks = [] + width = span * _OPEN_FRACTION + for (_, row), color in zip(rows.iterrows(), colors, strict=True): + for flag, edge in (("open_start", row["start"]), ("open_end", row["end"])): + if not row[flag]: + continue + left = edge if flag == "open_start" else edge - width + marks.append((Rectangle((left, y_low), width, height), color)) + if not marks: + return + patches = PatchCollection( + [x for x, _ in marks], + facecolors=[c for _, c in marks], + hatch="///", + edgecolor="white", + linewidth=0, + zorder=3, + ) + ax.add_collection(patches) + + +def _fit_labels(ax, placements, max_labels): + """Draw the labels which fit in their box, and drop the rest.""" + if len(placements) > max_labels: + return + figure = ax.get_figure() + # Lay the figure out before measuring: a label is compared against its + # box in pixels, and both move when the axes does. + figure.draw_without_rendering() + renderer = figure.canvas.get_renderer() + transform = ax.transData + for text, x_mid, y_mid, width in placements: + if not text: + continue + artist = ax.text( + x_mid, + y_mid, + text, + ha="center", + va="center", + fontsize=plt.rcParams["font.size"] * 0.8, + zorder=4, + clip_on=True, + # A dark fill would otherwise swallow the text sitting on it. + path_effects=[pe.withStroke(linewidth=1.3, foreground="white")], + ) + left = transform.transform((x_mid - width / 2, y_mid))[0] + right = transform.transform((x_mid + width / 2, y_mid))[0] + if artist.get_window_extent(renderer).width > (right - left): + artist.remove() + + +def plot_lanes( + intervals, + ax: plt.Axes | None = None, + *, + start: str = "start", + end: str = "end", + lane: str | None = None, + value: str | None = None, + label: str | None = None, + lanes: Sequence[str] | None = None, + color=None, + vocabulary: Sequence | None = None, + pack: bool = True, + legend: bool | str = "auto", + max_labels: int = 200, + x_limits: tuple | None = None, + x_label: str = "", + lane_height: float = 0.8, + colorbar_axes: Sequence[plt.Axes] | None = None, + show: bool = False, +) -> plt.Axes: + """ + Draw a frame of intervals as horizontal lanes. + + Parameters + ---------- + intervals + A dataframe with one row per interval. + ax + A matplotlib Axes; one is created when None. + start, end + Columns holding the interval bounds. They may be numbers or + datetimes, and equal bounds make the row a point marker. + lane + Column naming the lane a row belongs to; None puts every row in + one unnamed lane. + value + Column deciding each row's color. Strings are categorical, + numbers continuous, and booleans state membership of the lane. + label + Column holding the text drawn in each box. Values supply it by + default: text as itself, a number as its digits, a boolean as + nothing, since the lane it sits in already names it. + lanes + The lanes to draw, in order. Names with no rows are kept as empty + lanes, so two figures of different subjects still line up. + color + A color for every row, a mapping of value to color, or a mapping + of lane name to such a mapping. A lane whose values are numbers + reads a color string as the name of a colormap, or as a color + where it names no colormap. + vocabulary + Values to reserve colors for beyond those this frame holds, so a + figure of part of a subject colors it as a figure of all of it. + pack + Whether overlapping intervals are packed into sub-rows. + legend + Whether to draw a legend and any colorbars. False, or "off", + draws neither; anything else draws what the colors earn. + max_labels + Draw no text at all past this many intervals. + x_limits + Limits for the x axis, in data units. + x_label + Label for the x axis. + lane_height + Fraction of a lane's row filled by its bars. + colorbar_axes + The axes a colorbar takes its room from; the drawn axes alone by + default. Pass every axes of a shared-x figure, or the others keep + a width this one gives up. + show + Whether to call plt.show. + + Examples + -------- + >>> import pandas as pd + >>> from dascore.viz._lanes import plot_lanes + >>> + >>> frame = pd.DataFrame( + ... { + ... "group": ["zone", "zone", "noisy"], + ... "start": [0.0, 10.0, 5.0], + ... "end": [10.0, 20.0, 15.0], + ... "value": ["north", "south", True], + ... } + ... ) + >>> _ = plot_lanes(frame, lane="group", value="value") + """ + frame, dated = _read_frame(intervals, start, end, lane, value, label) + if not len(frame): + msg = "The interval frame holds no rows, so there is nothing to draw." + raise ParameterError(msg) + backwards = frame["end"] < frame["start"] + if backwards.any(): + row = frame[backwards].iloc[0] + msg = ( + f"Interval ({row['start']}, {row['end']}) in lane " + f"{row['lane']!r} ends before it starts." + ) + raise ParameterError(msg) + ax = _get_ax(ax) + order = list(dict.fromkeys(frame["lane"])) if lanes is None else list(lanes) + if lanes is not None and len(set(order)) != len(order): + msg = f"lanes names a lane twice; each lane is drawn once. Got {lanes}." + raise ParameterError(msg) + # A lane given its own mapping is colored from that, so its values + # must not also spend slots in the palette the other lanes draw from. + pinned = set() + if isinstance(color, Mapping): + pinned = {k for k, v in color.items() if isinstance(v, Mapping)} + unpinned = frame[~frame["lane"].isin(pinned)] if pinned else frame + string_map = _string_colors(unpinned, vocabulary) + # Fix the x limits before any text, since a label is measured in pixels. + if x_limits is None: + low = float(np.nanmin(frame["start"])) + high = float(np.nanmax(frame["end"])) + pad = (high - low) * 0.02 or 0.5 + x_limits = (low - pad, high + pad) + else: + x_limits = tuple(float(x) for x in _as_numeric(np.asarray(x_limits))) + ax.set_xlim(*x_limits) + span = x_limits[1] - x_limits[0] + + legend_entries: dict = {} + colorbars: list[tuple] = [] + placements: list[tuple] = [] + for index, name in enumerate(order): + rows = frame[frame["lane"] == name] + y_centre = -index + if not len(rows): + continue + kind = _lane_kind(rows["value"]) + if kind == "mixed": + msg = ( + f"Lane {name!r} mixes value kinds, so it has no one color " + "scheme. A group states one variable; split the kinds into " + "separate lanes." + ) + raise ParameterError(msg) + sub_rows = _pack_rows(rows) if pack else np.zeros(len(rows), dtype=int) + n_sub = int(sub_rows.max()) + 1 + height = lane_height / n_sub + colors, described = _resolve_colors(rows, kind, index, string_map, color) + if described and described[0] == "legend": + legend_entries.update(described[1]) + elif described and described[0] == "colorbar": + colorbars.append(described[1]) + boxes, box_colors, points, point_colors = [], [], [], [] + for (_, row), row_color, sub in zip( + rows.iterrows(), colors, sub_rows, strict=True + ): + low = y_centre - lane_height / 2 + sub * height + width = row["end"] - row["start"] + if width <= 0: + # A point marker covers nothing but still documents a place. + points.append((row["start"], low, height)) + point_colors.append(row_color) + continue + boxes.append(Rectangle((row["start"], low), width, height)) + box_colors.append(row_color) + placements.append( + (row["label"], row["start"] + width / 2, low + height / 2, width) + ) + if boxes: + ax.add_collection( + PatchCollection( + boxes, + facecolors=box_colors, + edgecolor="white", + linewidth=0.5, + zorder=2, + ) + ) + for (x, low, tall), point_color in zip(points, point_colors, strict=True): + ax.plot( + [x, x], + [low, low + tall], + color=point_color, + linewidth=2.0, + zorder=3, + solid_capstyle="butt", + ) + ax.plot([x], [low + tall], marker="v", markersize=5, color=point_color) + _draw_open_edges( + ax, rows, y_centre - lane_height / 2, lane_height, colors, span + ) + + ax.set_yticks(-np.arange(len(order)), [str(x) for x in order]) + ax.set_ylim(-(len(order) - 1) - lane_height, lane_height) + if x_label: + ax.set_xlabel(x_label) + ax.grid(axis="x", color="0.85", linewidth=0.5, zorder=0) + ax.set_axisbelow(True) + # The left spine is hidden, so its tick marks are dashes after a name. + ax.tick_params(axis="y", length=0, pad=4) + for side in ("top", "right", "left"): + ax.spines[side].set_visible(False) + if dated: + _format_time_axis(ax, x_label or "time", "x") + _fit_labels(ax, placements, max_labels) + if legend and legend != "off": + for name, cmap, norm in colorbars: + bar = ax.get_figure().colorbar( + plt.cm.ScalarMappable(norm=norm, cmap=cmap), + ax=list(colorbar_axes) if colorbar_axes else ax, + fraction=0.05, + pad=0.02, + ) + bar.set_label(name) + if legend and legend_entries and legend != "off": + handles = [ + PatchArtist(facecolor=color, label=name) + for name, color in legend_entries.items() + ] + # A colorbar already occupies the strip beside the axes. + offset = 1.01 + 0.17 * len(colorbars) + ax.legend( + handles=handles, + loc="upper left", + bbox_to_anchor=(offset, 1.0), + frameon=False, + fontsize="small", + ) + if show: + plt.show() + return ax diff --git a/dascore/viz/inventory.py b/dascore/viz/inventory.py new file mode 100644 index 000000000..99be03626 --- /dev/null +++ b/dascore/viz/inventory.py @@ -0,0 +1,1018 @@ +"""Visualizations of an inventory: its path, its layout, and its epochs.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import TYPE_CHECKING + +import matplotlib.colors as mcolors +import matplotlib.dates as mdates +import matplotlib.pyplot as plt +import numpy as np +import pandas as pd +from matplotlib.collections import LineCollection +from matplotlib.patches import Patch as PatchArtist + +from dascore.exceptions import InvalidInventoryError, ParameterError +from dascore.utils.intervals import interval_masks, normalize_value, value_kind +from dascore.utils.plotting import _format_time_axis, _get_ax + +from . import _lanes +from ._lanes import UNCOVERED_COLOR, _default_label, plot_lanes + +if TYPE_CHECKING: + from dascore.constants import timeable_types + from dascore.core.inventory import Inventory, OpticalPath + +# Components are a closed set, so their colors can be too. +# Okabe-Ito, so the components stay apart from each other under color +# blindness and off the hue grid the label groups are drawn from. +COMPONENT_COLORS = { + "FiberSegment": "#0072B2", + "Splice": "#E69F00", + "Connector": "#009E73", + "Terminator": "#CC79A7", +} + + +def _iter_paths(inventory): + """Yield every optical path with the address which names it.""" + for network in inventory.networks: + for array in network.fiber_arrays: + for path in array.optical_paths: + address = f"{network.code}.{array.code}.{path.location_code}" + yield address, network, array, path + + +def _effective_epoch(*models): + """The epoch a model is really valid over, clipped by its containers.""" + start, end = pd.NaT, pd.NaT + for model in models: + low, high = model.start_time, model.end_time + if not pd.isnull(low): + start = low if pd.isnull(start) else max(start, low) + if not pd.isnull(high): + end = high if pd.isnull(end) else min(end, high) + return start, end + + +def _effective_at(time, *models) -> bool: + """Whether a time falls inside every one of these epochs. + + A child which states no bound defers to its container, so asking the + path alone would call it valid whenever the path is, which is every + time it states nothing. + """ + if time is None: + return True + return all(x.is_effective_at(time) for x in models) + + +def _sample_distances(path, low: float, high: float, count: int) -> np.ndarray: + """Distances to read a path's geometry at, between low and high. + + A uniform grid can step straight over an unsurveyed stretch shorter + than its spacing, and the picture would then bridge fiber nobody + placed. Every gap contributes a sample, so every gap is seen. + """ + spans = sorted( + (float(min(x.distance)), float(max(x.distance))) for x in path.geometry + ) + covered: list[list[float]] = [] + for start, end in spans: + if covered and start <= covered[-1][1]: + covered[-1][1] = max(covered[-1][1], end) + else: + covered.append([start, end]) + holes = [ + 0.5 * (covered[index][1] + covered[index + 1][0]) + for index in range(len(covered) - 1) + ] + inside = [x for x in holes if low < x < high] + grid = np.linspace(low, high, count) + if not inside: + return grid + return np.unique(np.concatenate([grid, np.asarray(inside, dtype=float)])) + + +def _epoch_label(path) -> str: + """Name a path epoch by when it starts, for a chart title.""" + if pd.isnull(path.start_time): + return "from the beginning" + return f"from {str(path.start_time)[:10]}" + + +def _select_path(inventory, optical_path=None, acquisition_key=None, time=None): + """Return the (address, array, path) a caller means, or explain.""" + found = list(_iter_paths(inventory)) + if not found: + msg = "This inventory holds no optical paths, so there is nothing to plot." + raise ParameterError(msg) + if acquisition_key is not None: + try: + context = inventory.resolve(acquisition_key, time) + except InvalidInventoryError as error: + # Keep what resolve said; a key can fail for reasons no time + # fixes, and claiming ambiguity would hide them. + hint = ( + "" + if time is not None + else " Pass a time as well, if the key names several epochs." + ) + msg = ( + f"Acquisition key {acquisition_key!r} does not resolve to one " + f"acquisition: {error}{hint}" + ) + raise ParameterError(msg) from error + if context.optical_path is None: + msg = ( + f"Acquisition key {acquisition_key!r} resolves to no optical " + "path, so there is nothing to draw against optical distance." + ) + raise ParameterError(msg) + for address, _, array, path in found: + if path is context.optical_path: + return address, array, path + if optical_path is not None and not isinstance(optical_path, str): + for address, _, array, path in found: + if path is optical_path: + return address, array, path + msg = "That optical path is not part of this inventory." + raise ParameterError(msg) + candidates = found + if time is not None: + candidates = [x for x in found if _effective_at(time, x[1], x[2], x[3])] + if not candidates: + stated = sorted({f"{x[0]} ({_epoch_label(x[3])})" for x in found}) + msg = ( + f"No optical path is effective at {time}. The paths are: " + + ", ".join(stated) + + "." + ) + raise ParameterError(msg) + if optical_path is not None: + matched = [x for x in candidates if x[0] == optical_path] + if not matched: + matched = [x for x in candidates if x[3].name == optical_path] + if not matched: + names = sorted({x[0] for x in candidates}) + msg = f"No optical path matches {optical_path!r}. The paths are: {names}." + raise ParameterError(msg) + candidates = matched + if len(candidates) == 1: + address, _, array, path = candidates[0] + return address, array, path + names = sorted({f"{x[0]} ({_epoch_label(x[3])})" for x in candidates}) + if len({x[0] for x in candidates}) == 1: + # One address, several epochs of it: only a time tells them apart. + msg = ( + f"Optical path {candidates[0][0]!r} has {len(candidates)} epochs, " + "so which one to plot must be stated. Pass a time, since an " + "address names the path rather than one epoch of it. The epochs " + "are: " + ", ".join(names) + "." + ) + raise ParameterError(msg) + msg = ( + f"This inventory holds {len(candidates)} optical paths, so which one " + "to plot must be stated. Pass optical_path=
, " + "acquisition_key=, or a time. The paths are: " + ", ".join(names) + "." + ) + raise ParameterError(msg) + + +def _path_acquisitions(array, path, time=None): + """The acquisitions which interrogate a path while it is valid.""" + out = [] + for acquisition in array.acquisitions: + if acquisition.location_code != path.location_code: + continue + if not acquisition.overlaps(path): + continue + if time is not None and not acquisition.is_effective_at(time): + continue + out.append(acquisition) + return out + + +def _track_frame(path, acquisitions) -> pd.DataFrame: + """Flatten a path's tracks into one frame of intervals.""" + rows = [] + for acquisition in acquisitions: + dist_map = acquisition.distance_map + if dist_map is None: + continue + distances = dist_map.distance + rows.append( + { + "lane": f"channels ({acquisition.code})", + "start": float(distances[0]), + "end": float(distances[-1]), + "value": acquisition.code, + "label": acquisition.code, + } + ) + for component, (low, high) in zip( + path.optical_components, path.component_intervals(), strict=True + ): + rows.append( + { + "lane": "components", + "start": low, + "end": high, + "value": type(component).__name__, + "label": component.name or type(component).__name__, + } + ) + for coupling in path.coupling: + rows.append( + { + "lane": "coupling", + "start": coupling.start_distance, + "end": coupling.end_distance, + "value": coupling.coupling_type, + "label": coupling.coupling_type, + } + ) + for item in path.labels: + rows.append( + { + "lane": item.group, + "start": item.start_distance, + "end": item.end_distance, + "value": item.value, + # The renderer's own rule for what a value reads as. + "label": _default_label(item.value), + } + ) + return pd.DataFrame(rows) + + +TRACKS = ("channels", "components", "coupling") + + +def _column_panels(path, columns, crs) -> list[str]: + """Decide which geometry columns get their own line panel.""" + if not columns: + return [] + axes = tuple(crs.coordinate_labels) + stated = [x for x in path.geometry_columns() if x not in axes] + wanted = [columns] if isinstance(columns, str) else list(columns) + for name in wanted: + if name in axes: + msg = ( + f"{name!r} is a position axis of the CRS, which map() draws; " + f"path() draws the columns along the fiber, here {tuple(stated)}." + ) + raise ParameterError(msg) + if name not in stated: + msg = ( + f"This optical path states no geometry column named {name!r}; " + f"it states {tuple(stated)}." + ) + raise ParameterError(msg) + return wanted + + +def _column_units(path, name) -> str: + """The units a geometry column is stated in, if any.""" + for segment in path.geometry: + if name in segment.units: + return segment.units[name] + return "" + + +def _select_tracks(frame, tracks, path): + """Keep only the lanes a caller asked for, in the order asked.""" + if tracks is None: + return frame + groups = tuple(dict.fromkeys(x.group for x in path.labels)) + wanted = [tracks] if isinstance(tracks, str) else list(tracks) + keep = [] + for name in wanted: + if name == "channels": + keep.extend( + x for x in dict.fromkeys(frame["lane"]) if x.startswith("channels") + ) + elif name in TRACKS or name in groups: + keep.append(name) + else: + msg = ( + f"{name!r} is not a track of this optical path; the tracks are " + f"{TRACKS} and the label groups are {groups}." + ) + raise ParameterError(msg) + out = frame[frame["lane"].isin(keep)] + if out.empty: + msg = f"This optical path has nothing to draw for tracks={tracks!r}." + raise ParameterError(msg) + order = {lane: index for index, lane in enumerate(keep)} + return out.sort_values("lane", key=lambda col: col.map(order), kind="stable") + + +def _distance_window(asked, span): + """Resolve a (low, high) distance selection against a path's span.""" + if asked is None: + return span + try: + low, high = asked + except (TypeError, ValueError): + msg = f"distance={asked!r} must be a (low, high) pair." + raise ParameterError(msg) from None + low = span[0] if low is None or low is ... else float(low) + high = span[1] if high is None or high is ... else float(high) + if high <= low: + msg = f"distance={asked!r} must be increasing." + raise ParameterError(msg) + if high <= span[0] or low >= span[1]: + msg = ( + f"distance={asked!r} lies outside the path's span {span}, so it " + "clips everything away." + ) + raise ParameterError(msg) + return (low, high) + + +def path( + inventory: Inventory, + optical_path: str | OpticalPath | None = None, + *, + acquisition_key: str | None = None, + time: timeable_types | None = None, + distance: tuple | None = None, + tracks: str | Sequence[str] | None = None, + columns: str | Sequence[str] | None = None, + n_samples: int = 1000, + color: str | Mapping | None = None, + max_labels: int = 200, + ax: plt.Axes | None = None, + figsize: tuple[float, float] | None = None, + show: bool = False, +) -> plt.Axes: + """ + Plot what lies along one optical path, against optical distance. + + Every track the path describes becomes a lane: the channels each + acquisition places on it, the optical components which give it its + length, how it is coupled to the ground, and one lane per label + group. A geometry column such as chainage or depth can be drawn as a + line panel beneath, sharing the distance axis; it breaks wherever the + path states no value rather than bridging the gap. Where the fiber + physically is belongs to map(). + + Parameters + ---------- + inventory + The inventory holding the path. + optical_path + The path to draw, as an ``network.array.location`` address, a + path name, or the object. Optional when the choice is not + ambiguous. + acquisition_key + Resolve the path from an acquisition key instead. + time + The instant to resolve at, which is how one epoch of a repaired + path is chosen. + distance + The optical distances to draw between, as (low, high). Either + end may be None, or ..., to run to the path's own bound. A long + lead-in otherwise crushes the instrumented part into a corner. + tracks + Which lanes to draw, in order: any of "channels", "components", + "coupling", and the path's label group names. None draws all. + columns + Geometry columns to draw as line panels beneath the lanes. The + CRS's position axes are refused, since they belong on a map. + n_samples + How finely the columns are sampled. + color + Passed to the lane renderer to override its colors. + max_labels + Draw no lane text at all past this many intervals. + ax + An Axes to draw the lanes on. Column panels need their own + figure, so passing this and naming columns is refused. + figsize + Size of the figure built when ax is None. + show + Whether to call plt.show. + + Examples + -------- + >>> import dascore as dc + >>> from dascore.viz.inventory import path + >>> + >>> inventory = dc.get_example_inventory("tunnel") + >>> _ = path(inventory, time="2024-07-01", distance=(1495, 1780)) + >>> _ = path(inventory, time="2024-07-01", tracks=("coupling", "section")) + """ + address, array, chosen = _select_path( + inventory, optical_path, acquisition_key, time + ) + crs = inventory.coordinate_reference_system + columns = _column_panels(chosen, columns, crs) + if ax is not None and columns: + msg = ( + "path draws its columns in their own panels, so it builds the " + "figure and cannot add them to the axes passed as ax. Pass ax " + "without columns, or leave ax unset." + ) + raise ParameterError(msg) + if chosen.optical_length <= 0: + msg = ( + f"Optical path {address!r} has no length, since its components " + "state none, so there is no distance axis to draw." + ) + raise ParameterError(msg) + limits = _distance_window(distance, (chosen.start_distance, chosen.end_distance)) + frame = _track_frame(chosen, _path_acquisitions(array, chosen, time)) + # The palette is the path's, not this figure's, so drawing some of the + # tracks colors them as drawing all of them does. + vocabulary = list(frame.loc[frame["lane"] != "components", "value"]) + frame = _select_tracks(frame, tracks, chosen) + lanes = list(dict.fromkeys(frame["lane"])) + if ax is None: + # Capped: a figure taller than a page is not more readable. + height = min(1.2 + 0.42 * len(lanes) + 1.1 * len(columns), 14.0) + figure, all_axes = plt.subplots( + 1 + len(columns), + 1, + figsize=figsize or (10.0, height), + sharex=True, + height_ratios=[max(2.0, 0.5 * len(lanes))] + [1] * len(columns), + squeeze=False, + layout="constrained", + ) + all_axes = all_axes[:, 0] + ax, panels = all_axes[0], all_axes[1:] + else: + figure, panels = None, [] + pad = 0.02 * (limits[1] - limits[0]) + plot_lanes( + frame, + ax=ax, + lane="lane", + value="value", + label="label", + lanes=lanes, + color=_lane_colors(color), + vocabulary=vocabulary, + max_labels=max_labels, + x_limits=(limits[0] - pad, limits[1] + pad), + x_label="" if len(panels) else "Optical distance [m]", + colorbar_axes=[ax, *panels] if len(panels) else None, + ) + named = [address, *([chosen.name] if chosen.name else []), _epoch_label(chosen)] + ax.set_title(" · ".join(named), loc="left", fontsize="medium") + distances = _sample_distances(chosen, limits[0], limits[1], n_samples) + for index, (panel, name) in enumerate(zip(panels, columns, strict=True)): + values = chosen.column_at(name, distances) + panel.plot(distances, values, color=plt.get_cmap(_lanes.LANE_CMAP)(index % 10)) + units = _column_units(chosen, name) + panel.set_ylabel(f"{name} [{units}]" if units else name) + panel.grid(color="0.9", linewidth=0.5) + panel.set_axisbelow(True) + for side in ("top", "right"): + panel.spines[side].set_visible(False) + if figure is not None and len(panels): + panels[-1].set_xlabel("Optical distance [m]") + figure.align_ylabels() + if show: + plt.show() + return ax + + +def _time_window(asked): + """Resolve a (start, end) time selection to matplotlib dates.""" + if asked is None: + return None, None + try: + low, high = asked + except (TypeError, ValueError): + msg = f"time={asked!r} must be a (start, end) pair." + raise ParameterError(msg) from None + + def one(value): + if value is None or value is ...: + return None + try: + stamp = pd.Timestamp(value) + except (ValueError, TypeError) as error: + msg = f"time={asked!r} states a bound which is not a time." + raise ParameterError(msg) from error + if pd.isnull(stamp): + msg = f"time={asked!r} states a bound which is not a time." + raise ParameterError(msg) + return mdates.date2num(stamp.to_pydatetime()) + + low, high = one(low), one(high) + if low is not None and high is not None and high <= low: + msg = f"time={asked!r} must be increasing." + raise ParameterError(msg) + return low, high + + +def _lane_colors(color): + """Pin the tracks whose vocabulary is closed, honoring an override.""" + if color is not None: + return color + return {"components": COMPONENT_COLORS} + + +def map_path( + inventory: Inventory, + optical_path: str | OpticalPath | None = None, + *, + acquisition_key: str | None = None, + time: timeable_types | None = None, + x: str | None = None, + y: str | None = None, + color: str = "distance", + n_samples: int = 1000, + cmap: str = "viridis", + linewidth: float = 2.5, + aspect: str | float | None = None, + ax: plt.Axes | None = None, + legend: bool = True, + show: bool = False, +) -> plt.Axes: + """ + Plot where an inventory's fiber physically goes. + + The polyline is the path's geometry read through the inventory's + coordinate reference system. Stretches which state no position are + left out rather than bridged, so a slack coil or an unsurveyed run + reads as the gap it is. + + With no path named this draws every path which places itself, since + a map of one cable in an inventory of several is a strange default. + + Parameters + ---------- + inventory + The inventory to draw. + optical_path + A path address, name, or object; None draws all of them. + acquisition_key + Resolve one path from an acquisition key instead. + time + The instant to resolve at. + x, y + The CRS axes to draw, defaulting to the first two the CRS + declares. A borehole needs that default overridden: a hole runs + straight down, so a plan view collapses it to a point. + color + "distance", a geometry column, a label group, or "coupling". + n_samples + How finely the path is sampled. + cmap + Colormap for a continuous coloring. + linewidth + Width of the drawn fiber. + aspect + Axes aspect; None picks equal when both axes share units. + ax + An Axes to draw on. + legend + Whether to draw the legend or colorbar. + show + Whether to call plt.show. + + Examples + -------- + >>> import dascore as dc + >>> from dascore.viz.inventory import map_path + >>> + >>> inventory = dc.get_example_inventory("tunnel") + >>> _ = map_path(inventory, x="x", y="z", color="section") + """ + crs = inventory.coordinate_reference_system + labels = list(crs.coordinate_labels) + x = x if x is not None else labels[0] + y = y if y is not None else (labels[1] if len(labels) > 1 else labels[0]) + if x == y: + msg = f"x and y are both {x!r}; a map needs two different axes." + raise ParameterError(msg) + for name in (x, y): + try: + crs.axis_index(name) + except Exception as error: # the CRS explains itself better than we can + msg = ( + f"{name!r} is not an axis of this inventory's CRS, whose axes " + f"are {tuple(labels)}. A column which is not an axis is " + "drawn by path(), not on a map." + ) + raise ParameterError(msg) from error + if optical_path is None and acquisition_key is None: + chosen = [ + (a, arr, p) + for a, net, arr, p in _iter_paths(inventory) + if _effective_at(time, net, arr, p) + ] + if not chosen: + msg = f"No optical path in this inventory is effective at {time}." + raise ParameterError(msg) + else: + chosen = [_select_path(inventory, optical_path, acquisition_key, time)] + own_figure = ax is None + ax = _get_ax(ax) + x_axis, y_axis = crs.axis_index(x), crs.axis_index(y) + handles: dict = {} + palette: dict = {} + if color != "distance": + # Checked over every path drawn: one path saying nothing about a + # group is fiber with no value, which the map already draws. + stated = {"coupling"} + for _, _, one in chosen: + stated |= set(one.geometry_columns()) + stated |= {x.group for x in one.labels} + if color not in stated: + msg = ( + f"color={color!r} names neither optical distance, a geometry " + f"column, a label group, nor 'coupling'. This inventory " + f"states {tuple(sorted(stated))}." + ) + raise ParameterError(msg) + # Two passes: every path is measured before any is drawn, so that one + # color scale spans them all rather than the last one drawn winning. + pieces = [] + for address, _, one in chosen: + distances = _sample_distances( + one, one.start_distance, one.end_distance, n_samples + ) + coords = one.coordinates_at(distances, crs) + points = np.column_stack([coords[:, x_axis], coords[:, y_axis]]) + segments = np.stack([points[:-1], points[1:]], axis=1) + mid = 0.5 * (distances[:-1] + distances[1:]) + # A segment touching an unplaced sample is not fiber we can draw. + good = ~np.isnan(segments).any(axis=(1, 2)) + if not good.any(): + continue + values, colors = _segment_colors(one, color, mid[good], crs, handles, palette) + pieces.append((segments[good], values, colors)) + drawn = len(pieces) + scalar = None + # A value nothing states is not fiber nothing placed. Left as NaN it + # would map to a transparent color and the cable would simply vanish. + unstated = any( + values is not None and bool(np.isnan(values).any()) for _, values, _ in pieces + ) + if unstated: + handles.setdefault("n/a", PatchArtist(facecolor=UNPLACED, label="n/a")) + if drawn: + finite = _shown_values(pieces) + stated = [v[np.isfinite(v)] for _, v, _ in pieces if v is not None] + stated = [v for v in stated if len(v)] + norm, scale, ticks, beyond = None, None, None, "neither" + if finite: + # One scale for every path, stepped where the values are a + # handful of numbered categories rather than a quantity. + scale, norm, ticks = _lanes.numeric_scale(np.concatenate(finite), cmap) + scale = scale.with_extremes(bad=UNPLACED) + whole = np.concatenate(stated) + under = bool(whole.min() < norm.vmin) + over = bool(whole.max() > norm.vmax) + beyond = ( + ("both" if under else "max") + if over + else ("min" if under else "neither") + ) + for segments, values, colors in pieces: + collection = LineCollection( + list(segments), + linewidths=linewidth, + colors=colors, + cmap=scale if values is not None else None, + norm=norm if values is not None else None, + capstyle="round", + ) + if values is not None: + collection.set_array(values) + scalar = collection + ax.add_collection(collection) + ax.autoscale_view() + if not drawn: + msg = ( + "No optical path in this inventory places itself in the CRS, so " + "there is no layout to draw. A path is placed by a geometry " + f"segment stating the CRS's axes {tuple(labels)}." + ) + raise ParameterError(msg) + ax.set_xlabel(_axis_label(crs, x)) + ax.set_ylabel(_axis_label(crs, y)) + if aspect is None: + same = crs.units[x_axis] == crs.units[y_axis] + aspect = "equal" if same and "degree" not in crs.units[x_axis] else "auto" + ax.set_aspect(aspect) + shrink = 1.0 + if aspect == "equal" and own_figure: + # An equal aspect on a long thin cable draws a short strip in a + # tall figure, so give the figure the data's shape and the + # colorbar the strip's height rather than the figure's. + low_x, high_x = ax.get_xlim() + low_y, high_y = ax.get_ylim() + figure = ax.get_figure() + ratio = abs(high_y - low_y) / (abs(high_x - low_x) or 1.0) + drawn = figure.get_figwidth() * ratio + figure.set_figheight(float(np.clip(drawn + 1.2, 1.6, 9.0))) + shrink = float(np.clip(drawn / figure.get_figheight(), 0.25, 1.0)) + ax.grid(color="0.9", linewidth=0.5) + ax.set_axisbelow(True) + if legend and scalar is not None: + # A tall label beside a short strip of axes clips; lay the bar out + # the way the data is laid out instead. + flat = shrink < 0.45 + bar = ax.get_figure().colorbar( + scalar, + ax=ax, + location="bottom" if flat else "right", + fraction=0.12 if flat else 0.05, + pad=0.25 if flat else 0.02, + aspect=45 if flat else 20, + shrink=1.0 if flat else shrink, + # An arrow where fiber is drawn past the end of the scale. + extend=beyond, + ) + bar.set_label("Optical distance [m]" if color == "distance" else color) + if ticks is not None: + # A stepped scale reads at its steps; half a borehole is not one. + bar.set_ticks(list(ticks)) + if legend and handles: + # A colorbar already occupies the strip beside the axes. + offset = 1.12 if scalar is not None and shrink >= 0.45 else 1.01 + ax.legend( + handles=list(handles.values()), + loc="upper left", + bbox_to_anchor=(offset, 1.0), + frameon=False, + fontsize="small", + # The colorbar beside it already carries the name. + title=None if scalar is not None else color, + ) + if show: + plt.show() + return ax + + +def _shown_values(pieces) -> list: + """The colored values of segments this projection actually shows. + + A borehole seen from above is a point: it is drawn, but it displays + no length, and letting it into the scale spends most of the colormap + on fiber the reader cannot see. + """ + stated = [v[np.isfinite(v)] for _, v, _ in pieces if v is not None] + stated = [v for v in stated if len(v)] + if not stated: + return [] + corners = np.concatenate([x.reshape(-1, 2) for x, _, _ in pieces]) + floor = float(max(np.ptp(corners[:, 0]), np.ptp(corners[:, 1]))) * 1e-3 + shown = [] + for segments, values, _ in pieces: + if values is None: + continue + steps = segments[:, 1] - segments[:, 0] + drawn = np.hypot(steps[:, 0], steps[:, 1]) + keep = values[np.isfinite(values) & (drawn > floor)] + if len(keep): + shown.append(keep) + # Every segment collapsed, so the projection shows no lengths at all. + return shown or stated + + +def _axis_label(crs, name) -> str: + """Label a map axis with the CRS's name for it and its units.""" + index = crs.axis_index(name) + units = crs.units[index] if index < len(crs.units) else "" + return f"{name} [{units}]" if units else str(name) + + +UNPLACED = mcolors.to_rgba(UNCOVERED_COLOR) + + +def _segment_colors(one, color, mid, crs, handles, palette): + """Return (values, colors) for one path's segments; one of them is None.""" + if color == "distance": + return mid, None + if color in one.geometry_columns(): + return one.column_at(color, mid), None + if color == "coupling": + items = list(one.coupling) + keys = [x.coupling_type for x in items] + else: + items = [x for x in one.labels if x.group == color] + keys = [x.value for x in items] + if not items: + # This path states nothing under that name; another one does. + handles.setdefault("n/a", PatchArtist(facecolor=UNPLACED, label="n/a")) + return None, [UNPLACED] * len(mid) + masks = interval_masks(mid, [x.interval for x in items]) + kinds = {value_kind(normalize_value(k)) for k in keys} + if kinds == {"numeric"}: + values = np.full(len(mid), np.nan) + for item, mask in zip(items, masks, strict=True): + values[mask] = float(normalize_value(item.value)) + return values, None + # The palette is the figure's, not this path's, so one value is one + # color however many paths are drawn and whatever order they state it. + wheel = plt.get_cmap(_lanes.STRING_CMAP) + order = _lanes.WHEEL_ORDER + for key in dict.fromkeys(map(str, keys)): + palette.setdefault(key, wheel(order[len(palette) % len(order)])) + seen = palette + colors = [UNPLACED] * len(mid) + for key, mask in zip(keys, masks, strict=True): + placed = np.flatnonzero(mask) + if not len(placed): + # Stating a value over fiber which has no position places + # nothing, so it earns no entry in the legend. + continue + for position in placed: + colors[position] = seen[str(key)] + handles.setdefault( + str(key), PatchArtist(facecolor=seen[str(key)], label=str(key)) + ) + if any(c is UNPLACED for c in colors): + handles.setdefault("n/a", PatchArtist(facecolor=UNPLACED, label="n/a")) + return None, colors + + +def timeline( + inventory: Inventory, + *, + kind: str = "both", + color: str = "interrogator", + time: tuple | None = None, + ax: plt.Axes | None = None, + legend: bool = True, + show: bool = False, +) -> plt.Axes: + """ + Plot when each part of an inventory was valid. + + One lane per acquisition and per optical path lineage, drawn against + time. An epoch which states no start or no end is unbounded rather + than missing, and is drawn running off that side of the axis. + + Parameters + ---------- + inventory + The inventory to draw. + kind + "both", "acquisition", or "optical_path". + color + "interrogator", "data_type", or "kind". + time + The times to draw between, as (start, end). Either end may be + None, or ..., to run to what the epochs themselves state. + ax + An Axes to draw on. + legend + Whether to draw the legend. + show + Whether to call plt.show. + + Examples + -------- + >>> import dascore as dc + >>> from dascore.viz.inventory import timeline + >>> + >>> inventory = dc.get_example_inventory("tunnel") + >>> _ = timeline(inventory) + """ + if kind not in {"both", "acquisition", "optical_path"}: + msg = ( + f"kind={kind!r} is not a timeline selection; the options are " + "('both', 'acquisition', 'optical_path')." + ) + raise ParameterError(msg) + if color not in {"interrogator", "data_type", "kind"}: + msg = ( + f"color={color!r} is not a timeline coloring; the options are " + "('interrogator', 'data_type', 'kind')." + ) + raise ParameterError(msg) + rows = [] + for network in inventory.networks: + for array in network.fiber_arrays: + if kind in {"both", "optical_path"}: + for one in array.optical_paths: + start, end = _effective_epoch(network, array, one) + rows.append( + { + "lane": f"{network.code}.{array.code}." + f"{one.location_code} [path]", + "start": start, + "end": end, + "value": "optical path", + "label": one.name, + } + ) + if kind in {"both", "acquisition"}: + for acquisition in array.acquisitions: + start, end = _effective_epoch(network, array, acquisition) + rows.append( + { + "lane": f"{network.code}.{array.code}." + f"{acquisition.location_code}.{acquisition.code}", + "start": start, + "end": end, + "value": _acquisition_color_value( + inventory, acquisition, color + ), + "label": "", + } + ) + if not rows: + msg = ( + "This inventory holds nothing with a time epoch, so there is no " + "timeline to draw." + ) + raise ParameterError(msg) + frame = pd.DataFrame(rows) + known = pd.concat([frame["start"], frame["end"]]).dropna() + asked_low, asked_high = _time_window(time) + if ax is None: + lanes = len(dict.fromkeys(frame["lane"])) + _, ax = plt.subplots( + 1, figsize=(9.0, min(1.0 + 0.55 * lanes, 14.0)), layout="constrained" + ) + + if asked_low is not None or asked_high is not None: + # A month is an arbitrary width, and only reached where one end is + # asked for and nothing states a time to take the other from. + month, stated = ( + 30.0, + [ + mdates.date2num(pd.Timestamp(x).to_pydatetime()) + for x in (known.min(), known.max()) + ] + if len(known) + else [None, None], + ) + low = asked_low if asked_low is not None else stated[0] + high = asked_high if asked_high is not None else stated[1] + if low is None: + low = high - month + if high is None or high <= low: + high = low + month + dated = True + elif len(known): + low = mdates.date2num(pd.Timestamp(known.min()).to_pydatetime()) + high = mdates.date2num(pd.Timestamp(known.max()).to_pydatetime()) + pad = (high - low) * 0.05 or 30.0 + low, high = low - pad, high + pad + dated = True + else: + # Nothing states a time, which is legal and common. Drawing bars on + # a fabricated axis would invite the lengths to be read as facts. + low, high, dated = 0.0, 1.0, False + frame["open_start"] = frame["start"].isna() + frame["open_end"] = frame["end"].isna() + frame["start"] = [ + low if pd.isnull(x) else mdates.date2num(pd.Timestamp(x).to_pydatetime()) + for x in frame["start"] + ] + frame["end"] = [ + high if pd.isnull(x) else mdates.date2num(pd.Timestamp(x).to_pydatetime()) + for x in frame["end"] + ] + # An epoch outside the window is left out rather than clipped to a + # sliver at the edge, which would read as an epoch which ended there. + frame = frame[(frame["start"] < high) & (frame["end"] > low)] + if frame.empty: + msg = f"No epoch in this inventory falls within time={time!r}." + raise ParameterError(msg) + plot_lanes( + frame, + ax=ax, + lane="lane", + value="value", + label="label", + color=None, + x_limits=(low, high), + legend=legend, + ) + if dated: + _format_time_axis(ax, "time", "x") + ax.set_xlabel("Time") + else: + ax.set_xticks([]) + ax.set_xlabel("time (no epoch in this inventory states one)") + if show: + plt.show() + return ax + + +def _acquisition_color_value(inventory, acquisition, color) -> str: + """The string an acquisition is colored by.""" + if color == "kind": + return "acquisition" + if color == "data_type": + return acquisition.data_type or "unstated" + interrogator = acquisition.interrogator + if isinstance(interrogator, str): + interrogator = inventory.get_resource(interrogator) + if interrogator is None: + return "no interrogator" + name = f"{interrogator.manufacturer} {interrogator.model}".strip() + return name or interrogator.serial_number or "interrogator" diff --git a/docs/recipes/tunnel_inventory.qmd b/docs/recipes/tunnel_inventory.qmd index 06d9a3efc..2c4b97b6f 100644 --- a/docs/recipes/tunnel_inventory.qmd +++ b/docs/recipes/tunnel_inventory.qmd @@ -202,6 +202,22 @@ print("coords:", [x for x in names.coords if "." not in x]) `section` and `borehole` are among the coordinates because the labels table named them. `x`, `y`, and `z` are there because the CRS declares those axes and the geometry table resolves to them. +# Seeing it + +The inventory can now draw the deployment it describes. Along the fiber, past the lead-in, every track lines up against optical distance: + +```{python} +inventory.viz.path(distance=(1495, 1780), show=True); +``` + +And in the tunnel's own survey grid, looking side-on — the second of the two views the [hand-drawn figure](#what-the-drawing-says) pairs: + +```{python} +inventory.viz.map(x="x", y="z", color="section", show=True); +``` + +The three holes go to 20 m, the trench runs along the floor, and the break in it is the slack coil: ten meters of fiber wound into a one-meter loop, which the geometry table deliberately leaves unplaced. + # What the data gets out of it A patch recorded here carries the acquisition key, and that key with the time the patch covers is the whole of the join. @@ -291,6 +307,14 @@ for epoch in repaired.networks[0].fiber_arrays[0].optical_paths: print(f"from {began}: {epoch.optical_length:.1f} m") ``` +The two epochs are what the timeline draws, since it is the plot which looks along time: + +```{python} +repaired.viz.timeline(show=True); +``` + +The path lane splits at the repair. The acquisition's does not: the interrogator was never reconfigured, so its epoch runs unbounded in both directions and is hatched at each end. + A patch recorded across midnight on the first of September was recorded through both. [`Spool.conform_to_inventory`](`dascore.core.spool.Spool.conform_to_inventory`) is the step which insists every patch be describable by exactly one entry, and it subdivides that patch rather than choosing for you: ```{python} diff --git a/docs/tutorial/inventory.qmd b/docs/tutorial/inventory.qmd index 2e985f148..fdd279960 100644 --- a/docs/tutorial/inventory.qmd +++ b/docs/tutorial/inventory.qmd @@ -182,6 +182,30 @@ assert "gauge_length" in names.attrs Listing a name is not promising a value for it. This example's geometry states chainage and no position, so the spatial names resolve to nothing until some segment states them. +# Seeing an inventory + +An inventory draws itself through its `viz` namespace, and each of the three plots looks along one coordinate. + +[`Inventory.viz.path`](`dascore.viz.inventory.path`) looks along **optical distance**, which is the coordinate the data is in. Every track the path describes becomes a lane, so the channels an acquisition places, the components, the coupling, and each label group line up against the same axis: + +```{python} +inventory.viz.path(show=True); +``` + +A geometry column is a curve rather than a set of intervals, so it is drawn as a panel beneath. Here the chainage stands still through the slack coil while ten meters of fiber goes by: + +```{python} +inventory.viz.path(columns="chainage", show=True); +``` + +`tracks=` picks the lanes, in the order given, when the whole picture is more than the question needs: + +```{python} +inventory.viz.path(tracks=("coupling", "zone"), show=True); +``` + +The other two plots need something this small example does not have. [`Inventory.viz.map`](`dascore.viz.inventory.map_path`) looks along **space**, and needs geometry which states the CRS's axes; this path states chainage and no position. [`Inventory.viz.timeline`](`dascore.viz.inventory.timeline`) looks along **time**, and is worth reading where the epochs are real. The [tunnel recipe](../recipes/tunnel_inventory.qmd) has both. + # Attaching an inventory to a spool [`Spool.attach_inventory`](`dascore.core.spool.Spool.attach_inventory`) carries an inventory on a spool, and touches no data. No patch gains a field, no row moves, `len` does not change. It costs nothing per patch, which is what makes it safe to do early and decide later what to use it for. Attaching a *different* inventory does clear any enrichment set up from the old one, since applying the old instructions to new metadata would rewrite every patch behind your back. diff --git a/docs/tutorial/visualization.qmd b/docs/tutorial/visualization.qmd index d320ea6b9..cfdc947a1 100644 --- a/docs/tutorial/visualization.qmd +++ b/docs/tutorial/visualization.qmd @@ -4,9 +4,10 @@ execute: warning: false --- -# Viz +The [viz module](`dascore.viz`) holds DASCore's plots, and they hang off the object they are of: `Patch.viz` for data, `Inventory.viz` for the observing system which recorded it. + +# Patch The following provides some examples of patch visualization. -See the [viz module documentation](`dascore.viz`) for a list of visualization functions ## Waterfall The [`waterfall patch function`](`dascore.viz.waterfall`) creates a waterfall plot of the patch data. @@ -72,3 +73,49 @@ patch = dc.examples.get_example_patch( ) patch.viz.wiggle(show=True); ``` + +# Inventory + +An [inventory](inventory.qmd) plots what it knows about the fiber, with no data present. The examples below use the tunnel deployment the [tunnel recipe](../recipes/tunnel_inventory.qmd) builds. + +```{python} +import dascore as dc + +inventory = dc.get_example_inventory("tunnel") +``` + +## Path + +The [`path plot`](`dascore.viz.inventory.path`) draws every track along the fiber against optical distance. `distance=` is the window to draw, in the same coordinate; this deployment starts with 1.5 km of telemetry lead-in, which would otherwise crush the instrumented part into a corner. + +```{python} +inventory.viz.path(time="2024-07-01", distance=(1495, 1780), show=True); +``` + +The boreholes label themselves 3, 2, 1 — the fiber works back through them — and the splices are ticks rather than zero-width boxes. + +## Map + +The [`map plot`](`dascore.viz.inventory.map_path`) draws where the fiber physically goes, in two axes of the inventory's coordinate reference system. `x` and `y` choose them: a borehole runs straight down, so the default plan view collapses it to a point. + +```{python} +inventory.viz.map(x="x", y="z", color="section", time="2024-07-01", show=True); +``` + +The break near the middle of the trench is the slack coil, which nobody surveyed. Unplaced fiber is left out rather than bridged, since a made-up polyline would be worse than a gap. + +`color=` also takes a geometry column, a label group, or `"coupling"`. Its default is optical distance, which is how a distance read off a waterfall is found on the ground: + +```{python} +inventory.viz.map(x="x", y="z", time="2024-07-01", show=True); +``` + +## Timeline + +The [`timeline plot`](`dascore.viz.inventory.timeline`) draws when each acquisition and each optical path was valid. An epoch which states no start or no end is unbounded, and runs off that side of the axis hatched. + +```{python} +inventory.viz.timeline(show=True); +``` + +The path lane splits on the first of September, which is the day the trench cable was repaired. diff --git a/pyproject.toml b/pyproject.toml index 2225f6052..9b14b28ee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -196,6 +196,7 @@ viz = "dascore.viz:VizPatchNameSpace" [project.entry-points."dascore.inventory_namespace"] io = "dascore.io:InventoryIO" +viz = "dascore.viz:VizInventoryNameSpace" [project.entry-points."dascore.annotation_namespace"] io = "dascore.io:AnnotationIO" diff --git a/tests/test_viz/test_inventory_viz.py b/tests/test_viz/test_inventory_viz.py new file mode 100644 index 000000000..1f2e50572 --- /dev/null +++ b/tests/test_viz/test_inventory_viz.py @@ -0,0 +1,1193 @@ +"""Tests for the plots an inventory draws of itself.""" + +from __future__ import annotations + +from pathlib import Path + +import matplotlib.pyplot as plt +import numpy as np +import pytest +from matplotlib.collections import LineCollection, PatchCollection + +import dascore as dc +from dascore.core import inventory as inv +from dascore.exceptions import ParameterError +from dascore.viz import VizInventoryNameSpace +from dascore.viz.inventory import ( + COMPONENT_COLORS, + _distance_window, + map_path, + path, + timeline, +) + + +def _lanes(ax): + """The lane names an axes shows, top to bottom.""" + return [x.get_text() for x in ax.get_yticklabels()] + + +def _boxes(ax): + """The patch collections on an axes.""" + return [x for x in ax.collections if isinstance(x, PatchCollection)] + + +def _bar_label(ax) -> str: + """The label of the colorbar beside an axes, however it is laid out.""" + bar = ax.get_figure().axes[-1] + return bar.get_ylabel() or bar.get_xlabel() + + +def _legend_labels(ax): + """The legend entries, or an empty list with no legend.""" + legend = ax.get_legend() + return [] if legend is None else [x.get_text() for x in legend.get_texts()] + + +def _main_path(epoch: int) -> inv.OpticalPath: + """One epoch of the surveyed path; the second is the repaired fiber.""" + run = 400.0 if epoch == 1 else 402.0 + times = {"end_time": "2026-07-01"} if epoch == 1 else {"start_time": "2026-07-01"} + return inv.OpticalPath( + name="main", + location_code="00", + optical_components=( + inv.FiberSegment(name="lead", optical_length=100.0), + inv.Connector(name="patch"), + inv.FiberSegment(name="run", optical_length=run), + inv.Terminator(name="end"), + ), + geometry=( + # Two surveyed runs with an unsurveyed gap from 300 to 350. + inv.Geometry( + name="west", + distance=(100.0, 300.0), + coordinates={"x": (0.0, 200.0), "y": (0.0, 0.0), "z": (0.0, -1.0)}, + ), + inv.Geometry( + name="east", + distance=(350.0, 500.0), + coordinates={"x": (250.0, 400.0), "y": (0.0, 5.0), "z": (-1.0, 0.0)}, + ), + # Columns of those same stretches, so they share the runs' names. + inv.Geometry( + name="west", + distance=(100.0, 300.0), + coordinates={"chainage": (0.0, 200.0), "depth": (0.5, 1.5)}, + units={"chainage": "m"}, + ), + inv.Geometry( + name="east", + distance=(350.0, 500.0), + coordinates={"chainage": (250.0, 400.0)}, + units={"chainage": "m"}, + ), + ), + coupling=( + inv.CouplingCondition( + start_distance=100.0, end_distance=300.0, coupling_type="trench" + ), + inv.CouplingCondition( + start_distance=350.0, end_distance=500.0, coupling_type="conduit" + ), + ), + labels=( + inv.OpticalPathLabel( + start_distance=100.0, end_distance=200.0, group="zone", value="north" + ), + inv.OpticalPathLabel( + start_distance=200.0, end_distance=400.0, group="zone", value="south" + ), + inv.OpticalPathLabel( + start_distance=150.0, end_distance=300.0, group="noisy", value=True + ), + inv.OpticalPathLabel( + start_distance=300.0, end_distance=400.0, group="noisy", value=False + ), + inv.OpticalPathLabel( + start_distance=100.0, end_distance=200.0, group="count", value=0 + ), + inv.OpticalPathLabel( + start_distance=200.0, end_distance=300.0, group="count", value=2.5 + ), + ), + **times, + ) + + +def build_site_inventory() -> inv.Inventory: + """An inventory with two path epochs, a bare spur, and varied acquisitions.""" + spur = inv.OpticalPath( + name="spur", + location_code="01", + optical_components=(inv.FiberSegment(name="spur", optical_length=50.0),), + ) + common = dict(data_category="DAS", sample_rate=100.0, gauge_length=10.0) + acquisitions = ( + inv.Acquisition( + code="RAW", + location_code="00", + start_time="2026-06-01", + end_time="2026-06-15", + data_type="strain_rate", + spatial_interval=1.0, + interrogator=inv.Interrogator(manufacturer="Fake", model="FI-1"), + distance_map=inv.DistanceMap(channel=(0.0, 300.0), distance=(100.0, 400.0)), + **common, + ), + inv.Acquisition( + code="RAW", + location_code="00", + start_time="2026-07-01", + spatial_interval=1.0, + interrogator=inv.Interrogator(serial_number="sn-9"), + # One point states an origin but no extent, so it draws as a tick. + distance_map=inv.DistanceMap(channel=(0.0,), distance=(100.0,)), + **common, + ), + inv.Acquisition(code="AUX", location_code="01", interrogator="int-1", **common), + inv.Acquisition(code="NIL", location_code="02", **common), + ) + array = inv.FiberArray( + code="L1", + acquisitions=acquisitions, + optical_paths=(_main_path(1), _main_path(2), spur), + ) + return inv.Inventory( + coordinate_reference_system=inv.CoordinateReferenceSystem( + authority="", + code="", + name="site grid", + coordinate_labels=("x", "y", "z"), + units=("meter", "meter", "meter"), + ), + resources=[inv.Interrogator(resource_id="int-1")], + networks=(inv.Network(code="DAS", fiber_arrays=(array,)),), + ).check() + + +@pytest.fixture(scope="module") +def site(): + """The inventory most tests draw.""" + return build_site_inventory() + + +@pytest.fixture(scope="module") +def tunnel(): + """The tunnel example, which has real epochs and a surveyed coil gap.""" + return dc.get_example_inventory("tunnel") + + +class TestNamespace: + """The plots hang off inventory.viz.""" + + def test_registered(self, tunnel): + """Inventory.viz is the viz namespace, with the three verbs.""" + assert isinstance(tunnel.viz, VizInventoryNameSpace) + assert tunnel.viz.path.__name__ == "path" + assert tunnel.viz.map.__name__ == "map_path" + assert tunnel.viz.timeline.__name__ == "timeline" + + def test_declared_as_an_entry_point(self): + """An install must carry the namespace, not just an import of it. + + Importing dascore.viz registers the namespace as a side effect, so + every other test here would pass with the entry point deleted. + """ + text = (Path(dc.__file__).parent.parent / "pyproject.toml").read_text() + block = text.split('[project.entry-points."dascore.inventory_namespace"]')[1] + block = block.split("[")[0] + assert 'viz = "dascore.viz:VizInventoryNameSpace"' in block + + def test_namespace_call(self, tunnel): + """Calling through the namespace passes the inventory.""" + ax = tunnel.viz.timeline() + assert len(_lanes(ax)) == 2 + + +class TestSelectPath: + """Naming the path a plot is of.""" + + def test_no_paths(self): + """An inventory without paths has nothing to plot.""" + empty = inv.Inventory( + networks=( + inv.Network(code="DAS", fiber_arrays=(inv.FiberArray(code="A"),)), + ) + ) + with pytest.raises(ParameterError, match="holds no optical paths"): + path(empty) + + def test_ambiguous(self, site): + """Several candidates demand a choice, and are listed.""" + with pytest.raises(ParameterError, match="holds 3 optical paths") as info: + path(site) + assert "DAS.L1.00 (from the beginning)" in str(info.value) + assert "DAS.L1.00 (from 2026-07-01)" in str(info.value) + + def test_address_and_time(self, site): + """An address plus a time picks one epoch.""" + ax = path(site, "DAS.L1.00", time="2026-08-01") + assert ax.get_title("left").endswith("from 2026-07-01") + ax = path(site, "DAS.L1.00", time="2026-06-10") + assert ax.get_title("left").endswith("from the beginning") + + def test_name(self, site): + """A path's name works where it is unique.""" + ax = path(site, "spur") + assert ax.get_title("left").startswith("DAS.L1.01") + + def test_unknown_name(self, site): + """An unknown name lists the addresses.""" + with pytest.raises(ParameterError, match="No optical path matches 'nope'"): + path(site, "nope") + + def test_object(self, site): + """The path object itself is accepted, and a foreign one refused.""" + spur = site.networks[0].fiber_arrays[0].optical_paths[2] + assert path(site, spur).get_title("left").startswith("DAS.L1.01") + foreign = spur.model_copy() + with pytest.raises(ParameterError, match="not part of this inventory"): + path(site, foreign) + + def test_epochs_need_a_time(self, site): + """An address names a path, so it cannot pick among its epochs.""" + with pytest.raises(ParameterError, match="has 2 epochs") as info: + path(site, "DAS.L1.00") + assert "Pass a time" in str(info.value) + + def test_ambiguous_acquisition_key(self, site): + """A key naming two acquisition epochs asks for a time, in our terms.""" + with pytest.raises(ParameterError, match="does not resolve") as info: + path(site, acquisition_key="DAS.L1.00.RAW") + assert "2 acquisitions" in str(info.value) + assert "Pass a time" in str(info.value) + + def test_unknown_acquisition_key_keeps_its_error(self, site): + """A key which no time can fix reports what actually went wrong.""" + with pytest.raises(ParameterError, match="does not resolve") as info: + path(site, acquisition_key="DAS.L1.00.NOPE") + # The count resolve reported, not a story about epochs. + assert "0 acquisitions" in str(info.value) + + def test_containers_decide_which_epoch(self): + """A path stating no time is effective when its containers are.""" + + def build(code, **times): + one = inv.OpticalPath( + name="main", + location_code="00", + optical_components=(inv.FiberSegment(name="f", optical_length=100.0),), + ) + array = inv.FiberArray(code="L1", optical_paths=(one,), **times) + return inv.Network(code=code, fiber_arrays=(array,), **times) + + inventory = inv.Inventory( + networks=( + build("AA", start_time="2020-01-01", end_time="2021-01-01"), + build("BB", start_time="2021-01-01"), + ) + ).check() + # Neither path states a bound, so only their containers can tell + # them apart; asking the path alone would call both effective. + assert path(inventory, time="2020-06-01").get_title("left").startswith("AA") + assert path(inventory, time="2022-06-01").get_title("left").startswith("BB") + lanes = timeline(inventory, kind="optical_path") + assert lanes.get_xlabel() == "Time" + boxes = _boxes(lanes)[0].get_paths() + assert len(boxes) == 1 + + def test_no_path_effective_then(self): + """A time nothing is effective at says so, rather than listing paths.""" + one = inv.OpticalPath( + name="main", + location_code="00", + start_time="2020-01-01", + end_time="2021-01-01", + optical_components=(inv.FiberSegment(name="f", optical_length=100.0),), + ) + array = inv.FiberArray(code="L1", optical_paths=(one,)) + inventory = inv.Inventory( + networks=(inv.Network(code="N", fiber_arrays=(array,)),) + ).check() + with pytest.raises(ParameterError, match="is effective at"): + path(inventory, time="1999-01-01") + + def test_acquisition_key(self, site): + """An acquisition key resolves through the inventory.""" + ax = path(site, acquisition_key="DAS.L1.00.RAW", time="2026-06-10") + assert ax.get_title("left").startswith("DAS.L1.00") + + def test_acquisition_key_without_path(self, site): + """An acquisition on a location with no path cannot be drawn.""" + with pytest.raises(ParameterError, match="resolves to no optical path"): + path(site, acquisition_key="DAS.L1.02.NIL") + + +class TestPath: + """The tracks along one path.""" + + def test_all_tracks(self, site): + """Every track becomes a lane, channels first.""" + ax = path(site, "DAS.L1.00", time="2026-06-10") + assert _lanes(ax) == [ + "channels (RAW)", + "components", + "coupling", + "zone", + "noisy", + "count", + ] + assert ax.get_xlabel() == "Optical distance [m]" + # Components take their fixed colors, so the legend names the types. + assert "FiberSegment" in _legend_labels(ax) + + def test_tracks_selected_in_order(self, site): + """tracks= picks lanes and orders them.""" + ax = path(site, "DAS.L1.00", time="2026-06-10", tracks=("zone", "coupling")) + assert _lanes(ax) == ["zone", "coupling"] + ax = path(site, "DAS.L1.00", time="2026-06-10", tracks="channels") + assert _lanes(ax) == ["channels (RAW)"] + + def test_unknown_track(self, site): + """A track which is not a track nor a label group is refused.""" + with pytest.raises(ParameterError, match="'nope' is not a track"): + path(site, "DAS.L1.00", time="2026-06-10", tracks="nope") + + def test_tracks_with_nothing(self, site): + """Asking for a lane the path has no rows for is an error.""" + with pytest.raises(ParameterError, match="nothing to draw for tracks"): + path(site, "spur", tracks="channels") + + def test_acquisition_not_effective(self, site): + """An acquisition which overlaps the path but not the time is left out.""" + ax = path(site, "DAS.L1.00", time="2026-06-20") + assert not any(x.startswith("channels") for x in _lanes(ax)) + + def test_point_distance_map(self, site): + """A single-point distance map draws as a tick, not a guessed span.""" + ax = path(site, "DAS.L1.00", time="2026-08-01", tracks="channels") + assert ax.lines + assert ax.lines[0].get_xdata()[0] == 100.0 + + def test_columns(self, site): + """Named columns get panels under the lanes, labelled with units.""" + ax = path(site, "DAS.L1.00", time="2026-06-10", columns=("chainage", "depth")) + panels = ax.get_figure().axes[1:] + assert [x.get_ylabel() for x in panels] == ["chainage [m]", "depth"] + assert panels[-1].get_xlabel() == "Optical distance [m]" + assert ax.get_xlabel() == "" + # Depth is stated from 100 to 300 m only, so the line breaks outside. + xs, ys = panels[1].lines[0].get_data() + inside = (xs > 100) & (xs < 300) + assert np.isfinite(ys[inside]).all() and np.isnan(ys[~inside]).all() + + def test_column_string(self, site): + """A single column name is accepted without a tuple.""" + ax = path(site, "DAS.L1.00", time="2026-06-10", columns="chainage") + assert len(ax.get_figure().axes) == 2 + + def test_position_column_refused(self, site): + """A CRS axis is drawn by the map, not as a panel.""" + with pytest.raises(ParameterError, match="position axis of the CRS"): + path(site, "DAS.L1.00", time="2026-06-10", columns="x") + + def test_unknown_column(self, site): + """A column the path does not state is refused.""" + with pytest.raises(ParameterError, match="no geometry column named 'azimuth'"): + path(site, "DAS.L1.00", time="2026-06-10", columns="azimuth") + + def test_ax_with_columns_refused(self, site): + """Panels need their own figure, so ax and columns conflict.""" + _, ax = plt.subplots() + with pytest.raises(ParameterError, match="builds the figure"): + path(site, "DAS.L1.00", time="2026-06-10", columns="chainage", ax=ax) + + def test_ax_without_columns(self, site): + """Lanes alone draw onto an axes a caller provides.""" + _, ax = plt.subplots() + out = path(site, "DAS.L1.00", time="2026-06-10", ax=ax) + assert out is ax + assert ax.get_xlabel() == "Optical distance [m]" + + def test_distance_window(self, site): + """distance=(low, high) sets the window; None runs to the end.""" + ax = path(site, "DAS.L1.00", time="2026-06-10", distance=(200, 300)) + low, high = ax.get_xlim() + assert low < 200 and high > 300 and high < 320 + ax = path(site, "DAS.L1.00", time="2026-06-10", distance=(400, None)) + assert ax.get_xlim()[1] > 500 + + @pytest.mark.parametrize( + "asked, match", + [ + (5, "must be a .low, high. pair"), + ((10, 5), "must be increasing"), + ((900, 1000), "clips everything away"), + ], + ) + def test_bad_window(self, site, asked, match): + """A window which is not a window is explained.""" + with pytest.raises(ParameterError, match=match): + path(site, "DAS.L1.00", time="2026-06-10", distance=asked) + + def test_window_ellipsis(self): + """An Ellipsis means the same as None at either end.""" + assert _distance_window((..., 10), (0.0, 20.0)) == (0.0, 10.0) + assert _distance_window((5, ...), (0.0, 20.0)) == (5.0, 20.0) + + def test_zero_length_path(self): + """A path whose components state no length has no axis.""" + stub = inv.OpticalPath( + name="stub", location_code="09", optical_components=(inv.Connector(),) + ) + array = inv.FiberArray(code="A", optical_paths=(stub,)) + inventory = inv.Inventory( + networks=(inv.Network(code="N", fiber_arrays=(array,)),) + ) + with pytest.raises(ParameterError, match="has no length"): + path(inventory) + + def test_color_override_and_figsize(self, site, monkeypatch): + """color= reaches the renderer, figsize the figure, show plt.show.""" + called = [] + monkeypatch.setattr(plt, "show", lambda: called.append(True)) + ax = path( + site, + "DAS.L1.00", + time="2026-06-10", + tracks="coupling", + color="black", + figsize=(4, 3), + show=True, + ) + assert called + assert tuple(ax.get_figure().get_size_inches()) == (4.0, 3.0) + assert np.allclose(_boxes(ax)[0].get_facecolors()[0][:3], [0, 0, 0]) + + def test_components_keep_their_own_colors(self, site): + """The component vocabulary is closed, so its colors are pinned.""" + ax = path(site, "DAS.L1.00", time="2026-06-10", tracks="components") + colors = _boxes(ax)[0].get_facecolors() + expected = plt.matplotlib.colors.to_rgba(COMPONENT_COLORS["FiberSegment"]) + assert np.allclose(colors[0], expected) + + def test_tracks_do_not_move_the_palette(self, site): + """Drawing some tracks colors them as drawing all of them does.""" + every = path(site, "DAS.L1.00", time="2026-06-10") + full = { + x.get_text(): tuple(np.round(y.get_facecolor(), 5)) + for x, y in zip( + every.get_legend().get_texts(), + every.get_legend().legend_handles, + strict=True, + ) + } + plt.close("all") + some = path(site, "DAS.L1.00", time="2026-06-10", tracks=("zone",)) + part = { + x.get_text(): tuple(np.round(y.get_facecolor(), 5)) + for x, y in zip( + some.get_legend().get_texts(), + some.get_legend().legend_handles, + strict=True, + ) + } + shared = set(full) & set(part) + assert shared + for name in shared: + assert full[name] == part[name], f"{name} changed color with tracks=" + + def test_a_refusal_leaves_no_figure(self, site): + """A window which clips everything away builds no figure to leak.""" + plt.close("all") + before = plt.get_fignums() + with pytest.raises(ParameterError): + path(site, "DAS.L1.00", time="2026-06-10", distance=(5000, 6000)) + assert plt.get_fignums() == before + + def test_columns_stay_aligned_with_the_lanes(self): + """A colorbar must not steal width from the lanes alone.""" + readings = tuple( + inv.OpticalPathLabel( + start_distance=100.0 + 10 * index, + end_distance=110.0 + 10 * index, + group="reading", + value=float(index), + ) + # Enough distinct numbers to earn a colorbar rather than labels. + for index in range(9) + ) + one = inv.OpticalPath( + name="main", + location_code="00", + optical_components=(inv.FiberSegment(name="f", optical_length=300.0),), + geometry=( + inv.Geometry( + name="run", + distance=(100.0, 300.0), + coordinates={"chainage": (0.0, 200.0)}, + units={"chainage": "m"}, + ), + ), + labels=readings, + ) + array = inv.FiberArray(code="L1", optical_paths=(one,)) + inventory = inv.Inventory( + networks=(inv.Network(code="N", fiber_arrays=(array,)),) + ).check() + ax = path(inventory, columns="chainage") + figure = ax.get_figure() + figure.draw_without_rendering() + assert len(figure.axes) == 3, "no colorbar was drawn, so nothing is tested" + panel = figure.axes[1] + assert ax.get_position().x1 == pytest.approx(panel.get_position().x1, abs=1e-6) + + def test_tunnel_epochs(self, tunnel): + """The tunnel's repair splits its path; both epochs draw.""" + before = path(tunnel, time="2024-07-01", distance=(1495, 1780)) + after = path(tunnel, time="2024-10-01", distance=(1495, 1780)) + assert before.get_title("left") != after.get_title("left") + assert _lanes(before) == _lanes(after) + + +class TestMap: + """Where the fiber is.""" + + def test_default_axes(self, site): + """The first two CRS axes are the plan view; all paths draw.""" + ax = map_path(site) + assert ax.get_xlabel() == "x [meter]" + assert ax.get_ylabel() == "y [meter]" + lines = [x for x in ax.collections if isinstance(x, LineCollection)] + # Two epochs of the main path; the spur places nothing. + assert len(lines) == 2 + assert ax.get_aspect() == 1.0 + + def test_gap_breaks_polyline(self, site): + """Unsurveyed fiber is a break in the line, never a bridge.""" + ax = map_path(site, "DAS.L1.00", time="2026-06-10", x="x", y="z") + segments = next( + x for x in ax.collections if isinstance(x, LineCollection) + ).get_segments() + xs = np.concatenate([s[:, 0] for s in segments]) + assert not ((xs > 201.0) & (xs < 249.0)).any() + # A bridge is one long segment with no interior point, so looking + # only at where samples fell would not see it. + crossing = [ + s for s in segments if s[:, 0].min() < 205.0 < 245.0 < s[:, 0].max() + ] + assert not crossing, "a segment spans fiber nobody placed" + + def test_a_short_gap_is_still_a_gap(self): + """A gap narrower than the sample spacing still breaks the line.""" + one = inv.OpticalPath( + name="long", + location_code="00", + optical_components=(inv.FiberSegment(name="f", optical_length=100_000.0),), + geometry=( + inv.Geometry( + name="west", + distance=(0.0, 50_000.0), + coordinates={"x": (0.0, 500.0), "y": (0.0, 0.0), "z": (0.0, 0.0)}, + ), + inv.Geometry( + name="east", + distance=(50_010.0, 100_000.0), + coordinates={ + "x": (600.0, 1000.0), + "y": (0.0, 0.0), + "z": (0.0, 0.0), + }, + ), + ), + ) + array = inv.FiberArray(code="L1", optical_paths=(one,)) + inventory = inv.Inventory( + coordinate_reference_system=inv.CoordinateReferenceSystem( + authority="", + code="", + name="grid", + coordinate_labels=("x", "y", "z"), + units=("meter", "meter", "meter"), + ), + networks=(inv.Network(code="N", fiber_arrays=(array,)),), + ).check() + # 10 m of gap on a 100 km path: a 1000-point grid steps over it. + ax = map_path(inventory, x="x", y="y") + segments = next( + x for x in ax.collections if isinstance(x, LineCollection) + ).get_segments() + crossing = [ + s for s in segments if s[:, 0].min() < 505.0 < 595.0 < s[:, 0].max() + ] + assert not crossing + + def test_time_filters(self, site): + """A time keeps only the epochs valid then.""" + ax = map_path(site, time="2026-08-01") + lines = [x for x in ax.collections if isinstance(x, LineCollection)] + assert len(lines) == 1 + + def test_same_axis(self, site): + """X and y must differ.""" + with pytest.raises(ParameterError, match="both 'x'"): + map_path(site, x="x", y="x") + + def test_not_an_axis(self, site): + """A non-axis column is refused with a pointer at path().""" + with pytest.raises(ParameterError, match="is not an axis"): + map_path(site, x="chainage", y="y") + + def test_nothing_placed(self, site): + """A path with no geometry cannot be mapped.""" + with pytest.raises(ParameterError, match="places itself in the CRS"): + map_path(site, "spur") + + def test_color_distance_colorbar(self, site): + """The default coloring earns a distance colorbar.""" + ax = map_path(site, "DAS.L1.00", time="2026-06-10") + assert "Optical distance" in _bar_label(ax) + + def test_scale_covers_what_the_view_shows(self, tunnel): + """Fiber a projection collapses to a point spends no colormap.""" + plan = map_path(tunnel, time="2024-07-01") + flat = next(x for x in plan.collections if isinstance(x, LineCollection)) + plt.close("all") + section = map_path(tunnel, x="x", y="z", time="2024-07-01") + deep = next(x for x in section.collections if isinstance(x, LineCollection)) + # Seen from above the boreholes are points, so the trench gets the + # whole scale; side-on they are 20 m of visible fiber and count. + assert flat.norm.vmax < deep.norm.vmax + drawn = np.asarray(deep.get_array()) + assert flat.norm.vmax < drawn.max() + + def test_a_view_which_shows_no_length(self): + """Where every segment collapses, the scale still spans the values.""" + one = inv.OpticalPath( + name="hole", + location_code="00", + optical_components=(inv.FiberSegment(name="f", optical_length=40.0),), + geometry=( + inv.Geometry( + name="down", + distance=(0.0, 40.0), + # Straight down: nothing to see in plan view at all. + coordinates={"x": (5.0, 5.0), "y": (2.0, 2.0), "z": (0.0, -40.0)}, + ), + ), + ) + array = inv.FiberArray(code="L1", optical_paths=(one,)) + inventory = inv.Inventory( + coordinate_reference_system=inv.CoordinateReferenceSystem( + authority="", + code="", + name="grid", + coordinate_labels=("x", "y", "z"), + units=("meter", "meter", "meter"), + ), + networks=(inv.Network(code="N", fiber_arrays=(array,)),), + ).check() + ax = map_path(inventory) + line = next(x for x in ax.collections if isinstance(x, LineCollection)) + assert line.norm.vmax > line.norm.vmin + + def test_discrete_values_get_a_stepped_scale(self, tunnel): + """Three boreholes are three categories, not a ramp through 1.5.""" + ax = map_path(tunnel, x="x", y="z", color="borehole", time="2024-07-01") + bar = ax.get_figure().axes[-1] + ticks = [x for x in bar.get_yticks() if x] or list(bar.get_xticks()) + assert [round(float(x), 3) for x in ticks] == [1.0, 2.0, 3.0] + + def test_one_number_is_not_a_scale(self): + """A column stating one value everywhere still draws.""" + + def build(location, value, group="reading"): + return inv.OpticalPath( + name=f"p{location}", + location_code=location, + optical_components=(inv.FiberSegment(name="f", optical_length=200.0),), + geometry=( + inv.Geometry( + name="run", + distance=(0.0, 200.0), + coordinates={ + "x": (0.0, 100.0), + "y": (float(location), float(location)), + "z": (0.0, 0.0), + }, + ), + ), + labels=( + ( + inv.OpticalPathLabel( + start_distance=0.0, + end_distance=200.0, + group=group, + value=value, + ), + ) + if value is not None + else () + ), + ) + + def wrap(*paths): + array = inv.FiberArray(code="L1", optical_paths=paths) + return inv.Inventory( + coordinate_reference_system=inv.CoordinateReferenceSystem( + authority="", + code="", + name="grid", + coordinate_labels=("x", "y", "z"), + units=("meter", "meter", "meter"), + ), + networks=(inv.Network(code="N", fiber_arrays=(array,)),), + ).check() + + ax = map_path(wrap(build("01", 7.0)), color="reading") + line = next(x for x in ax.collections if isinstance(x, LineCollection)) + assert line.norm.vmax > line.norm.vmin + + # One path states the number, the other says nothing under that + # name, so the drawn pieces are a mixture of scaled and unscaled. + plt.close("all") + ax = map_path(wrap(build("01", 7.0), build("02", None)), color="reading") + assert "n/a" in _legend_labels(ax) + assert len([x for x in ax.collections if isinstance(x, LineCollection)]) == 2 + + def test_color_column(self, site): + """A geometry column colors continuously, labelled by its name.""" + ax = map_path(site, "DAS.L1.00", time="2026-06-10", color="chainage") + assert _bar_label(ax) == "chainage" + + def test_color_label_group(self, site): + """A string label group gives a legend, with unplaced fiber named.""" + ax = map_path(site, "DAS.L1.00", time="2026-06-10", color="zone") + labels = _legend_labels(ax) + assert labels[:2] == ["north", "south"] + assert "n/a" in labels + assert ax.get_legend().get_title().get_text() == "zone" + + def test_color_numeric_group(self, site): + """A numeric label group colors continuously.""" + ax = map_path(site, "DAS.L1.00", time="2026-06-10", color="count") + assert _bar_label(ax) == "count" + + def test_color_coupling(self, site): + """Coupling types color the fiber.""" + ax = map_path(site, "DAS.L1.00", time="2026-06-10", color="coupling") + assert _legend_labels(ax)[:2] == ["trench", "conduit"] + + def test_unstated_numeric_is_drawn(self, site): + """Fiber whose color value is unstated is drawn grey, not made invisible.""" + ax = map_path(site, "DAS.L1.00", time="2026-06-10", color="count") + collection = next(x for x in ax.collections if isinstance(x, LineCollection)) + # The cable is placed from 350 m on, but states no count there, so + # those segments are masked and take the colormap's "bad" color. + assert np.ma.getmaskarray(collection.get_array()).any() + bad = collection.get_cmap().get_bad() + assert bad[3] == pytest.approx(1.0), "unstated fiber would be invisible" + assert "n/a" in _legend_labels(ax) + + def test_one_palette_for_every_path(self): + """A value is one color across the paths of one figure.""" + + def build(location, values): + return inv.OpticalPath( + name=f"p{location}", + location_code=location, + optical_components=(inv.FiberSegment(name="f", optical_length=200.0),), + geometry=( + inv.Geometry( + name="run", + distance=(0.0, 200.0), + coordinates={ + "x": (0.0, 100.0), + "y": (float(location), float(location)), + "z": (0.0, 0.0), + }, + ), + ), + labels=tuple( + inv.OpticalPathLabel( + start_distance=100.0 * index, + end_distance=100.0 * (index + 1), + group="zone", + value=value, + ) + for index, value in enumerate(values) + ), + ) + + # The two paths state the same two values in opposite order, so a + # palette built per path would give each value two colors. + array = inv.FiberArray( + code="L1", + optical_paths=( + build("01", ("north", "south")), + build("02", ("south", "north")), + ), + ) + inventory = inv.Inventory( + coordinate_reference_system=inv.CoordinateReferenceSystem( + authority="", + code="", + name="grid", + coordinate_labels=("x", "y", "z"), + units=("meter", "meter", "meter"), + ), + networks=(inv.Network(code="N", fiber_arrays=(array,)),), + ).check() + ax = map_path(inventory, color="zone") + lines = [x for x in ax.collections if isinstance(x, LineCollection)] + assert len(lines) == 2 + first, second = (x.get_colors() for x in lines) + # Path 01 begins in north and ends in south; path 02 is the other + # way round. A palette built per path would give both first + # segments color zero, so it is the crossed pairs which tell. + assert np.allclose(first[0], second[-1]), "north has two colors" + assert np.allclose(second[0], first[-1]), "south has two colors" + assert not np.allclose(first[0], second[0]) + assert [x.get_text() for x in ax.get_legend().get_texts()] == ["north", "south"] + + def test_shared_color_scale_across_paths(self): + """One numeric scale spans every path, not one scale each.""" + + def build(location, value): + return inv.OpticalPath( + name=f"p{location}", + location_code=location, + optical_components=(inv.FiberSegment(name="f", optical_length=200.0),), + geometry=( + inv.Geometry( + name="run", + distance=(0.0, 200.0), + coordinates={ + "x": (0.0, 100.0), + "y": (float(location), float(location)), + "z": (0.0, 0.0), + }, + ), + ), + labels=( + inv.OpticalPathLabel( + start_distance=0.0, + end_distance=100.0, + group="reading", + value=value, + ), + inv.OpticalPathLabel( + start_distance=100.0, + end_distance=200.0, + group="reading", + value=value + 1.0, + ), + ), + ) + + # One path states 0-1, the other 100-101. Normalized per path they + # would take identical colors despite stating different numbers. + array = inv.FiberArray( + code="L1", optical_paths=(build("01", 0.0), build("02", 100.0)) + ) + inventory = inv.Inventory( + coordinate_reference_system=inv.CoordinateReferenceSystem( + authority="", + code="", + name="grid", + coordinate_labels=("x", "y", "z"), + units=("meter", "meter", "meter"), + ), + networks=(inv.Network(code="N", fiber_arrays=(array,)),), + ).check() + ax = map_path(inventory, color="reading") + lines = [ + x + for x in ax.collections + if isinstance(x, LineCollection) and x.get_array() is not None + ] + assert len(lines) == 2 + # One scale object, spanning what both paths state. + assert lines[0].norm is lines[1].norm + norm = lines[0].norm + assert norm.vmin <= 0.0 and norm.vmax >= 101.0 + assert norm(0.0) != norm(100.0) + + def test_a_path_without_the_color_is_unstated(self): + """A placed path saying nothing under that name is drawn, not fatal.""" + + def build(location, labels=()): + return inv.OpticalPath( + name=f"p{location}", + location_code=location, + optical_components=(inv.FiberSegment(name="f", optical_length=200.0),), + geometry=( + inv.Geometry( + name="run", + distance=(0.0, 200.0), + coordinates={ + "x": (0.0, 100.0), + "y": (float(location), float(location)), + "z": (0.0, 0.0), + }, + ), + ), + labels=labels, + ) + + zoned = build( + "01", + ( + inv.OpticalPathLabel( + start_distance=0.0, end_distance=200.0, group="zone", value="north" + ), + ), + ) + array = inv.FiberArray(code="L1", optical_paths=(zoned, build("02"))) + inventory = inv.Inventory( + coordinate_reference_system=inv.CoordinateReferenceSystem( + authority="", + code="", + name="grid", + coordinate_labels=("x", "y", "z"), + units=("meter", "meter", "meter"), + ), + networks=(inv.Network(code="N", fiber_arrays=(array,)),), + ).check() + ax = map_path(inventory, color="zone") + assert len([x for x in ax.collections if isinstance(x, LineCollection)]) == 2 + assert _legend_labels(ax) == ["north", "n/a"] + + def test_map_needs_a_path_effective_then(self): + """A time no path is effective at draws nothing, and says why.""" + one = inv.OpticalPath( + name="main", + location_code="00", + start_time="2020-01-01", + end_time="2021-01-01", + optical_components=(inv.FiberSegment(name="f", optical_length=100.0),), + geometry=( + inv.Geometry( + name="run", + distance=(0.0, 100.0), + coordinates={"x": (0.0, 1.0), "y": (0.0, 0.0), "z": (0.0, 0.0)}, + ), + ), + ) + array = inv.FiberArray(code="L1", optical_paths=(one,)) + inventory = inv.Inventory( + coordinate_reference_system=inv.CoordinateReferenceSystem( + authority="", + code="", + name="grid", + coordinate_labels=("x", "y", "z"), + units=("meter", "meter", "meter"), + ), + networks=(inv.Network(code="N", fiber_arrays=(array,)),), + ).check() + with pytest.raises(ParameterError, match="is effective at"): + map_path(inventory, time="1999-01-01") + + def test_unknown_color_lists_what_the_inventory_states(self, site): + """A name no path states is refused, and the message says what is.""" + with pytest.raises(ParameterError, match="names neither") as info: + map_path(site, color="nope") + assert "zone" in str(info.value) + + def test_color_unknown(self, site): + """An unknown coloring lists what would work.""" + with pytest.raises(ParameterError, match="names neither"): + map_path(site, "DAS.L1.00", time="2026-06-10", color="nope") + + def test_legend_off_and_ax(self, site): + """legend=False draws none; a given ax keeps its figure size.""" + figure, ax = plt.subplots(figsize=(3, 3)) + out = map_path(site, "DAS.L1.00", time="2026-06-10", ax=ax, legend=False) + assert out is ax + assert len(figure.axes) == 1 + assert tuple(figure.get_size_inches()) == (3.0, 3.0) + + def test_explicit_aspect(self, site): + """aspect= is honored.""" + ax = map_path(site, aspect=2.0) + assert ax.get_aspect() == 2.0 + + def test_geographic_aspect(self): + """Degrees are not metres, so a geographic map is not forced equal.""" + ax = map_path(dc.get_example_inventory("random_das")) + assert ax.get_aspect() == "auto" + assert "degree" in ax.get_xlabel() + + def test_tunnel_coil(self, tunnel): + """The tunnel's slack coil is a gap in its section view.""" + ax = map_path(tunnel, x="x", y="z", color="section", time="2024-07-01") + assert "borehole" in _legend_labels(ax) + + def test_show(self, site, monkeypatch): + """Show calls plt.show.""" + called = [] + monkeypatch.setattr(plt, "show", lambda: called.append(True)) + map_path(site, show=True) + assert called + + +class TestTimeline: + """When each part was valid.""" + + def test_lanes_and_interrogators(self, site): + """Paths and acquisitions each get a lane; colors name interrogators.""" + ax = timeline(site) + lanes = _lanes(ax) + assert lanes[:2] == ["DAS.L1.00 [path]", "DAS.L1.01 [path]"] + assert "DAS.L1.00.RAW" in lanes and "DAS.L1.02.NIL" in lanes + labels = _legend_labels(ax) + for expected in ("optical path", "Fake FI-1", "sn-9", "interrogator"): + assert expected in labels + assert "no interrogator" in labels + assert ax.get_xlabel() == "Time" + + def test_kind(self, site): + """kind= keeps only acquisitions or only paths.""" + assert all("[path]" in x for x in _lanes(timeline(site, kind="optical_path"))) + assert not any( + "[path]" in x for x in _lanes(timeline(site, kind="acquisition")) + ) + + def test_color_data_type(self, site): + """Data type coloring names the unstated ones.""" + labels = _legend_labels(timeline(site, color="data_type")) + assert "strain_rate" in labels and "unstated" in labels + + def test_color_kind(self, site): + """Kind coloring has two entries.""" + labels = _legend_labels(timeline(site, color="kind")) + assert sorted(labels) == ["acquisition", "optical path"] + + @pytest.mark.parametrize("bad", [dict(kind="nope"), dict(color="nope")]) + def test_bad_options(self, site, bad): + """Unknown kind or color is refused.""" + with pytest.raises(ParameterError, match="nope"): + timeline(site, **bad) + + def test_time_window(self, site): + """time=(start, end) sets the axis and leaves out epochs beyond it.""" + ax = timeline(site, time=("2026-06-01", "2026-06-30")) + low, high = ax.get_xlim() + assert high - low == pytest.approx(29.0) + # The repaired path and its acquisition start in July. + assert len(_boxes(ax)[0].get_paths()) == 1 + assert "DAS.L1.00.RAW" in _lanes(ax) + + def test_open_epochs_are_hatched(self, site): + """An epoch stating no bound runs off that side of the axis.""" + ax = timeline(site, kind="acquisition") + hatched = [x for x in _boxes(ax) if x.get_hatch()] + assert hatched, "an unbounded epoch drew no open edge" + + def test_half_open_window(self, site): + """One end of the window may be left to the data.""" + ax = timeline(site, time=("2026-06-20", None)) + low, high = ax.get_xlim() + assert low < high + ax = timeline(site, time=(None, "2026-06-20")) + assert ax.get_xlim()[0] < ax.get_xlim()[1] + + @pytest.mark.parametrize( + "bad, match", + [ + (("2026-07-01", "2026-06-01"), "must be increasing"), + ("nope", "must be a .start, end. pair"), + (("not a time", None), "not a time"), + ], + ) + def test_bad_time_window(self, site, bad, match): + """A window which is not a window is refused before anything is drawn.""" + plt.close("all") + with pytest.raises(ParameterError, match=match): + timeline(site, time=bad) + assert plt.get_fignums() == [] + + def test_window_on_an_inventory_stating_no_time(self): + """A window still works where the epochs state nothing themselves.""" + undated = dc.get_example_inventory("random_das") + low, high = timeline(undated, time=("2026-01-01", None)).get_xlim() + assert high > low + low, high = timeline(undated, time=(None, "2026-01-01")).get_xlim() + assert high > low + + def test_a_bound_which_is_not_a_time(self, site): + """A bound which parses to nothing is refused like any other.""" + with pytest.raises(ParameterError, match="not a time"): + timeline(site, time=(float("nan"), None)) + + def test_time_window_empty(self): + """A window nothing falls in is an error, not a blank figure.""" + acquisition = inv.Acquisition( + code="RAW", + location_code="00", + start_time="2026-06-01", + end_time="2026-06-15", + data_category="DAS", + sample_rate=1.0, + gauge_length=1.0, + ) + array = inv.FiberArray(code="A", acquisitions=(acquisition,)) + bounded = inv.Inventory( + networks=(inv.Network(code="N", fiber_arrays=(array,)),) + ) + with pytest.raises(ParameterError, match="falls within time"): + timeline(bounded, time=("2020-01-01", "2020-02-01")) + # And a window after it, since the epoch states both of its bounds. + with pytest.raises(ParameterError, match="falls within time"): + timeline(bounded, time=("2030-01-01", "2030-02-01")) + + def test_window_excludes_touching_epoch(self): + """An epoch which ends where the window starts does not overlap it.""" + acquisition = inv.Acquisition( + code="RAW", + location_code="00", + start_time="2026-06-01", + end_time="2026-06-15", + data_category="DAS", + sample_rate=1.0, + gauge_length=1.0, + ) + array = inv.FiberArray(code="A", acquisitions=(acquisition,)) + bounded = inv.Inventory( + networks=(inv.Network(code="N", fiber_arrays=(array,)),) + ) + with pytest.raises(ParameterError, match="falls within time"): + timeline(bounded, time=("2026-06-15", "2026-07-01")) + with pytest.raises(ParameterError, match="falls within time"): + timeline(bounded, time=("2026-05-01", "2026-06-01")) + + def test_no_epochs(self): + """An inventory whose epochs state no time still draws, and says so.""" + ax = timeline(dc.get_example_inventory("random_das")) + assert "states one" in ax.get_xlabel() + assert list(ax.get_xticks()) == [] + + def test_nothing_to_draw(self): + """No acquisitions and no paths is an error.""" + empty = inv.Inventory( + networks=( + inv.Network(code="DAS", fiber_arrays=(inv.FiberArray(code="A"),)), + ) + ) + with pytest.raises(ParameterError, match="nothing with a time epoch"): + timeline(empty) + + def test_ax_and_show(self, site, monkeypatch): + """A given ax is drawn on; show calls plt.show.""" + called = [] + monkeypatch.setattr(plt, "show", lambda: called.append(True)) + _, ax = plt.subplots() + assert timeline(site, ax=ax, show=True) is ax + assert called + + def test_tunnel_repair(self, tunnel): + """The tunnel's path lane holds two epochs split at the repair.""" + ax = timeline(tunnel, kind="optical_path") + boxes = _boxes(ax)[0] + assert len(boxes.get_paths()) == 2 diff --git a/tests/test_viz/test_lanes.py b/tests/test_viz/test_lanes.py new file mode 100644 index 000000000..60278c82d --- /dev/null +++ b/tests/test_viz/test_lanes.py @@ -0,0 +1,426 @@ +"""Tests for the interval-lane renderer.""" + +from __future__ import annotations + +import datetime + +import matplotlib.dates as mdates +import matplotlib.pyplot as plt +import numpy as np +import pandas as pd +import pytest +from matplotlib.collections import PatchCollection + +from dascore.exceptions import ParameterError +from dascore.viz._lanes import UNCOVERED_COLOR, _pack_rows, plot_lanes + + +def _collections(ax): + """The patch collections an axes holds, in drawing order.""" + return [x for x in ax.collections if isinstance(x, PatchCollection)] + + +def _extents(collection): + """The (x0, width) of every path in a collection.""" + out = [] + for path in collection.get_paths(): + box = path.get_extents() + out.append((float(box.x0), float(box.width))) + return out + + +def _texts(ax): + """Every label drawn on the axes.""" + return [x.get_text() for x in ax.texts] + + +@pytest.fixture() +def string_frame(): + """Two lanes of named zones.""" + return pd.DataFrame( + { + "group": ["zone", "zone", "other", "other"], + "start": [0.0, 10.0, 0.0, 30.0], + "end": [10.0, 20.0, 20.0, 40.0], + "value": ["north", "south", "north", "west"], + } + ) + + +@pytest.fixture() +def kinds_frame(): + """One lane of every value kind, plus a point marker.""" + return pd.DataFrame( + { + "lane": ["text", "text", "flag", "flag", "count", "count", "tick"], + "start": [0.0, 5.0, 0.0, 5.0, 0.0, 5.0, 3.0], + "end": [5.0, 10.0, 5.0, 10.0, 5.0, 10.0, 3.0], + "value": ["a", "b", True, False, 1, 2, None], + } + ) + + +class TestReadFrame: + """The frame contract: named columns, kinds, and refusals.""" + + def test_renamed_columns(self): + """Bound columns are named by the caller, as a spool frame needs.""" + frame = pd.DataFrame({"time_min": [0.0, 5.0], "time_max": [4.0, 9.0]}) + ax = plot_lanes(frame, start="time_min", end="time_max") + assert _extents(_collections(ax)[0]) == [(0.0, 4.0), (5.0, 4.0)] + + def test_mapping_input(self): + """A plain mapping of columns is accepted as a frame.""" + ax = plot_lanes({"start": [0.0], "end": [1.0]}) + assert len(_collections(ax)) == 1 + + def test_datetime_bounds(self): + """Datetime bounds convert to matplotlib dates once, up front.""" + frame = pd.DataFrame( + { + "start": pd.to_datetime(["2024-01-01", "2024-01-03"]), + "end": pd.to_datetime(["2024-01-02", "2024-01-05"]), + } + ) + limits = pd.to_datetime(["2024-01-01", "2024-01-06"]).to_numpy() + ax = plot_lanes(frame, x_limits=limits) + widths = [w for _, w in _extents(_collections(ax)[0])] + assert widths == pytest.approx([1.0, 2.0]) + + def test_timezone_aware_bounds(self): + """Zoned datetimes are drawn at their UTC instant.""" + frame = pd.DataFrame( + { + "start": pd.to_datetime(["2024-01-01T00:00"]).tz_localize("UTC"), + "end": pd.to_datetime(["2024-01-02T00:00"]).tz_localize("UTC"), + } + ) + ax = plot_lanes(frame) + x0, width = _extents(_collections(ax)[0])[0] + assert width == pytest.approx(1.0) + assert x0 == pytest.approx(mdates.date2num(np.datetime64("2024-01-01"))) + + def test_datetime_x_limits(self): + """Limits given as plain datetimes land on the same axis as the bars.""" + frame = pd.DataFrame( + { + "start": pd.to_datetime(["2024-01-02"]), + "end": pd.to_datetime(["2024-01-03"]), + } + ) + limits = (datetime.datetime(2024, 1, 1), datetime.datetime(2024, 1, 4)) + ax = plot_lanes(frame, x_limits=limits) + assert ax.get_xlim()[0] == pytest.approx( + mdates.date2num(np.datetime64("2024-01-01")) + ) + + def test_datetime_axis_is_formatted(self): + """Dated bounds get a date axis, not raw ordinals.""" + frame = pd.DataFrame( + { + "start": pd.to_datetime(["2024-01-01"]), + "end": pd.to_datetime(["2024-01-05"]), + } + ) + ax = plot_lanes(frame) + assert isinstance(ax.xaxis.get_major_formatter(), mdates.ConciseDateFormatter) + ticks = " ".join(x.get_text() for x in ax.get_xticklabels()) + assert "Jan" in ticks, f"expected dates, got {ticks}" + + def test_missing_bounds(self): + """A frame without the bound columns names what it has.""" + with pytest.raises(ParameterError, match="needs the columns"): + plot_lanes(pd.DataFrame({"a": [1]})) + + def test_missing_named_column(self): + """A lane/value/label name not in the frame is refused.""" + frame = pd.DataFrame({"start": [0.0], "end": [1.0]}) + with pytest.raises(ParameterError, match="lane='group' is not a column"): + plot_lanes(frame, lane="group") + + def test_empty_frame(self): + """Nothing to draw is an error, not a blank figure.""" + with pytest.raises(ParameterError, match="no rows"): + plot_lanes(pd.DataFrame({"start": [], "end": []})) + + def test_backwards_interval(self): + """An interval ending before it starts is refused by lane.""" + frame = pd.DataFrame({"start": [5.0], "end": [1.0], "lane": ["x"]}) + with pytest.raises(ParameterError, match="lane 'x' ends before it starts"): + plot_lanes(frame, lane="lane") + + def test_mixed_kinds(self): + """A lane mixing strings and numbers has no one color scheme.""" + frame = pd.DataFrame({"start": [0.0, 1.0], "end": [1.0, 2.0], "v": ["a", 1]}) + with pytest.raises(ParameterError, match="mixes value kinds"): + plot_lanes(frame, value="v") + + def test_duplicate_lanes(self): + """Naming a lane twice in lanes is refused.""" + frame = pd.DataFrame({"start": [0.0], "end": [1.0], "lane": ["a"]}) + with pytest.raises(ParameterError, match="names a lane twice"): + plot_lanes(frame, lane="lane", lanes=("a", "a")) + + +class TestLayout: + """Lane order, packing, points, and open edges.""" + + def test_lane_order_first_appearance(self, string_frame): + """Lanes appear in the order the frame first names them.""" + ax = plot_lanes(string_frame, lane="group", value="value") + assert [x.get_text() for x in ax.get_yticklabels()] == ["zone", "other"] + + def test_explicit_lanes_filter_and_pad(self, string_frame): + """lanes= orders, filters, and keeps an empty lane for alignment.""" + ax = plot_lanes( + string_frame, lane="group", value="value", lanes=("other", "empty") + ) + assert [x.get_text() for x in ax.get_yticklabels()] == ["other", "empty"] + # Only the one populated lane produced boxes. + assert len(_collections(ax)) == 1 + + def test_packing_overlaps(self): + """Overlapping intervals take separate sub-rows.""" + frame = pd.DataFrame({"start": [0.0, 5.0, 20.0], "end": [10.0, 15.0, 30.0]}) + assert _pack_rows(frame).tolist() == [0, 1, 0] + ax = plot_lanes(frame) + heights = { + round(float(p.get_extents().height), 3) + for p in _collections(ax)[0].get_paths() + } + assert heights == {0.4} + + def test_packing_caps_sub_rows(self): + """A pileup degrades to the last sub-row rather than growing forever.""" + n = 12 + frame = pd.DataFrame({"start": [0.0] * n, "end": [10.0] * n}) + assert _pack_rows(frame).max() == 7 + + def test_no_packing(self): + """pack=False draws everything in one row.""" + frame = pd.DataFrame({"start": [0.0, 5.0], "end": [10.0, 15.0]}) + ax = plot_lanes(frame, pack=False) + heights = { + round(float(p.get_extents().height), 3) + for p in _collections(ax)[0].get_paths() + } + assert heights == {0.8} + + def test_point_marker(self): + """An interval of zero width is drawn as a tick, not lost.""" + frame = pd.DataFrame({"start": [0.0, 5.0], "end": [10.0, 5.0]}) + ax = plot_lanes(frame) + assert len(_collections(ax)[0].get_paths()) == 1 + xs = [line.get_xdata()[0] for line in ax.lines] + assert xs == [5.0, 5.0] + + def test_open_edges(self): + """Open bounds earn a hatched sliver at that end.""" + frame = pd.DataFrame( + { + "start": [0.0, 10.0], + "end": [10.0, 20.0], + "open_start": [True, False], + "open_end": [False, True], + } + ) + ax = plot_lanes(frame) + hatched = [c for c in _collections(ax) if c.get_hatch()] + assert len(hatched) == 1 + starts = sorted(x for x, _ in _extents(hatched[0])) + assert starts[0] == pytest.approx(0.0) + assert starts[1] < 20.0 + + def test_labels_fit_or_drop(self): + """A label wider than its box is dropped; others are drawn.""" + frame = pd.DataFrame( + {"start": [0.0, 50.0], "end": [50.0, 50.5], "v": ["wide", "narrow"]} + ) + ax = plot_lanes(frame, value="v") + assert _texts(ax) == ["wide"] + + def test_max_labels(self): + """Past max_labels no text is drawn at all.""" + frame = pd.DataFrame({"start": [0.0, 50.0], "end": [50.0, 100.0]}) + frame["v"] = ["a", "b"] + ax = plot_lanes(frame, value="v", max_labels=1) + assert _texts(ax) == [] + + def test_label_column(self): + """label= overrides the default text.""" + frame = pd.DataFrame({"start": [0.0], "end": [100.0], "v": [3.5], "t": ["x"]}) + ax = plot_lanes(frame, value="v", label="t") + assert _texts(ax) == ["x"] + + def test_default_labels(self, kinds_frame): + """Numbers state themselves; booleans and None draw no text.""" + ax = plot_lanes(kinds_frame, lane="lane", value="value") + assert sorted(_texts(ax)) == ["1", "2", "a", "b"] + + def test_x_label_and_show(self, monkeypatch): + """x_label is applied and show calls plt.show.""" + called = [] + monkeypatch.setattr(plt, "show", lambda: called.append(True)) + ax = plot_lanes({"start": [0.0], "end": [1.0]}, x_label="Time", show=True) + assert ax.get_xlabel() == "Time" + assert called + + +class TestColors: + """The color policy per value kind and the overrides.""" + + def test_string_colors_frame_wide(self, string_frame): + """One string value is one color in every lane.""" + ax = plot_lanes(string_frame, lane="group", value="value") + zone, other = _collections(ax) + assert np.allclose(zone.get_facecolors()[0], other.get_facecolors()[0]) + labels = [x.get_text() for x in ax.get_legend().get_texts()] + assert labels == ["north", "south", "west"] + + def test_boolean_lane(self, kinds_frame): + """False is the lane's color at low alpha; the legend names the lane.""" + ax = plot_lanes(kinds_frame, lane="lane", value="value", lanes=("flag",)) + colors = _collections(ax)[0].get_facecolors() + assert colors[0][3] == pytest.approx(1.0) + assert colors[1][3] == pytest.approx(0.25) + assert [x.get_text() for x in ax.get_legend().get_texts()] == ["flag"] + + def test_numeric_few_values(self, kinds_frame): + """A few numbers are colored continuously but earn no colorbar.""" + ax = plot_lanes(kinds_frame, lane="lane", value="value", lanes=("count",)) + colors = _collections(ax)[0].get_facecolors() + assert not np.allclose(colors[0], colors[1]) + assert len(ax.get_figure().axes) == 1 + + def test_numeric_many_values_colorbar(self): + """Past a handful of distinct numbers a colorbar is drawn.""" + n = 10 + frame = pd.DataFrame( + {"start": np.arange(n) * 1.0, "end": np.arange(n) + 1.0, "v": range(n)} + ) + ax = plot_lanes(frame, value="v") + assert len(ax.get_figure().axes) == 2 + + def test_numeric_lanes_get_their_own_colorbar(self): + """Each numeric lane is its own scale, so each names its own bar.""" + n = 8 + frame = pd.DataFrame( + { + "lane": ["a"] * n + ["b"] * n, + "start": list(range(n)) * 2, + "end": [x + 1 for x in range(n)] * 2, + "value": list(range(n)) + [100 + x for x in range(n)], + } + ) + ax = plot_lanes(frame, lane="lane", value="value") + bars = [x for x in ax.get_figure().axes if x is not ax] + assert [x.get_ylabel() for x in bars] == ["a", "b"] + assert bars[0].get_ylim() == pytest.approx((0.0, 7.0)) + assert bars[1].get_ylim() == pytest.approx((100.0, 107.0)) + + def test_numeric_one_value(self): + """One number is not a scale, so every box shares one color.""" + frame = pd.DataFrame({"start": [0.0, 1.0], "end": [1.0, 2.0], "v": [4, 4]}) + ax = plot_lanes(frame, value="v") + colors = _collections(ax)[0].get_facecolors() + assert np.allclose(colors[0], colors[1]) + + def test_color_string(self, string_frame): + """A single color string paints every box, legend and all.""" + ax = plot_lanes(string_frame, lane="group", value="value", color="red") + for collection in _collections(ax): + assert np.allclose(collection.get_facecolors()[:, :3], [1, 0, 0]) + assert ax.get_legend() is None + + def test_color_numeric_cmap(self, kinds_frame): + """For a numeric lane a color string names the colormap.""" + ax = plot_lanes( + kinds_frame, lane="lane", value="value", lanes=("count",), color="Greys" + ) + colors = _collections(ax)[0].get_facecolors() + assert np.allclose(colors[:, 0], colors[:, 1]) + + def test_color_mapping(self, string_frame): + """A value->color mapping applies, and unmapped values are grey.""" + ax = plot_lanes( + string_frame, lane="group", value="value", color={"north": "blue"} + ) + zone = _collections(ax)[0].get_facecolors() + assert np.allclose(zone[0][:3], [0, 0, 1]) + assert np.allclose(zone[1][:3], plt.matplotlib.colors.to_rgb(UNCOVERED_COLOR)) + assert [x.get_text() for x in ax.get_legend().get_texts()] == ["north"] + + def test_color_by_lane_mapping(self, string_frame): + """A lane->mapping mapping colors each lane its own way.""" + color = {"zone": {"north": "blue"}, "missing": "red"} + ax = plot_lanes(string_frame, lane="group", value="value", color=color) + zone, other = _collections(ax) + assert np.allclose(zone.get_facecolors()[0][:3], [0, 0, 1]) + # The lane the mapping does not name takes the default string colors. + assert not np.allclose(other.get_facecolors()[0][:3], [0, 0, 1]) + + def test_color_name_on_a_numeric_lane(self, kinds_frame): + """A color which names no colormap is a color, not an error.""" + ax = plot_lanes( + kinds_frame, lane="lane", value="value", lanes=("count",), color="red" + ) + colors = _collections(ax)[0].get_facecolors() + assert np.allclose(colors[:, :3], [1, 0, 0]) + + def test_a_value_which_is_not_a_value(self): + """A lane value of NaN is refused, the way the model refuses it.""" + n = 10 + values = [float(x) for x in range(n)] + values[3] = float("nan") + frame = pd.DataFrame( + {"start": np.arange(n) * 1.0, "end": np.arange(n) + 1.0, "v": values} + ) + with pytest.raises(ParameterError, match="must be finite"): + plot_lanes(frame, value="v") + + def test_legend_off_suppresses_the_colorbar(self): + """legend='off' means no colorbar either.""" + n = 10 + frame = pd.DataFrame( + {"start": np.arange(n) * 1.0, "end": np.arange(n) + 1.0, "v": range(n)} + ) + ax = plot_lanes(frame, value="v", legend="off") + assert len(ax.get_figure().axes) == 1 + + def test_vocabulary_widens_the_palette(self, string_frame): + """A value the frame lacks still reserves its color.""" + partial = string_frame[string_frame["group"] == "zone"] + alone = plot_lanes(partial, lane="group", value="value") + plt.close("all") + together = plot_lanes(partial, lane="group", value="value", vocabulary=["west"]) + # 'west' sorts after 'south', so reserving it must not move north. + assert np.allclose( + _collections(alone)[0].get_facecolors()[0], + _collections(together)[0].get_facecolors()[0], + ) + shifted = plot_lanes(partial, lane="group", value="value", vocabulary=["a"]) + assert not np.allclose( + _collections(alone)[0].get_facecolors()[0], + _collections(shifted)[0].get_facecolors()[0], + ) + + def test_labels_decided_the_same_at_any_dpi(self): + """Whether a label fits is a question about the figure, not its dpi.""" + frame = pd.DataFrame( + {"start": [0.0], "end": [1.0], "v": ["a rather long label"]} + ) + drawn = [] + for dpi in (50, 200): + _, ax = plt.subplots(figsize=(2, 1), dpi=dpi) + plot_lanes(frame, ax=ax, value="v") + drawn.append(_texts(ax)) + plt.close("all") + # Measuring text in points against a box in pixels answers this + # differently at each dpi, which is how the same figure saved at + # two resolutions loses its labels. + assert drawn[0] == drawn[1] + + def test_legend_off(self, string_frame): + """legend=False draws none.""" + ax = plot_lanes(string_frame, lane="group", value="value", legend=False) + assert ax.get_legend() is None