10691143420

Committed 03 Sep 2024 09:48PM UTC coverage: 76.813% (+0.8%) from 76.01%

Build # 10691143420

Build Type

Pull #35

github

Committed by

web-flow

Commit Message

Merge 1ebfe99fb into 81a2299af

Pull Request Pull Request #35: Bump actions/download-artifact from 2 to 4.1.7 in /.github/workflows

Run Details

1928 of 2510 relevant lines covered (76.81%)

10.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.71

/src/welltestpy/data/varlib.py

"""welltestpy subpackage providing flow datastructures for variables."""
import numbers
from copy import deepcopy as dcopy

import numpy as np

from . import data_io

__all__ = [
    "Variable",
    "TimeVar",
    "HeadVar",
    "TemporalVar",
    "CoordinatesVar",
    "Observation",
    "StdyObs",
    "DrawdownObs",
    "StdyHeadObs",
    "TimeSeries",
    "Well",
]


class Variable:
    """Class for a variable.

    This is a class for a physical variable which is either a scalar or an
    array.

    It has a name, a value, a symbol, a unit and a descrition string.

    Parameters
    ----------
    name : :class:`str`
        Name of the Variable.
    value : :class:`int` or :class:`float` or :class:`numpy.ndarray`
        Value of the Variable.
    symbole : :class:`str`, optional
        Name of the Variable. Default: ``"x"``
    units : :class:`str`, optional
        Units of the Variable. Default: ``"-"``
    description : :class:`str`, optional
        Description of the Variable. Default: ``"no description"``
    """

    def __init__(
        self, name, value, symbol="x", units="-", description="no description"
    ):
        self.name = data_io._formstr(name)
        self.__value = None
        self.value = value
        self.symbol = str(symbol)
        self.units = str(units)
        self.description = str(description)

    def __call__(self, value=None):
        """Call a variable.

        Here you can set a new value or you can get the value of the variable.

        Parameters
        ----------
        value : :class:`int` or :class:`float` or :class:`numpy.ndarray`,
        optional
            Value of the Variable. Default: ``None``

        Returns
        -------
        value : :class:`int` or :class:`float` or :class:`numpy.ndarray`
            Value of the Variable.
        """
        if value is not None:
            self.value = value
        return self.value

    @property
    def info(self):
        """:class:`str`: Info about the Variable."""
        info = ""
        info += " Variable-name: " + str(self.name) + "\n"
        info += " -Value:        " + str(self.value) + "\n"
        info += " -Symbol:       " + str(self.symbol) + "\n"
        info += " -Units:        " + str(self.units) + "\n"
        info += " -Description:  " + str(self.description) + "\n"
        return info

    @property
    def scalar(self):
        """:class:`bool`: State if the variable is of scalar type."""
        return np.isscalar(self.__value)

    @property
    def label(self):
        """:class:`str`: String containing: ``symbol in units``."""
        return f"{self.symbol} in {self.units}"

    @property
    def value(self):
        """:class:`int` or :class:`float` or :class:`numpy.ndarray`: Value."""
        return self.__value

    @value.setter
    def value(self, value):
        if issubclass(np.asanyarray(value).dtype.type, numbers.Real):
            if np.ndim(np.squeeze(value)) == 0:
                self.__value = float(np.squeeze(value))
            else:
                self.__value = np.squeeze(np.array(value, dtype=float))
        elif issubclass(np.asanyarray(value).dtype.type, numbers.Integral):
            if np.ndim(np.squeeze(value)) == 0:
                self.__value = int(np.squeeze(value))
            else:
                self.__value = np.squeeze(np.array(value, dtype=int))
        else:
            raise ValueError("Variable: 'value' is neither integer nor float")

    def __repr__(self):
        """Representation."""
        return f"{self.name} {self.symbol}: {self.value} {self.units}"

    def __str__(self):
        """Representation."""
        return f"{self.name} {self.label}"

    def save(self, path="", name=None):
        """Save a variable to file.

        This writes the variable to a csv file.

        Parameters
        ----------
        path : :class:`str`, optional
            Path where the variable should be saved. Default: ``""``
        name : :class:`str`, optional
            Name of the file. If ``None``, the name will be generated by
            ``"Var_"+name``. Default: ``None``

        Notes
        -----
        The file will get the suffix ``".var"``.
        """
        return data_io.save_var(self, path, name)


class TimeVar(Variable):
    """Variable class special for time series.

    Parameters
    ----------
    value : :class:`int` or :class:`float` or :class:`numpy.ndarray`
        Value of the Variable.
    symbole : :class:`str`, optional
        Name of the Variable. Default: ``"t"``
    units : :class:`str`, optional
        Units of the Variable. Default: ``"s"``
    description : :class:`str`, optional
        Description of the Variable. Default: ``"time given in seconds"``

    Notes
    -----
    Here the variable should be at most 1 dimensional and the name is fix set
    to ``"time"``.
    """

    def __init__(
        self, value, symbol="t", units="s", description="time given in seconds"
    ):
        super().__init__("time", value, symbol, units, description)
        if np.ndim(self.value) > 1:
            raise ValueError(
                "TimeVar: 'time' should have at most one dimension"
            )


class HeadVar(Variable):
    """
    Variable class special for groundwater head.

    Parameters
    ----------
    value : :class:`int` or :class:`float` or :class:`numpy.ndarray`
        Value of the Variable.
    symbole : :class:`str`, optional
        Name of the Variable. Default: ``"h"``
    units : :class:`str`, optional
        Units of the Variable. Default: ``"m"``
    description : :class:`str`, optional
        Description of the Variable. Default: ``"head given in meters"``

    Notes
    -----
    Here the variable name is fix set to ``"head"``.
    """

    def __init__(
        self, value, symbol="h", units="m", description="head given in meters"
    ):
        super().__init__("head", value, symbol, units, description)


class TemporalVar(Variable):
    """
    Variable class for a temporal variable.

    Parameters
    ----------
    value : :class:`int` or :class:`float` or :class:`numpy.ndarray`,
    optional
        Value of the Variable. Default: ``0.0``
    """

    def __init__(self, value=0.0):
        super().__init__("temporal", value, description="temporal variable")


class CoordinatesVar(Variable):
    """Variable class special for coordinates.

    Parameters
    ----------
    lat : :class:`int` or :class:`float` or :class:`numpy.ndarray`
        Lateral values of the coordinates.
    lon : :class:`int` or :class:`float` or :class:`numpy.ndarray`
        Longitutional values of the coordinates.
    symbole : :class:`str`, optional
        Name of the Variable. Default: ``"[Lat,Lon]"``
    units : :class:`str`, optional
        Units of the Variable. Default: ``"[deg,deg]"``
    description : :class:`str`, optional
        Description of the Variable. Default: ``"Coordinates given in
        degree-North and degree-East"``

    Notes
    -----
    Here the variable name is fix set to ``"coordinates"``.

    ``lat`` and ``lon`` should have the same shape.
    """

    def __init__(
        self,
        lat,
        lon,
        symbol="[Lat,Lon]",
        units="[deg,deg]",
        description="Coordinates given in degree-North and degree-East",
    ):
        ilat = np.array(np.squeeze(lat), ndmin=1)
        ilon = np.array(np.squeeze(lon), ndmin=1)

        if (
            len(ilat.shape) != 1
            or len(ilon.shape) != 1
            or ilat.shape != ilon.shape
        ):
            raise ValueError(
                "CoordinatesVar: 'lat' and 'lon' should have "
                "same quantity and should be given as lists"
            )

        value = np.array([ilat, ilon]).T

        super().__init__("coordinates", value, symbol, units, description)


class Observation:
    """
    Class for a observation.

    This is a class for time-dependent observations.
    It has a name and a description.

    Parameters
    ----------
    name : :class:`str`
        Name of the Variable.
    observation : :class:`Variable`
        Name of the Variable. Default: ``"x"``
    time : :class:`Variable`
        Value of the Variable.
    description : :class:`str`, optional
        Description of the Variable. Default: ``"Observation"``
    """

    def __init__(
        self, name, observation, time=None, description="Observation"
    ):
        self.__it = None
        self.__itfinished = None
        self._time = None
        self._observation = None
        self.name = data_io._formstr(name)
        self.description = str(description)

        self._setobservation(observation)
        self._settime(time)
        self._checkshape()

    def __call__(self, observation=None, time=None):
        """Call a variable.

        Here you can set a new value or you can get the value of the variable.

        Parameters
        ----------
        observation : scalar, :class:`numpy.ndarray`, :class:`Variable`, optional
            New Value for observation.
            Default: ``"None"``
        time : scalar, :class:`numpy.ndarray`, :class:`Variable`, optional
            New Value for time.
            Default: ``"None"``

        Returns
        -------
        [:class:`tuple` of] :class:`int` or :class:`float`
        or :class:`numpy.ndarray`
            ``(time, observation)`` or ``observation``.
        """
        if observation is not None:
            self._setobservation(observation)
        if time is not None:
            self._settime(time)
        if observation is not None or time is not None:
            self._checkshape()
        return self.value

    def __repr__(self):
        """Representation."""
        return f"Observation '{self.name}' {self.label}"

    def __str__(self):
        """Representation."""
        return self.__repr__()

    @property
    def labels(self):
        """[:class:`tuple` of] :class:`str`: ``symbol in units``."""
        if self.state == "transient":
            return self._time.label, self._observation.label
        return self._observation.label

    @property
    def label(self):
        """[:class:`tuple` of] :class:`str`: ``symbol in units``."""
        return self.labels

    @property
    def info(self):
        """Get information about the observation.

        Here you can display information about the observation.
        """
        info = ""
        info += "Observation-name: " + str(self.name) + "\n"
        info += " -Description:    " + str(self.description) + "\n"
        info += " -Kind:           " + str(self.kind) + "\n"
        info += " -State:          " + str(self.state) + "\n"
        if self.state == "transient":
            info += " --- \n"
            info += self._time.info + "\n"
        info += " --- \n"
        info += self._observation.info + "\n"
        return info

    @property
    def value(self):
        """
        Value of the Observation.

        [:class:`tuple` of] :class:`int` or :class:`float`
        or :class:`numpy.ndarray`
        """
        if self.state == "transient":
            return self.observation, self.time
        return self.observation

    @property
    def state(self):
        """
        :class:`str`: String containing state of the observation.

        Either ``"steady"`` or ``"transient"``.
        """
        return "steady" if self._time is None else "transient"

    @property
    def kind(self):
        """:class:`str`: name of the observation variable."""
        return self._observation.name

    @property
    def time(self):
        """
        Time values of the observation.

        :class:`int` or :class:`float` or :class:`numpy.ndarray`
        """
        return self._time.value if self.state == "transient" else None

    @time.setter
    def time(self, time):
        self._settime(time)
        self._checkshape()

    @time.deleter
    def time(self):
        self._time = None

    @property
    def observation(self):
        """
        Observed values of the observation.

        :class:`int` or :class:`float` or :class:`numpy.ndarray`
        """
        return self._observation.value

    @observation.setter
    def observation(self, observation):
        self._setobservation(observation)
        self._checkshape()

    @property
    def units(self):
        """[:class:`tuple` of] :class:`str`: units of the observation."""
        if self.state == "steady":
            return self._observation.units
        return f"{self._time.units}, {self._observation.units}"

    def reshape(self):
        """Reshape observations to flat array."""
        if self.state == "transient":
            tmp = len(np.shape(self.time))
            self._settime(np.reshape(self.time, -1))
            shp = np.shape(self.time) + np.shape(self.observation)[tmp:]
            self._setobservation(np.reshape(self.observation, shp))

    def _settime(self, time):
        if isinstance(time, Variable):
            self._time = dcopy(time)
        elif time is None:
            self._time = None
        elif self._time is None:
            self._time = TimeVar(time)
        else:
            self._time(time)

    def _setobservation(self, observation):
        if isinstance(observation, Variable):
            self._observation = dcopy(observation)
        elif observation is None:
            self._observation = None
        else:
            self._observation(observation)

    def _checkshape(self):
        if self.state == "transient" and (
            np.shape(self.time)
            != np.shape(self.observation)[: len(np.shape(self.time))]
        ):
            raise ValueError(
                "Observation: 'observation' has a shape-mismatch with 'time'"
            )

    def __iter__(self):
        """Iterate over Observations."""
        if self.state == "transient":
            self.__it = np.nditer(self.time, flags=["multi_index"])
        else:
            self.__itfinished = False
        return self

    def __next__(self):
        """Iterate through observations."""
        if self.state == "transient":
            if self.__it.finished:
                raise StopIteration
            ret = (
                self.__it[0].item(),
                self.observation[self.__it.multi_index],
            )
            self.__it.iternext()
        else:
            if self.__itfinished:
                raise StopIteration
            ret = self.observation
            self.__itfinished = True
        return ret

    def save(self, path="", name=None):
        """Save an observation to file.

        This writes the observation to a csv file.

        Parameters
        ----------
        path : :class:`str`, optional
            Path where the variable should be saved. Default: ``""``
        name : :class:`str`, optional
            Name of the file. If ``None``, the name will be generated by
            ``"Obs_"+name``. Default: ``None``

        Notes
        -----
        The file will get the suffix ``".obs"``.
        """
        return data_io.save_obs(self, path, name)


class StdyObs(Observation):
    """
    Observation class special for steady observations.

    Parameters
    ----------
    name : :class:`str`
        Name of the Variable.
    observation : :class:`Variable`
        Name of the Variable. Default: ``"x"``
    description : :class:`str`, optional
        Description of the Variable. Default: ``"Steady observation"``
    """

    def __init__(self, name, observation, description="Steady observation"):
        super().__init__(name, observation, None, description)

    def _settime(self, time):
        """For steady observations, this raises a ``ValueError``."""
        if time is not None:
            raise ValueError("Observation: 'time' not allowed in steady-state")


class TimeSeries(Observation):
    """
    Time series observation.

    Parameters
    ----------
    name : :class:`str`
        Name of the Variable.
    values : :class:`Variable`
        Values of the time-series.
    time : :class:`Variable`
        Time points of the time-series.
    description : :class:`str`, optional
        Description of the Variable. Default: ``"Timeseries."``
    """

    def __init__(self, name, values, time, description="Timeseries."):
        if not isinstance(time, Variable):
            time = TimeVar(time)
        if not isinstance(values, Variable):
            values = Variable(name, values, description=description)
        super().__init__(name, values, time, description)


class DrawdownObs(Observation):
    """
    Observation class special for drawdown observations.

    Parameters
    ----------
    name : :class:`str`
        Name of the Variable.
    observation : :class:`Variable`
        Observation.
    time : :class:`Variable`
        Time points of observation.
    description : :class:`str`, optional
        Description of the Variable. Default: ``"Drawdown observation"``
    """

    def __init__(
        self, name, observation, time, description="Drawdown observation"
    ):
        if not isinstance(time, Variable):
            time = TimeVar(time)
        if not isinstance(observation, Variable):
            observation = HeadVar(observation)
        super().__init__(name, observation, time, description)


class StdyHeadObs(Observation):
    """
    Observation class special for steady drawdown observations.

    Parameters
    ----------
    name : :class:`str`
        Name of the Variable.
    observation : :class:`Variable`
        Observation.
    description : :class:`str`, optional
        Description of the Variable. Default: ``"Steady observation"``
    """

    def __init__(
        self,
        name,
        observation,
        description="Steady State Drawdown observation",
    ):
        if not isinstance(observation, Variable):
            observation = HeadVar(observation)
        super().__init__(name, observation, None, description)

    def _settime(self, time):
        """For steady observations, this raises a ``ValueError``."""
        if time is not None:
            raise ValueError("Observation: 'time' not allowed in steady-state")


class Well:
    """Class for a pumping-/observation-well.

    This is a class for a well within a aquifer-testing campaign.

    It has a name, a radius, coordinates and a depth.

    Parameters
    ----------
    name : :class:`str`
        Name of the Variable.
    radius : :class:`Variable` or :class:`float`
        Value of the Variable.
    coordinates : :class:`Variable` or :class:`numpy.ndarray`
        Value of the Variable.
    welldepth : :class:`Variable` or :class:`float`, optional
        Depth of the well (in saturated zone). Default: 1.0
    aquiferdepth : :class:`Variable` or :class:`float`, optional
        Aquifer depth at the well (saturated zone). Defaults to welldepth.
        Default: ``"None"``
    screensize : :class:`Variable` or :class:`float`, optional
        Size of the screen at the well. Defaults to 0.0.
        Default: ``"None"``

    Notes
    -----
    You can calculate the distance between two wells ``w1`` and ``w2`` by
    simply calculating the difference ``w1 - w2``.
    """

    def __init__(
        self,
        name,
        radius,
        coordinates,
        welldepth=1.0,
        aquiferdepth=None,
        screensize=None,
    ):
        self._radius = None
        self._coordinates = None
        self._welldepth = None
        self._aquiferdepth = None
        self._screensize = None

        self.name = data_io._formstr(name)
        self.wellradius = radius
        self.coordinates = coordinates
        self.welldepth = welldepth
        self.aquiferdepth = aquiferdepth
        self.screensize = screensize

    @property
    def info(self):
        """Get information about the variable.

        Here you can display information about the variable.
        """
        info = ""
        info += "----\n"
        info += "Well-name: " + str(self.name) + "\n"
        info += "--\n"
        info += self._radius.info + "\n"
        info += self.coordinates.info + "\n"
        info += self._welldepth.info + "\n"
        info += self._aquiferdepth.info + "\n"
        info += self._screensize.info + "\n"
        info += "----\n"
        return info

    @property
    def radius(self):
        """:class:`float`: Radius of the well."""
        return self._radius.value

    @property
    def wellradius(self):
        """:class:`Variable`: Radius variable of the well."""
        return self._radius

    @wellradius.setter
    def wellradius(self, radius):
        if isinstance(radius, Variable):
            self._radius = dcopy(radius)
        elif self._radius is None:
            self._radius = Variable(
                "radius",
                float(radius),
                "r",
                "m",
                f"Inner radius of well '{self.name}'",
            )
        else:
            self._radius(radius)
        if not self._radius.scalar:
            raise ValueError("Well: 'radius' needs to be scalar")
        if not self.radius > 0.0:
            raise ValueError("Well: 'radius' needs to be positive")

    @property
    def pos(self):
        """:class:`numpy.ndarray`: Position of the well."""
        return self._coordinates.value

    @property
    def coordinates(self):
        """:class:`Variable`: Coordinates variable of the well."""
        return self._coordinates

    @coordinates.setter
    def coordinates(self, coordinates):
        if isinstance(coordinates, Variable):
            self._coordinates = dcopy(coordinates)
        elif self._coordinates is None:
            self._coordinates = Variable(
                "coordinates",
                coordinates,
                "XY",
                "m",
                f"coordinates of well '{self.name}'",
            )
        else:
            self._coordinates(coordinates)
        if np.shape(self.pos) != (2,) and not np.isscalar(self.pos):
            raise ValueError(
                "Well: 'coordinates' should be given as "
                "[x,y] values or one single distance value"
            )

    @property
    def depth(self):
        """:class:`float`: Depth of the well."""
        return self._welldepth.value

    @property
    def welldepth(self):
        """:class:`Variable`: Depth variable of the well."""
        return self._welldepth

    @welldepth.setter
    def welldepth(self, welldepth):
        if isinstance(welldepth, Variable):
            self._welldepth = dcopy(welldepth)
        elif self._welldepth is None:
            self._welldepth = Variable(
                "welldepth",
                float(welldepth),
                "L_w",
                "m",
                f"depth of well '{self.name}'",
            )
        else:
            self._welldepth(welldepth)
        if not self._welldepth.scalar:
            raise ValueError("Well: 'welldepth' needs to be scalar")
        if not self.depth > 0.0:
            raise ValueError("Well: 'welldepth' needs to be positive")

    @property
    def aquifer(self):
        """:class:`float`: Aquifer depth at the well."""
        return self._aquiferdepth.value

    @property
    def aquiferdepth(self):
        """:class:`Variable`: Aquifer depth at the well."""
        return self._aquiferdepth

    @aquiferdepth.setter
    def aquiferdepth(self, aquiferdepth):
        if isinstance(aquiferdepth, Variable):
            self._aquiferdepth = dcopy(aquiferdepth)
        elif self._aquiferdepth is None:
            self._aquiferdepth = Variable(
                "aquiferdepth",
                self.depth if aquiferdepth is None else float(aquiferdepth),
                "L_a",
                self.welldepth.units,
                f"aquifer depth at well '{self.name}'",
            )
        else:
            self._aquiferdepth(aquiferdepth)
        if not self._aquiferdepth.scalar:
            raise ValueError("Well: 'aquiferdepth' needs to be scalar")
        if not self.aquifer > 0.0:
            raise ValueError("Well: 'aquiferdepth' needs to be positive")

    @property
    def is_piezometer(self):
        """:class:`bool`: Whether the well is only a standpipe piezometer."""
        return np.isclose(self.screen, 0)

    @property
    def screen(self):
        """:class:`float`: Screen size at the well."""
        return self._screensize.value

    @property
    def screensize(self):
        """:class:`Variable`: Screen size at the well."""
        return self._screensize

    @screensize.setter
    def screensize(self, screensize):
        if isinstance(screensize, Variable):
            self._screensize = dcopy(screensize)
        elif self._screensize is None:
            self._screensize = Variable(
                "screensize",
                0.0 if screensize is None else float(screensize),
                "L_s",
                self.welldepth.units,
                f"screen size at well '{self.name}'",
            )
        else:
            self._screensize(screensize)
        if not self._screensize.scalar:
            raise ValueError("Well: 'screensize' needs to be scalar")
        if self.screen < 0.0:
            raise ValueError("Well: 'screensize' needs to be non-negative")

    def distance(self, well):
        """Calculate distance to the well.

        Parameters
        ----------
        well : :class:`Well` or :class:`tuple` of :class:`float`
            Coordinates to calculate the distance to or another well.
        """
        if isinstance(well, Well):
            return np.linalg.norm(self.pos - well.pos)
        try:
            return np.linalg.norm(self.pos - well)
        except ValueError:
            raise ValueError(
                "Well: the distant-well needs to be an "
                "instance of Well-class "
                "or a tuple of x-y coordinates "
                "or a single distance value "
                "and of same coordinates-type."
            )

    def __repr__(self):
        """Representation."""
        return f"{self.name} r={self.radius} at {self._coordinates}"

    def __sub__(self, well):
        """Distance between wells."""
        return self.distance(well)

    def __add__(self, well):
        """Distance between wells."""
        return self.distance(well)

    def __and__(self, well):
        """Distance between wells."""
        return self.distance(well)

    def __abs__(self):
        """Distance to origin."""
        return np.linalg.norm(self.pos)

    def save(self, path="", name=None):
        """Save a well to file.

        This writes the variable to a csv file.

        Parameters
        ----------
        path : :class:`str`, optional
            Path where the variable should be saved. Default: ``""``
        name : :class:`str`, optional
            Name of the file. If ``None``, the name will be generated by
            ``"Well_"+name``. Default: ``None``

        Notes
        -----
        The file will get the suffix ``".wel"``.
        """
        return data_io.save_well(self, path, name)

GeoStat-Framework / welltestpy / 10691143420

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous