Variable

el_paso.variable.Variable

Variable class holding data and metadata.

Attributes:

Name	Type	Description
_data	NDArray[generic]	The numerical data of the variable.
metadata	VariableMetadata	An instance of VariableMetadata holding information about the variable.

Source code in el_paso/variable.py

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 = "",
        standard_name: StandardName = "",  # ty:ignore[invalid-parameter-default]
    ) -> 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 "".
            standard_name (StandardName): The standard name of the variable. Defaults to "".
        """
        self._data = np.array([]) if data is None else data

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

    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))

            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))

    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):
            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_mask(self, mask: NDArray[np.bool_]) -> None:
        """Applies a boolean mask to the data.

        Elements where the mask is False are invalidated by setting them to NaN.

        Args:
            mask (NDArray[np.bool_]): Boolean array of the same shape as the data.
                False indicates values to be masked.

        Raises:
            TypeError: If the data is not a floating-point numeric type.
        """
        if not np.issubdtype(self._data.dtype, np.floating):
            msg = f"Masking is only supported for floating-point types! Encountered for variable {self}."
            raise TypeError(msg)

        self._data[~mask] = np.nan

    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 a floating-point numeric type.
        """
        if not np.issubdtype(self._data.dtype, np.floating):
            msg = f"Thresholds are only supported for floating-point types! Encountered for variable {self}."
            raise TypeError(msg)
        self._data = typing.cast("NDArray[np.floating]", 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())

Methods:

el_paso.variable.Variable.__hash__

__hash__

Computes a hash value for the variable based on its holding data.

Returns:

Name	Type	Description
int	int	The integer hash value.

Source code in el_paso/variable.py

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())

el_paso.variable.Variable.__init__

__init__

Initializes a Variable instance.

Parameters:

Name	Type	Description	Default
original_unit	UnitBase	The original unit of the data.	required
data	NDArray[generic] | None	The numerical data. Defaults to an empty numpy array if None.	None
description	str	A description of the variable. Defaults to "".	''
processing_notes	str	Notes on how the data was processed. Defaults to "".	''
standard_name	StandardName	The standard name of the variable. Defaults to "".	''

Source code in el_paso/variable.py

def __init__(
    self,
    original_unit: u.UnitBase,
    data: NDArray[np.generic] | None = None,
    description: str = "",
    processing_notes: str = "",
    standard_name: StandardName = "",  # ty:ignore[invalid-parameter-default]
) -> 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 "".
        standard_name (StandardName): The standard name of the variable. Defaults to "".
    """
    self._data = np.array([]) if data is None else data

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

el_paso.variable.Variable.__repr__

__repr__

Returns a string representation of the Variable object.

Source code in el_paso/variable.py

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}"

el_paso.variable.Variable.apply_mask

apply_mask

Applies a boolean mask to the data.

Elements where the mask is False are invalidated by setting them to NaN.

Parameters:

Name	Type	Description	Default
mask	NDArray[bool_]	Boolean array of the same shape as the data. False indicates values to be masked.	required

Raises:

Type	Description
TypeError	If the data is not a floating-point numeric type.

Source code in el_paso/variable.py

def apply_mask(self, mask: NDArray[np.bool_]) -> None:
    """Applies a boolean mask to the data.

    Elements where the mask is False are invalidated by setting them to NaN.

    Args:
        mask (NDArray[np.bool_]): Boolean array of the same shape as the data.
            False indicates values to be masked.

    Raises:
        TypeError: If the data is not a floating-point numeric type.
    """
    if not np.issubdtype(self._data.dtype, np.floating):
        msg = f"Masking is only supported for floating-point types! Encountered for variable {self}."
        raise TypeError(msg)

    self._data[~mask] = np.nan

el_paso.variable.Variable.apply_thresholds_on_data

apply_thresholds_on_data

Applies lower and upper thresholds to the data.

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

Parameters:

Name	Type	Description	Default
lower_threshold	float	The lower bound for the data. Defaults to negative infinity.	-inf
upper_threshold	float	The upper bound for the data. Defaults to positive infinity.	inf

Raises:

Type	Description
TypeError	If the data is not a floating-point numeric type.

Source code in el_paso/variable.py

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 a floating-point numeric type.
    """
    if not np.issubdtype(self._data.dtype, np.floating):
        msg = f"Thresholds are only supported for floating-point types! Encountered for variable {self}."
        raise TypeError(msg)
    self._data = typing.cast("NDArray[np.floating]", self._data)

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

el_paso.variable.Variable.convert_to_unit

convert_to_unit

Converts the data to a given unit.

Parameters:

Name	Type	Description	Default
target_unit	UnitBase | str	The unit the data should be converted to.	required

Source code in el_paso/variable.py

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))

        self.metadata.unit = target_unit

el_paso.variable.Variable.get_data

get_data

get_data

get_data

Gets the data of the variable.

Parameters:

Name	Type	Description	Default
target_unit	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.	None

Returns:

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

Raises:

Type	Description
TypeError	If target_unit is provided and the data is not numeric.

Source code in el_paso/variable.py

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))

el_paso.variable.Variable.set_data

set_data

Sets the data and optionally updates the unit of the variable.

Parameters:

Name	Type	Description	Default
data	NDArray[generic]	The new data array.	required
unit	Literal['same'] | str | 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.	required

Raises:

Type	Description
TypeError	If unit is not "same", a string, or an astropy.units.UnitBase object.

Source code in el_paso/variable.py

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):
        self.metadata.unit = unit
    else:
        msg = "unit must be either a str or a astropy unit!"
        raise TypeError(msg)

el_paso.variable.Variable.transpose_data

transpose_data

Transposes the internal data array.

Parameters:

Name	Type	Description	Default
seq	list[int] | tuple[int, ...]	The axes to transpose to. See numpy.transpose for details.	required

Source code in el_paso/variable.py

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)

el_paso.variable.Variable.truncate

truncate

Truncates the variable's data based on a time variable and a time range.

Parameters:

Name	Type	Description	Default
time_variable	Variable	A Variable object containing the time data.	required
start_time	float | datetime	The start time for truncation. Can be a Unix timestamp (float) or a datetime object.	required
end_time	float | datetime	The end time for truncation. Can be a Unix timestamp (float) or a datetime object.	required

Raises:

Type	Description
ValueError	If the length of the variable's data does not match the length of the time_variable's data.

Source code in el_paso/variable.py

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)]