Source code for droplets.trackers

"""
Module defining classes for tracking droplets in simulations.

.. autosummary::
   :nosignatures:

   LengthScaleTracker
   DropletTracker

.. codeauthor:: David Zwicker <david.zwicker@ds.mpg.de>
"""

from __future__ import annotations

import math
from typing import Any, Callable, Literal

from pde.fields.base import FieldBase
from pde.tools.docstrings import fill_in_docstring
from pde.trackers.base import InfoDict, InterruptData, TrackerBase

from .emulsions import EmulsionTimeCourse


[docs] class LengthScaleTracker(TrackerBase): """Tracker that stores length scales measured in simulations Attributes: times (list): The time points at which the length scales are stored length_scales (list): The associated length scales """ @fill_in_docstring def __init__( self, interrupts: InterruptData = 1, filename: str | None = None, *, method: Literal[ "structure_factor_mean", "structure_factor_maximum", "droplet_detection" ] = "structure_factor_mean", source: None | int | Callable = None, verbose: bool = False, interval=None, ): r""" Args: interrupts: {ARG_TRACKER_INTERRUPTS} filename (str, optional): Determines the file to which the data is written in JSON format method (str): Method used for determining the length scale. Details are explained in the function :func:`~droplets.image_analysis.get_length_scale`. source (int or callable, optional): Determines how a field is extracted from `fields`. If `None`, `fields` is passed as is, assuming it is already a scalar field. This works for the simple, standard case where only a single :class:`~pde.fields.scalar.ScalarField` is treated. Alternatively, `source` can be an integer, indicating which field is extracted from an instance of :class:`~pde.fields.collection.FieldCollection`. Lastly, `source` can be a function that takes `fields` as an argument and returns the desired field. verbose (bool): Determines whether errors in determining the length scales are logged. """ super().__init__(interrupts=interrupts, interval=interval) self.length_scales: list[float] = [] self.times: list[float] = [] self.filename = filename self.method = method self.source = source self.verbose = verbose
[docs] def handle(self, field: FieldBase, t: float): """handle data supplied to this tracker Args: field (:class:`~pde.fields.FieldBase`): The current state of the simulation t (float): The associated time """ # determine length scale from pde.visualization.plotting import extract_field from .image_analysis import get_length_scale scalar_field = extract_field(field, self.source, 0) try: length = get_length_scale(scalar_field, method=self.method) # type: ignore except Exception: if self.verbose: self._logger.exception("Could not determine length scale") length = math.nan # store data self.times.append(t) self.length_scales.append(length) # type: ignore
[docs] def finalize(self, info: InfoDict | None = None) -> None: """finalize the tracker, supplying additional information Args: info (dict): Extra information from the simulation """ super().finalize(info) if self.filename: import json data = {"times": self.times, "length_scales": self.length_scales} with open(self.filename, "w") as fp: json.dump(data, fp)
[docs] class DropletTracker(TrackerBase): """Detect droplets in a scalar field during simulations This tracker is useful when only the parameters of actual droplets are needed, since it stores considerably less information compared to the full scalar field. Attributes: data (:class:`~droplets.emulsions.EmulsionTimeCourse`): Contains the data of the tracked droplets after the simulation is done. """ @fill_in_docstring def __init__( self, interrupts: InterruptData = 1, filename: str | None = None, *, emulsion_timecourse=None, source: None | int | Callable = None, threshold: float | Literal["auto", "extrema", "mean", "otsu"] = 0.5, minimal_radius: float = 0, refine: bool = False, refine_args: dict[str, Any] | None = None, perturbation_modes: int = 0, interval=None, ): """ Example: To track droplets and determine their position, radii, and interfacial widths, the following tracker can be used .. code-block:: python droplet_tracker = DropletTracker( 1, refine=True, refine_args={'vmin': None, 'vmax': None} ) :code:`field` is the scalar field, in which the droplets are located. The `refine_args` set flexibel boundaries for the intensities inside and outside the droplet. Args: interrupts: {ARG_TRACKER_INTERRUPTS} filename (str, optional): Determines the file to which the final data is written as an HDF5 file. emulsion_timecourse (:class:`EmulsionTimeCourse`, optional): Can be an instance of :class:`~droplets.emulsions.EmulsionTimeCourse` that is used to store the data. source (int or callable, optional): Determines how a field is extracted from `fields`. If `None`, `fields` is passed as is, assuming it is already a scalar field. This works for the simple, standard case where only a single ScalarField is treated. Alternatively, `source` can be an integer, indicating which field is extracted from an instance of :class:`~pde.fields.FieldCollection`. Lastly, `source` can be a function that takes `fields` as an argument and returns the desired field. threshold (float or str): The threshold for binarizing the image. If a value is given it is used directly. Otherwise, the following algorithms are supported: * `extrema`: take mean between the minimum and the maximum of the data * `mean`: take the mean over the entire data * `otsu`: use Otsu's method implemented in :func:`threshold_otsu` The special value `auto` currently defaults to the `extrema` method. minimal_radius (float): Minimal radius of droplets that will be retained. refine (bool): Flag determining whether the droplet coordinates should be refined using fitting. This is a potentially slow procedure. refine_args (dict): Additional keyword arguments passed on to :func:`refine_droplet`. Only has an effect if `refine=True`. perturbation_modes (int): An option describing how many perturbation modes should be considered when refining droplets. Only has an effect if `refine=True`. """ super().__init__(interrupts=interrupts, interval=interval) if emulsion_timecourse is None: self.data = EmulsionTimeCourse() else: self.data = emulsion_timecourse self.filename = filename self.source = source self.threshold = threshold self.minimal_radius = minimal_radius self.refine = refine self.refine_args = refine_args self.perturbation_modes = perturbation_modes
[docs] def handle(self, field: FieldBase, t: float) -> None: """handle data supplied to this tracker Args: field (:class:`~pde.fields.base.FieldBase`): The current state of the simulation t (float): The associated time """ from pde.visualization.plotting import extract_field from .image_analysis import locate_droplets scalar_field = extract_field(field, self.source, 0) emulsion = locate_droplets( scalar_field, # type: ignore threshold=self.threshold, refine=self.refine, refine_args=self.refine_args, modes=self.perturbation_modes, minimal_radius=self.minimal_radius, ) self.data.append(emulsion, t)
[docs] def finalize(self, info: InfoDict | None = None) -> None: """finalize the tracker, supplying additional information Args: info (dict): Extra information from the simulation """ super().finalize(info) if self.filename: self.data.to_file(self.filename)