21579147278

Committed 02 Feb 2026 05:59AM UTC coverage: 65.805% (+0.2%) from 65.635%

Build Type

push

github

Committed by Bernhard Haas

Commit Message

Pinned and updated versions in pyproject.toml.

Run Details

1705 of 2591 relevant lines covered (65.8%)

1.32 hits per line

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

81.01

/el_paso/variable.py

# SPDX-FileCopyrightText: 2025 GFZ Helmholtz Centre for Geosciences
# SPDX-FileContributor: Bernhard Haas
#
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations

import typing
from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal, overload

import numpy as np
from astropy import units as u  # type: ignore[reportMissingTypeStubs]

import el_paso as ep
from el_paso.utils import enforce_utc_timezone

if typing.TYPE_CHECKING:
    from numpy.typing import NDArray


@dataclass
class VariableMetadata:
    """A class holding the metadata of a variable.

    Attributes:
        unit (u.UnitBase): The unit of the variable. Defaults to
            `u.dimensionless_unscaled`.
        original_cadence_seconds (float): The original cadence of the data in seconds.
            Defaults to 0.
        source_files (list[str]): The list of SourceFiles, which variable contains
            data from. Defaults to an empty list.
        description (str): The description of the variable explaining what kind of data
            this variable contains. Defaults to "".
        processing_notes (str): The processing notes of the variable explaining all
            steps done to achieve the final result. Defaults to "".
        standard_name (str): The name of the standard variable this variable complies
            to. Defaults to "".
    """

    unit: u.UnitBase = u.dimensionless_unscaled
    original_cadence_seconds: float = 0
    source_files: list[str] = field(default_factory=list[str])
    description: str = ""
    processing_notes: str = ""
    standard_name: str = ""

    def __post_init__(self) -> None:
        """Initializes the processing_steps_counter attribute to 1 after the dataclass has been instantiated.

        This method is automatically called by the dataclass after the __init__ method.
        """
        self.processing_steps_counter = 1

        if ep.is_in_release_mode():
            self.processing_notes += ep.get_release_msg() + "\n"


    def add_processing_note(self, processing_note: str) -> None:
        """Adds a processing note to the metadata.

        The note is prefixed with the current processing steps counter and a newline
        character is appended. The processing steps counter is then incremented.

        Args:
            processing_note (str): The note to be added to the processing notes.
        """
        processing_note = f"{self.processing_steps_counter}) {processing_note}\n"

        self.processing_notes += processing_note
        self.processing_steps_counter += 1


class Variable:
    """Variable class holding data and metadata.

    Attributes:
        _data (NDArray[np.generic]): The numerical data of the variable.
        metadata (VariableMetadata): An instance of `VariableMetadata` holding
            information about the variable.
    """

    __slots__ = "_data", "metadata"

    _data: NDArray[np.generic]
    metadata: VariableMetadata

    def __init__(
        self,
        original_unit: u.UnitBase,
        data: NDArray[np.generic] | None = None,
        description: str = "",
        processing_notes: str = "",
    ) -> None:
        """Initializes a Variable instance.

        Args:
            original_unit (u.UnitBase): The original unit of the data.
            data (NDArray[np.generic] | None): The numerical data. Defaults to an empty
                numpy array if None.
            description (str): A description of the variable. Defaults to "".
            processing_notes (str): Notes on how the data was processed. Defaults to "".
        """
        self._data = np.array([]) if data is None else data

        self.metadata = VariableMetadata(
            unit=original_unit,
            description=description,
            processing_notes=processing_notes,
        )

    def __repr__(self) -> str:
        """Returns a string representation of the Variable object."""
        return f"Variable holding {self._data.shape} data points with metadata: {self.metadata}"

    def convert_to_unit(self, target_unit: u.UnitBase | str) -> None:
        """Converts the data to a given unit.

        Args:
            target_unit (u.UnitBase | str): The unit the data should be converted to.
        """
        if isinstance(target_unit, str):
            target_unit = u.Unit(target_unit)

        if self.metadata.unit != target_unit:
            data_with_unit = u.Quantity(self._data, self.metadata.unit)
            self._data = typing.cast("NDArray[np.generic]", data_with_unit.to_value(target_unit))  # type: ignore[reportUnknownMemberType]

            self.metadata.unit = target_unit

    @overload
    def get_data(self, target_unit: u.UnitBase | str) -> NDArray[np.floating | np.integer]: ...

    @overload
    def get_data(self, target_unit: None = None) -> NDArray[np.generic]: ...

    def get_data(self, target_unit: u.UnitBase | str | None = None) -> NDArray[np.generic]:
        """Gets the data of the variable.

        Args:
            target_unit (u.UnitBase | str | None): The unit to convert the data to
                before returning. If None, the data is returned in its current unit.
                Defaults to None.

        Returns:
            NDArray[np.generic]: The data of the variable.

        Raises:
            TypeError: If `target_unit` is provided and the data is not numeric.
        """
        if target_unit is None:
            return self._data

        if isinstance(target_unit, str):
            target_unit = u.Unit(target_unit)

        if not np.issubdtype(self._data.dtype, np.number):
            msg = f"Unit conversion is only supported for numeric types! Encountered for variable {self}."
            raise TypeError(msg)

        return typing.cast("NDArray[np.generic]", u.Quantity(self._data, self.metadata.unit).to_value(target_unit))  # type: ignore[reportUnknownMemberType]

    def set_data(self, data: NDArray[np.generic], unit: Literal["same"] | str | u.UnitBase) -> None:  # noqa: PYI051
        """Sets the data and optionally updates the unit of the variable.

        Args:
            data (NDArray[np.generic]): The new data array.
            unit (Literal["same"] | str | u.UnitBase): The unit of the new data.
                If "same", the existing unit is kept. Can be a string representation
                of a unit or an `astropy.units.UnitBase` object.

        Raises:
            TypeError: If `unit` is not "same", a string, or an `astropy.units.UnitBase` object.
        """
        self._data = data

        if isinstance(unit, str):
            if unit != "same":
                self.metadata.unit = u.Unit(unit)
        elif isinstance(unit, u.UnitBase):  # type: ignore[reportUnknownMemberType]
            self.metadata.unit = unit
        else:
            msg = "unit must be either a str or a astropy unit!"
            raise TypeError(msg)

    def transpose_data(self, seq: list[int] | tuple[int, ...]) -> None:
        """Transposes the internal data array.

        Args:
            seq (list[int] | tuple[int, ...]): The axes to transpose to. See
                `numpy.transpose` for details.
        """
        self._data = np.transpose(self._data, axes=seq)

    def apply_thresholds_on_data(self, lower_threshold: float = -np.inf, upper_threshold: float = np.inf) -> None:
        """Applies lower and upper thresholds to the data.

        Values outside the thresholds (exclusive) are set to NaN.

        Args:
            lower_threshold (float): The lower bound for the data. Defaults to
                negative infinity.
            upper_threshold (float): The upper bound for the data. Defaults to
                positive infinity.

        Raises:
            TypeError: If the data is not numeric.
        """
        if not np.issubdtype(self._data.dtype, np.number):
            msg = f"Thresholds are only supported for numeric types! Encountered for variable {self}."
            raise TypeError(msg)
        self._data = typing.cast("NDArray[np.number]", self._data)

        self._data = np.where((self._data > lower_threshold) & (self._data < upper_threshold), self._data, np.nan)

    def truncate(self, time_variable: Variable, start_time: float | datetime, end_time: float | datetime) -> None:
        """Truncates the variable's data based on a time variable and a time range.

        Args:
            time_variable (Variable): A `Variable` object containing the time data.
            start_time (float | datetime): The start time for truncation. Can be a
                Unix timestamp (float) or a `datetime` object.
            end_time (float | datetime): The end time for truncation. Can be a
                Unix timestamp (float) or a `datetime` object.

        Raises:
            ValueError: If the length of the variable's data does not match the
                length of the `time_variable`'s data.
        """
        if isinstance(start_time, datetime):
            start_time = enforce_utc_timezone(start_time).timestamp()
        if isinstance(end_time, datetime):
            end_time = enforce_utc_timezone(end_time).timestamp()

        if self._data.shape[0] != time_variable.get_data().shape[0]:
            msg = f"Encountered length missmatch between variable and time variable! Variable: {self}"
            raise ValueError(msg)

        time_var_data = time_variable.get_data(ep.units.posixtime)

        self._data = self._data[(time_var_data >= start_time) & (time_var_data <= end_time)]

    def __hash__(self) -> int:
        """Computes a hash value for the variable based on its holding data.

        Returns:
            int: The integer hash value.
        """
        return hash(self._data.tobytes())

1	# SPDX-FileCopyrightText: 2025 GFZ Helmholtz Centre for Geosciences
2	# SPDX-FileContributor: Bernhard Haas
3	#
4	# SPDX-License-Identifier: Apache-2.0
5
6	from __future__ import annotations	2✔
7
8	import typing	2✔
9	from dataclasses import dataclass, field	2✔
10	from datetime import datetime	2✔
11	from typing import Literal, overload	2✔
12
13	import numpy as np	2✔
14	from astropy import units as u # type: ignore[reportMissingTypeStubs]	2✔
15
16	import el_paso as ep	2✔
17	from el_paso.utils import enforce_utc_timezone	2✔
18
19	if typing.TYPE_CHECKING:
20	from numpy.typing import NDArray
21
22
23	@dataclass	2✔
24	class VariableMetadata:	2✔
25	"""A class holding the metadata of a variable.
26
27	Attributes:
28	unit (u.UnitBase): The unit of the variable. Defaults to
29	`u.dimensionless_unscaled`.
30	original_cadence_seconds (float): The original cadence of the data in seconds.
31	Defaults to 0.
32	source_files (list[str]): The list of SourceFiles, which variable contains
33	data from. Defaults to an empty list.
34	description (str): The description of the variable explaining what kind of data
35	this variable contains. Defaults to "".
36	processing_notes (str): The processing notes of the variable explaining all
37	steps done to achieve the final result. Defaults to "".
38	standard_name (str): The name of the standard variable this variable complies
39	to. Defaults to "".
40	"""
41
42	unit: u.UnitBase = u.dimensionless_unscaled	2✔
43	original_cadence_seconds: float = 0	2✔
44	source_files: list[str] = field(default_factory=list[str])	2✔
45	description: str = ""	2✔
46	processing_notes: str = ""	2✔
47	standard_name: str = ""	2✔
48
49	def __post_init__(self) -> None:	2✔
50	"""Initializes the processing_steps_counter attribute to 1 after the dataclass has been instantiated.
51
52	This method is automatically called by the dataclass after the __init__ method.
53	"""
54	self.processing_steps_counter = 1	2✔
55
56	if ep.is_in_release_mode():	2✔
57	self.processing_notes += ep.get_release_msg() + "\n"	×
58
59
60	def add_processing_note(self, processing_note: str) -> None:	2✔
61	"""Adds a processing note to the metadata.
62
63	The note is prefixed with the current processing steps counter and a newline
64	character is appended. The processing steps counter is then incremented.
65
66	Args:
67	processing_note (str): The note to be added to the processing notes.
68	"""
69	processing_note = f"{self.processing_steps_counter}) {processing_note}\n"	2✔
70
71	self.processing_notes += processing_note	2✔
72	self.processing_steps_counter += 1	2✔
73
74
75	class Variable:	2✔
76	"""Variable class holding data and metadata.
77
78	Attributes:
79	_data (NDArray[np.generic]): The numerical data of the variable.
80	metadata (VariableMetadata): An instance of `VariableMetadata` holding
81	information about the variable.
82	"""
83
84	__slots__ = "_data", "metadata"	2✔
85
86	_data: NDArray[np.generic]	2✔
87	metadata: VariableMetadata	2✔
88
89	def __init__(	2✔
90	self,
91	original_unit: u.UnitBase,
92	data: NDArray[np.generic] \| None = None,
93	description: str = "",
94	processing_notes: str = "",
95	) -> None:
96	"""Initializes a Variable instance.
97
98	Args:
99	original_unit (u.UnitBase): The original unit of the data.
100	data (NDArray[np.generic] \| None): The numerical data. Defaults to an empty
101	numpy array if None.
102	description (str): A description of the variable. Defaults to "".
103	processing_notes (str): Notes on how the data was processed. Defaults to "".
104	"""
105	self._data = np.array([]) if data is None else data	2✔
106
107	self.metadata = VariableMetadata(	2✔
108	unit=original_unit,
109	description=description,
110	processing_notes=processing_notes,
111	)
112
113	def __repr__(self) -> str:	2✔
114	"""Returns a string representation of the Variable object."""
115	return f"Variable holding {self._data.shape} data points with metadata: {self.metadata}"	×
116
117	def convert_to_unit(self, target_unit: u.UnitBase \| str) -> None:	2✔
118	"""Converts the data to a given unit.
119
120	Args:
121	target_unit (u.UnitBase \| str): The unit the data should be converted to.
122	"""
123	if isinstance(target_unit, str):	2✔
124	target_unit = u.Unit(target_unit)	×
125
126	if self.metadata.unit != target_unit:	2✔
127	data_with_unit = u.Quantity(self._data, self.metadata.unit)	2✔
128	self._data = typing.cast("NDArray[np.generic]", data_with_unit.to_value(target_unit)) # type: ignore[reportUnknownMemberType]	2✔
129
130	self.metadata.unit = target_unit	2✔
131
132	@overload
133	def get_data(self, target_unit: u.UnitBase \| str) -> NDArray[np.floating \| np.integer]: ...
134
135	@overload
136	def get_data(self, target_unit: None = None) -> NDArray[np.generic]: ...
137
138	def get_data(self, target_unit: u.UnitBase \| str \| None = None) -> NDArray[np.generic]:	2✔
139	"""Gets the data of the variable.
140
141	Args:
142	target_unit (u.UnitBase \| str \| None): The unit to convert the data to
143	before returning. If None, the data is returned in its current unit.
144	Defaults to None.
145
146	Returns:
147	NDArray[np.generic]: The data of the variable.
148
149	Raises:
150	TypeError: If `target_unit` is provided and the data is not numeric.
151	"""
152	if target_unit is None:	2✔
153	return self._data	2✔
154
155	if isinstance(target_unit, str):	2✔
156	target_unit = u.Unit(target_unit)	×
157
158	if not np.issubdtype(self._data.dtype, np.number):	2✔
159	msg = f"Unit conversion is only supported for numeric types! Encountered for variable {self}."	×
160	raise TypeError(msg)	×
161
162	return typing.cast("NDArray[np.generic]", u.Quantity(self._data, self.metadata.unit).to_value(target_unit)) # type: ignore[reportUnknownMemberType]	2✔
163
164	def set_data(self, data: NDArray[np.generic], unit: Literal["same"] \| str \| u.UnitBase) -> None: # noqa: PYI051	2✔
165	"""Sets the data and optionally updates the unit of the variable.
166
167	Args:
168	data (NDArray[np.generic]): The new data array.
169	unit (Literal["same"] \| str \| u.UnitBase): The unit of the new data.
170	If "same", the existing unit is kept. Can be a string representation
171	of a unit or an `astropy.units.UnitBase` object.
172
173	Raises:
174	TypeError: If `unit` is not "same", a string, or an `astropy.units.UnitBase` object.
175	"""
176	self._data = data	2✔
177
178	if isinstance(unit, str):	2✔
179	if unit != "same":	2✔
180	self.metadata.unit = u.Unit(unit)	×
181	elif isinstance(unit, u.UnitBase): # type: ignore[reportUnknownMemberType]	2✔
182	self.metadata.unit = unit	2✔
183	else:
184	msg = "unit must be either a str or a astropy unit!"	×
185	raise TypeError(msg)	×
186
187	def transpose_data(self, seq: list[int] \| tuple[int, ...]) -> None:	2✔
188	"""Transposes the internal data array.
189
190	Args:
191	seq (list[int] \| tuple[int, ...]): The axes to transpose to. See
192	`numpy.transpose` for details.
193	"""
194	self._data = np.transpose(self._data, axes=seq)	2✔
195
196	def apply_thresholds_on_data(self, lower_threshold: float = -np.inf, upper_threshold: float = np.inf) -> None:	2✔
197	"""Applies lower and upper thresholds to the data.
198
199	Values outside the thresholds (exclusive) are set to NaN.
200
201	Args:
202	lower_threshold (float): The lower bound for the data. Defaults to
203	negative infinity.
204	upper_threshold (float): The upper bound for the data. Defaults to
205	positive infinity.
206
207	Raises:
208	TypeError: If the data is not numeric.
209	"""
210	if not np.issubdtype(self._data.dtype, np.number):	2✔
211	msg = f"Thresholds are only supported for numeric types! Encountered for variable {self}."	×
212	raise TypeError(msg)	×
213	self._data = typing.cast("NDArray[np.number]", self._data)	2✔
214
215	self._data = np.where((self._data > lower_threshold) & (self._data < upper_threshold), self._data, np.nan)	2✔
216
217	def truncate(self, time_variable: Variable, start_time: float \| datetime, end_time: float \| datetime) -> None:	2✔
218	"""Truncates the variable's data based on a time variable and a time range.
219
220	Args:
221	time_variable (Variable): A `Variable` object containing the time data.
222	start_time (float \| datetime): The start time for truncation. Can be a
223	Unix timestamp (float) or a `datetime` object.
224	end_time (float \| datetime): The end time for truncation. Can be a
225	Unix timestamp (float) or a `datetime` object.
226
227	Raises:
228	ValueError: If the length of the variable's data does not match the
229	length of the `time_variable`'s data.
230	"""
231	if isinstance(start_time, datetime):	2✔
232	start_time = enforce_utc_timezone(start_time).timestamp()	×
233	if isinstance(end_time, datetime):	2✔
234	end_time = enforce_utc_timezone(end_time).timestamp()	×
235
236	if self._data.shape[0] != time_variable.get_data().shape[0]:	2✔
237	msg = f"Encountered length missmatch between variable and time variable! Variable: {self}"	×
238	raise ValueError(msg)	×
239
240	time_var_data = time_variable.get_data(ep.units.posixtime)	2✔
241
242	self._data = self._data[(time_var_data >= start_time) & (time_var_data <= end_time)]	2✔
243
244	def __hash__(self) -> int:	2✔
245	"""Computes a hash value for the variable based on its holding data.
246
247	Returns:
248	int: The integer hash value.
249	"""
250	return hash(self._data.tobytes())	2✔

GFZ / EL_PASO / 21579147278

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