Merge branch '381-parallel-plotting-hides-stderr' into 'main' (35062f03) · Commits · Utopia / dantro

CHANGELOG.md

+2 −2

Original line number	Diff line number	Diff line
		@@ -3,8 +3,8 @@
		`dantro` aims to adhere to [semantic versioning](https://semver.org/).
		However, given the rather burst-like development on this package, features are often released immediately, sometimes also as a "patch" (`+0.0.1`) release.

		## v0.21.2 🚧
		...
		## v0.21.2
		- !376 fixes an error that suppressed `stderr` output during parallel plotting.

		#### Internal
		- !375 Addresses several warnings in tests and elsewhere, mostly related to upstream packages.

README.md

+1 −1

Original line number	Diff line number	Diff line
		@@ -236,7 +236,7 @@ dantro is licensed under the [GNU Lesser General Public License Version 3][LGPLv
		### Copyright Notice

		dantro -- a python package for handling and plotting hierarchical data
		Copyright (C) 2018 – 2025 dantro developers
		Copyright (C) 2018 – 2026 dantro developers

		This program is free software: you can redistribute it and/or modify
		it under the terms of the GNU Lesser General Public License as published by

dantro/init.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -15,7 +15,7 @@ an automated sequence of predefined, configurable operations.
		See :ref:`the user manual <welcome>` for more information.
		"""

		__version__ = "0.21.2b0"
		__version__ = "0.21.2"
		"""Package version"""

		# Set up the root logger such that the logging configuration is applied

dantro/data_ops/arr_ops.py

+30 −27

Original line number	Diff line number	Diff line
		"""Implements data operations that work on array-like data, e.g. from numpy
		or xarray."""

		from __future__ import annotations

		import logging
		from typing import (
		TYPE_CHECKING,
		@@ -15,6 +17,7 @@ from typing import (
		)

		if TYPE_CHECKING:
		import numpy
		import xarray

		import numpy as np
		@@ -30,8 +33,8 @@ xr = LazyLoader("xarray")


		def apply_along_axis(
		func: Callable, axis: int, arr: np.ndarray, args, *kwargs
		) -> np.ndarray:
		func: Callable, axis: int, arr: numpy.ndarray, args, *kwargs
		) -> numpy.ndarray:
		"""This is like numpy's function of the same name, but does not try to
		cast the results of func to an :py:class:`numpy.ndarray` but tries to keep
		them as dtype object. Thus, the return value of this function will always
		@@ -84,8 +87,8 @@ def apply_along_axis(


		def create_mask(
		data: "xarray.DataArray", operator_name: str, rhs_value: float
		) -> "xarray.DataArray":
		data: xarray.DataArray, operator_name: str, rhs_value: float
		) -> xarray.DataArray:
		"""Given the data, returns a binary mask by applying the following
		comparison: ``data <operator> rhs value``.

		@@ -128,8 +131,8 @@ def create_mask(


		def where(
		data: "xarray.DataArray", operator_name: str, rhs_value: float, **kwargs
		) -> "xarray.DataArray":
		data: xarray.DataArray, operator_name: str, rhs_value: float, **kwargs
		) -> xarray.DataArray:
		"""Filter elements from the given data according to a condition. Only
		those elemens where the condition is fulfilled are not masked.

		@@ -151,7 +154,7 @@ def where(
		)


		def count_unique(data, dims: List[str] = None) -> "xarray.DataArray":
		def count_unique(data, dims: List[str] = None) -> xarray.DataArray:
		"""Applies :py:func:`numpy.unique` to the given data and constructs a
		:py:class:`xarray.DataArray` for the results.

		@@ -165,7 +168,7 @@ def count_unique(data, dims: List[str] = None) -> "xarray.DataArray":

		"""

		def _count_unique(data) -> "xarray.DataArray":
		def _count_unique(data) -> xarray.DataArray:
		unique, counts = np.unique(data, return_counts=True)

		# remove np.nan values
		@@ -212,11 +215,11 @@ def count_unique(data, dims: List[str] = None) -> "xarray.DataArray":
		def populate_ndarray(
		objs: Iterable,
		shape: Tuple[int] = None,
		dtype: Union[str, type, np.dtype] = float,
		dtype: Union[str, type, numpy.dtype] = float,
		order: str = "C",
		out: np.ndarray = None,
		out: numpy.ndarray = None,
		ufunc: Callable = None,
		) -> np.ndarray:
		) -> numpy.ndarray:
		"""Populates an empty :py:class:`numpy.ndarray` of the given ``dtype`` with
		the given objects by zipping over a new array of the given ``shape`` and
		the sequence of objects.
		@@ -272,7 +275,7 @@ def build_object_array(
		*,
		dims: Tuple[str] = ("label",),
		fillna: Any = None,
		) -> "xarray.DataArray":
		) -> xarray.DataArray:
		"""Creates a simple labelled multidimensional object array.

		It accepts simple iterable types like dictionaries or lists and unpacks
		@@ -374,14 +377,14 @@ def build_object_array(


		def multi_concat(
		arrs: np.ndarray,
		arrs: numpy.ndarray,
		*,
		dims: Sequence[str],
		join: str = "outer",
		compat: str = "no_conflicts",
		coords: str = "different",
		**kwargs,
		) -> "xarray.DataArray":
		) -> xarray.DataArray:
		"""Concatenates :py:class:`xarray.Dataset` or :py:class:`xarray.DataArray`
		objects using :py:func:`xarray.concat`. This function expects the xarray
		objects to be pre-aligned inside the numpy object array ``arrs``, with
		@@ -461,14 +464,14 @@ def multi_concat(

		def merge(
		arrs: Union[
		Sequence[Union["xarray.DataArray", "xarray.Dataset"]], np.ndarray
		Sequence[Union[xarray.DataArray, xarray.Dataset]], numpy.ndarray
		],
		*,
		reduce_to_array: bool = False,
		join: str = "outer",
		compat: str = "no_conflicts",
		**merge_kwargs,
		) -> Union["xarray.Dataset", "xarray.DataArray"]:
		) -> Union[xarray.Dataset, xarray.DataArray]:
		"""Merges the given sequence of xarray objects into an
		:py:class:`xarray.Dataset`.

		@@ -480,7 +483,7 @@ def merge(
		making that array the return value of this operation.

		Args:
		arrs (Union[Sequence[Union["xarray.DataArray", "xarray.Dataset"]], numpy.ndarray]):
		arrs (Union[Sequence[Union[xarray.DataArray, xarray.Dataset]], numpy.ndarray]):
		The sequence of xarray objects to merge.
		If a numpy array is given, it is flattened.
		reduce_to_array (bool, optional): If True, the resulting Dataset is
		@@ -518,8 +521,8 @@ def merge(


		def expand_dims(
		d: Union[np.ndarray, "xarray.DataArray"], , dim: dict = None, *kwargs
		) -> "xarray.DataArray":
		d: Union[numpy.ndarray, xarray.DataArray], , dim: dict = None, *kwargs
		) -> xarray.DataArray:
		"""Expands the dimensions of the given object.

		If the object does not support a ``expand_dims`` method call, it will be
		@@ -543,16 +546,16 @@ def expand_dims(


		def expand_object_array(
		d: "xarray.DataArray",
		d: xarray.DataArray,
		*,
		shape: Sequence[int] = None,
		astype: Union[str, type, np.dtype] = None,
		astype: Union[str, type, numpy.dtype] = None,
		dims: Sequence[str] = None,
		coords: Union[dict, str] = "trivial",
		combination_method: str = "concat",
		allow_reshaping_failure: bool = False,
		**combination_kwargs,
		) -> "xarray.DataArray":
		) -> xarray.DataArray:
		"""Expands a labelled object-array that contains array-like objects into a
		higher-dimensional labelled array.

		@@ -623,15 +626,15 @@ def expand_object_array(
		"""

		def prepare_item(
		d: "xarray.DataArray",
		d: xarray.DataArray,
		*,
		midx: Sequence[int],
		shape: Sequence[int],
		astype: Union[str, type, np.dtype, None],
		astype: Union[str, type, numpy.dtype, None],
		name: str,
		dims: Sequence[str],
		generate_coords: Callable,
		) -> Union["xarray.DataArray", None]:
		) -> Union[xarray.DataArray, None]:
		"""Extracts the desired element and reshapes and labels it accordingly.
		If any of this fails, returns ``None``.
		"""
		@@ -753,12 +756,12 @@ def expand_object_array(


		def transform_coords(
		d: "xarray.DataArray",
		d: xarray.DataArray,
		dim: Union[str, Sequence[str]],
		func: Callable,
		*,
		func_kwargs: dict = None,
		) -> "xarray.DataArray":
		) -> xarray.DataArray:
		"""Assigns new, transformed coordinates to a data array by applying a
		function on the existing coordinates.

dantro/plot_mngr.py

+34 −11

Original line number	Diff line number	Diff line
		@@ -698,25 +698,38 @@ class PlotManager:
		out_path: str,
		cfg: dict,
		plot_cfg: dict,
		) -> Tuple[int, Tuple[str, str], Union[bool, str]]:
		) -> Tuple[int, str, str, Union[bool, str]]:
		"""Shallow wrapper around plot creator invocation, preparing arguments
		for invocation in the context of parallel execution."""
		for invocation in the context of parallel execution.

		Returns:
		Tuple of (task_key, captured_stdout, captured_stderr, return_value)
		"""
		import contextlib
		import io

		from .logging import DantroLogger

		captured = io.StringIO("")
		captured_stdout = io.StringIO("")
		captured_stderr = io.StringIO("")
		DantroLogger.change_settings(
		divert_to=captured,
		divert_to=captured_stdout,
		suppress_in_child_process=True,
		)

		with contextlib.redirect_stdout(captured):
		with (
		contextlib.redirect_stdout(captured_stdout),
		contextlib.redirect_stderr(captured_stderr),
		):
		rv = self._invoke_plot_creation(
		plot_creator, out_path=out_path, cfg, plot_cfg
		)
		return task_key, captured.getvalue(), rv
		return (
		task_key,
		captured_stdout.getvalue(),
		captured_stderr.getvalue(),
		rv,
		)

		def _invoke_parallel_executor_benchmark(self) -> bool:
		"""A method that is passed to a parallel executor for benchmarking
		@@ -1821,6 +1834,7 @@ class PlotManager:
		len(tasks),
		type(executor).__name__,
		)
		futures_to_task: Dict[concfu.Future, int] = {}
		for task_key, task_kwargs in tasks.items():
		log.debug(
		" Submitting task '%s':\n %s\n", task_key, task_kwargs
		@@ -1831,6 +1845,7 @@ class PlotManager:
		**task_kwargs,
		)
		futures.append(future)
		futures_to_task[future] = task_key

		# TODO Can add callbacks to future here perhaps?

		@@ -1842,19 +1857,22 @@ class PlotManager:
		for n, future in enumerate(concfu.as_completed(futures)):
		exc = future.exception()
		if exc:
		task_key = futures_to_task[future]
		log.error(
		"Parallel plotting task %s failed with a %s: %s",
		future,
		"Parallel plotting task %d failed with a %s: %s",
		task_key,
		type(exc).__name__,
		exc,
		)
		num_failed += 1
		exceptions[n] = exc
		exceptions[task_key] = exc
		continue

		# Get result
		log.debug("Task completed: %s ", future)
		task_key, captured_stdout, rv = future.result()
		task_key, captured_stdout, captured_stderr, rv = (
		future.result()
		)

		# Get parameters
		task = tasks[task_key]
		@@ -1902,9 +1920,14 @@ class PlotManager:
		)
		if captured_stdout:
		log.remark(
		"Captured output:\n\n%s\n",
		"Captured stdout:\n\n%s\n",
		textwrap.indent(captured_stdout, " " * 4),
		)
		if captured_stderr:
		log.caution(
		"Captured stderr:\n\n%s\n",
		textwrap.indent(captured_stderr, " " * 4),
		)

		# Estimate for time remaining
		log.progress("Finished '%s' plot %d/%d.", name, n + 1, n_max)