Time Setting

Overview¤

A module for handling time settings passed in the RetrievalSettings class. Set time settings used by DisruptionPy to set the timebase for data retrieval from MDSPlus and any SQL tables.

This module defines the abstract class TimeSetting that can have subclasses passed as the time_setting parameter to the RetrievalSettings class. It also provides built-in classes and mappings to easily set the timebase for data retrievel for common use cases.

Usage¤

Currently, these are the options that can be passed to the time_setting parameter in RetrievalSettings:

An instance of a subclass of TimeSetting
A string identifier in the _time_setting_mappings dictionary:

_time_setting_mappings: Dict[str, TimeSetting] = {
    "efit": EfitTimeSetting(),
    "disruption_warning": {
        Tokamak.CMOD: EfitTimeSetting(),
        Tokamak.D3D: DisruptionTimeSetting(),
        Tokamak.EAST: DisruptionTimeSetting(),
    },
    "ip": IpTimeSetting(),
    "ip_efit": SharedTimeSetting([IpTimeSetting(), EfitTimeSetting()]),
}

- A Python list, NumPy array, or Pandas Series (with the timebase as the values) that should be used as the times for the timebase. See ListTimeSetting for more details. - A dictionary mapping tokamak type strings to the desired TimeSetting for that tokamak. E.g. {'cmod': 'efit'}.

Built-in Implemenations¤

This module defines classes for time settings, used to manage the timebase for retrieving data in disruption_py for various tokamaks and shot configurations.

disruption_py.settings.time_setting.DisruptionTimeSetting ¤

Bases: TimeSetting

Time setting for using the disruption timebase.

The disruption timebase is kept for compatibility with the pre-computed matlab-based disruption warning DB, and consists in an a-priori computation depending on machine-specific settings.

Source code in disruption_py/settings/time_setting.py

class DisruptionTimeSetting(TimeSetting):
    """
    Time setting for using the disruption timebase.

    The disruption timebase is kept for compatibility with the pre-computed matlab-based disruption
    warning DB, and consists in an a-priori computation depending on machine-specific settings.
    """

    def __init__(self):
        """
        Initialize with tokamak overrides.
        """
        self.tokamak_overrides = {
            Tokamak.D3D: self.d3d_times,
            Tokamak.EAST: self.east_times,
        }

    def _get_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Fallback method for retrieving the disruption timebase.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        raise ValueError("Disruption time setting not implemented")

    def d3d_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the disruption timebase for DIII-D.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        raw_ip, ip_time = params.mds_conn.get_data_with_dims(
            f"ptdata('ip', {params.shot_id})", tree_name="d3d"
        )
        ip_time = ip_time / 1.0e3
        baseline = np.mean(raw_ip[:10])
        ip = raw_ip - baseline
        return self._calculate_disruption_times(params, ip, ip_time)

    def east_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the disruption timebase for EAST.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        ip, ip_time = EastUtilMethods.retrieve_ip(params.mds_conn, params.shot_id)
        return self._calculate_disruption_times(params, ip, ip_time)

    @classmethod
    def _calculate_disruption_times(
        cls, params: TimeSettingParams, ip: np.ndarray, ip_time: np.ndarray
    ) -> np.ndarray:
        """
        Calculate the disruption time base given ip, ip_time, and time setting parameters.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to compute the timebase.
        ip : np.ndarray
            Array of values for the plasma current.
        ip_time : np.ndarray
            Array of time for the plasma current.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.

        References
        ----------
        - D3D: [disruption_warning_database_d3d.m](https://github.com/MIT-PSFC/disruption-py/
        blob/matlab/DIII-D/disruption_warning_database_d3d.m), [check_for_valid_
        plasma.m](https://github.com/MIT-PSFC/disruption-py/blob/matlab/DIII-D/utils/check
        _for_valid_plasma.m)
        - EAST: [disruption_warning_database.m](https://github.com/MIT-PSFC/disruption-py/blob/mat
        lab/EAST/disruption_warning_database.m), [check_for_valid_plasma.m]https://github.com/MIT-
        PSFC/disruption-py/blob/matlab/EAST/utils/check_for_valid_plasma.m
        """
        time_config = config(params.tokamak).time
        duration, ip_max = cls._get_end_of_shot(
            ip, ip_time, time_config.end_of_current_threshold
        )
        if (
            duration < time_config.minimum_duration
            or np.abs(ip_max) < time_config.minimum_ip
        ):
            raise NotImplementedError(
                "Duration or current maximum are below thresholds."
            )

        times = np.arange(
            time_config.disruption_time_start,
            duration + time_config.time_const,
            time_config.disruption_time_step,
        )
        if params.disrupted:
            additional_times = np.arange(
                params.disruption_time - time_config.duration_before_disruption,
                params.disruption_time + time_config.time_const,
                time_config.dt_before_disruption,
            )
            times = times[
                np.where(
                    times
                    < (
                        params.disruption_time
                        - time_config.duration_before_disruption
                        - time_config.time_const
                    )
                )
            ]
            times = np.concatenate((times, additional_times))
        return times

    @classmethod
    def _get_end_of_shot(
        cls, signal, signal_time, threshold=1.0e5
    ) -> Tuple[float, float]:
        """
        Calculate the end of shot based on signal and threshold.

        Parameters
        ----------
        signal : np.ndarray
            Signal array representing the current.
        signal_time : np.ndarray
            Time array corresponding to the signal.
        threshold : float
            Threshold value for determining shot end.

        Returns
        -------
        duration : float
            Duration of the shot.
        signal_max : float
            Maximum signal value.
        """
        duration = 0
        signal_max = 0
        if threshold < 0:
            raise Warning("Threshold is negative.")
        (base_indices,) = np.where(signal_time <= 0.0)
        baseline = np.mean(signal[base_indices]) if len(base_indices) > 0 else 0
        signal -= baseline
        # Check if there was a finite signal otherwise consider the shot a "no plasma" shot
        (finite_indices,) = np.where(
            (signal_time >= 0.0) & (np.abs(signal) > threshold)
        )
        if len(finite_indices) == 0:
            return duration, signal_max
        dt = np.diff(signal_time)
        duration = np.sum(dt[finite_indices[:-1]])
        if duration < 0.1:  # Assuming < 100 ms is not a bona fide plasma
            duration = 0
            return duration, signal_max
        polarity = np.sign(
            np.trapz(signal[finite_indices], signal_time[finite_indices])
        )
        polarized_signal = polarity * signal
        (valid_indices,) = np.where(
            (polarized_signal >= threshold) & (signal_time > 0.0)
        )
        duration = signal_time[np.max(valid_indices)]
        if len(valid_indices) == signal_time.size:
            duration = -duration
        signal_max = np.max(polarized_signal) * polarity
        return duration, signal_max

init ¤

__init__()

Source code in disruption_py/settings/time_setting.py

def __init__(self):
    """
    Initialize with tokamak overrides.
    """
    self.tokamak_overrides = {
        Tokamak.D3D: self.d3d_times,
        Tokamak.EAST: self.east_times,
    }

_calculate_disruption_times `classmethod` ¤

_calculate_disruption_times(
    params: TimeSettingParams, ip: ndarray, ip_time: ndarray
) -> np.ndarray

Calculate the disruption time base given ip, ip_time, and time setting parameters.

PARAMETER	DESCRIPTION
`params`	Parameters needed to compute the timebase. TYPE: `TimeSettingParams`
`ip`	Array of values for the plasma current. TYPE: `ndarray`
`ip_time`	Array of time for the plasma current. TYPE: `ndarray`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

References

D3D: disruption_warning_database_d3d.m, check_for_valid_ plasma.m
EAST: disruption_warning_database.m, [check_for_valid_plasma.m]https://github.com/MIT- PSFC/disruption-py/blob/matlab/EAST/utils/check_for_valid_plasma.m

Source code in disruption_py/settings/time_setting.py

@classmethod
def _calculate_disruption_times(
    cls, params: TimeSettingParams, ip: np.ndarray, ip_time: np.ndarray
) -> np.ndarray:
    """
    Calculate the disruption time base given ip, ip_time, and time setting parameters.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to compute the timebase.
    ip : np.ndarray
        Array of values for the plasma current.
    ip_time : np.ndarray
        Array of time for the plasma current.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.

    References
    ----------
    - D3D: [disruption_warning_database_d3d.m](https://github.com/MIT-PSFC/disruption-py/
    blob/matlab/DIII-D/disruption_warning_database_d3d.m), [check_for_valid_
    plasma.m](https://github.com/MIT-PSFC/disruption-py/blob/matlab/DIII-D/utils/check
    _for_valid_plasma.m)
    - EAST: [disruption_warning_database.m](https://github.com/MIT-PSFC/disruption-py/blob/mat
    lab/EAST/disruption_warning_database.m), [check_for_valid_plasma.m]https://github.com/MIT-
    PSFC/disruption-py/blob/matlab/EAST/utils/check_for_valid_plasma.m
    """
    time_config = config(params.tokamak).time
    duration, ip_max = cls._get_end_of_shot(
        ip, ip_time, time_config.end_of_current_threshold
    )
    if (
        duration < time_config.minimum_duration
        or np.abs(ip_max) < time_config.minimum_ip
    ):
        raise NotImplementedError(
            "Duration or current maximum are below thresholds."
        )

    times = np.arange(
        time_config.disruption_time_start,
        duration + time_config.time_const,
        time_config.disruption_time_step,
    )
    if params.disrupted:
        additional_times = np.arange(
            params.disruption_time - time_config.duration_before_disruption,
            params.disruption_time + time_config.time_const,
            time_config.dt_before_disruption,
        )
        times = times[
            np.where(
                times
                < (
                    params.disruption_time
                    - time_config.duration_before_disruption
                    - time_config.time_const
                )
            )
        ]
        times = np.concatenate((times, additional_times))
    return times

_get_end_of_shot `classmethod` ¤

_get_end_of_shot(
    signal, signal_time, threshold=100000.0
) -> Tuple[float, float]

Calculate the end of shot based on signal and threshold.

PARAMETER	DESCRIPTION
`signal`	Signal array representing the current. TYPE: `ndarray`
`signal_time`	Time array corresponding to the signal. TYPE: `ndarray`
`threshold`	Threshold value for determining shot end. TYPE: `float` DEFAULT: `100000.0`

RETURNS	DESCRIPTION
`duration`	Duration of the shot. TYPE: `float`
`signal_max`	Maximum signal value. TYPE: `float`

Source code in disruption_py/settings/time_setting.py

@classmethod
def _get_end_of_shot(
    cls, signal, signal_time, threshold=1.0e5
) -> Tuple[float, float]:
    """
    Calculate the end of shot based on signal and threshold.

    Parameters
    ----------
    signal : np.ndarray
        Signal array representing the current.
    signal_time : np.ndarray
        Time array corresponding to the signal.
    threshold : float
        Threshold value for determining shot end.

    Returns
    -------
    duration : float
        Duration of the shot.
    signal_max : float
        Maximum signal value.
    """
    duration = 0
    signal_max = 0
    if threshold < 0:
        raise Warning("Threshold is negative.")
    (base_indices,) = np.where(signal_time <= 0.0)
    baseline = np.mean(signal[base_indices]) if len(base_indices) > 0 else 0
    signal -= baseline
    # Check if there was a finite signal otherwise consider the shot a "no plasma" shot
    (finite_indices,) = np.where(
        (signal_time >= 0.0) & (np.abs(signal) > threshold)
    )
    if len(finite_indices) == 0:
        return duration, signal_max
    dt = np.diff(signal_time)
    duration = np.sum(dt[finite_indices[:-1]])
    if duration < 0.1:  # Assuming < 100 ms is not a bona fide plasma
        duration = 0
        return duration, signal_max
    polarity = np.sign(
        np.trapz(signal[finite_indices], signal_time[finite_indices])
    )
    polarized_signal = polarity * signal
    (valid_indices,) = np.where(
        (polarized_signal >= threshold) & (signal_time > 0.0)
    )
    duration = signal_time[np.max(valid_indices)]
    if len(valid_indices) == signal_time.size:
        duration = -duration
    signal_max = np.max(polarized_signal) * polarity
    return duration, signal_max

_get_times ¤

_get_times(params: TimeSettingParams) -> np.ndarray

Fallback method for retrieving the disruption timebase.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def _get_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Fallback method for retrieving the disruption timebase.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    raise ValueError("Disruption time setting not implemented")

d3d_times ¤

d3d_times(params: TimeSettingParams) -> np.ndarray

Retrieve the disruption timebase for DIII-D.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def d3d_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the disruption timebase for DIII-D.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    raw_ip, ip_time = params.mds_conn.get_data_with_dims(
        f"ptdata('ip', {params.shot_id})", tree_name="d3d"
    )
    ip_time = ip_time / 1.0e3
    baseline = np.mean(raw_ip[:10])
    ip = raw_ip - baseline
    return self._calculate_disruption_times(params, ip, ip_time)

east_times ¤

east_times(params: TimeSettingParams) -> np.ndarray

Retrieve the disruption timebase for EAST.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def east_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the disruption timebase for EAST.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    ip, ip_time = EastUtilMethods.retrieve_ip(params.mds_conn, params.shot_id)
    return self._calculate_disruption_times(params, ip, ip_time)

disruption_py.settings.time_setting.EfitTimeSetting ¤

Bases: TimeSetting

Time setting for using the EFIT timebase.

Source code in disruption_py/settings/time_setting.py

class EfitTimeSetting(TimeSetting):
    """
    Time setting for using the EFIT timebase.
    """

    def _get_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the EFIT timebase for the tested tokamaks.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        (efit_time,) = params.mds_conn.get_dims(
            r"\efit_aeqdsk:ali", tree_name="_efit_tree", astype="float64"
        )
        efit_time_unit = params.mds_conn.get_data(
            r"units_of(dim_of(\efit_aeqdsk:ali))", tree_name="_efit_tree", astype="str"
        )
        if efit_time_unit not in {"s", "ms", "us"}:
            params.logger.verbose(
                "Failed to get the time units of EFIT tree '{tree}', assuming seconds.",
                tree=params.mds_conn.get_tree_name_of_nickname("_efit_tree"),
            )
        return _postprocess(times=efit_time, units=efit_time_unit)

_get_times ¤

_get_times(params: TimeSettingParams) -> np.ndarray

Retrieve the EFIT timebase for the tested tokamaks.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def _get_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the EFIT timebase for the tested tokamaks.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    (efit_time,) = params.mds_conn.get_dims(
        r"\efit_aeqdsk:ali", tree_name="_efit_tree", astype="float64"
    )
    efit_time_unit = params.mds_conn.get_data(
        r"units_of(dim_of(\efit_aeqdsk:ali))", tree_name="_efit_tree", astype="str"
    )
    if efit_time_unit not in {"s", "ms", "us"}:
        params.logger.verbose(
            "Failed to get the time units of EFIT tree '{tree}', assuming seconds.",
            tree=params.mds_conn.get_tree_name_of_nickname("_efit_tree"),
        )
    return _postprocess(times=efit_time, units=efit_time_unit)

disruption_py.settings.time_setting.IpTimeSetting ¤

Bases: TimeSetting

Time setting for using the timebase of the plasma current.

Source code in disruption_py/settings/time_setting.py

class IpTimeSetting(TimeSetting):
    """
    Time setting for using the timebase of the plasma current.
    """

    def __init__(self):
        """
        Initialize with tokamak-specific overrides.
        """
        self.tokamak_overrides = {
            Tokamak.CMOD: self.cmod_times,
            Tokamak.D3D: self.d3d_times,
            Tokamak.EAST: self.east_times,
        }

    def _get_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Abstract method for retrieving the Ip timebase.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        raise ValueError("Ip time setting not implemented")

    def cmod_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the Ip timebase for the CMOD tokamak.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        (ip_time,) = params.mds_conn.get_dims(r"\ip", tree_name="magnetics")
        return ip_time

    def d3d_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the Ip timebase for the D3D tokamak.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        (ip_time,) = params.mds_conn.get_dims(
            f"ptdata('ip', {params.shot_id})", tree_name=None
        )
        ip_time /= 1e3  # [ms] -> [s]
        return ip_time

    def east_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the Ip timebase for the EAST tokamak.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        (ip_time,) = params.mds_conn.get_dims(r"\pcrl01", tree_name="pcs_east")
        # For shots before year 2014, the PCRL01 timebase needs to be shifted
        # by 17.0 ms
        if params.shot_id < 44432:
            ip_time -= 0.0170
        return ip_time

init ¤

__init__()

Source code in disruption_py/settings/time_setting.py

def __init__(self):
    """
    Initialize with tokamak-specific overrides.
    """
    self.tokamak_overrides = {
        Tokamak.CMOD: self.cmod_times,
        Tokamak.D3D: self.d3d_times,
        Tokamak.EAST: self.east_times,
    }

_get_times ¤

_get_times(params: TimeSettingParams) -> np.ndarray

Abstract method for retrieving the Ip timebase.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def _get_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Abstract method for retrieving the Ip timebase.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    raise ValueError("Ip time setting not implemented")

cmod_times ¤

cmod_times(params: TimeSettingParams) -> np.ndarray

Retrieve the Ip timebase for the CMOD tokamak.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def cmod_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the Ip timebase for the CMOD tokamak.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    (ip_time,) = params.mds_conn.get_dims(r"\ip", tree_name="magnetics")
    return ip_time

d3d_times ¤

d3d_times(params: TimeSettingParams) -> np.ndarray

Retrieve the Ip timebase for the D3D tokamak.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def d3d_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the Ip timebase for the D3D tokamak.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    (ip_time,) = params.mds_conn.get_dims(
        f"ptdata('ip', {params.shot_id})", tree_name=None
    )
    ip_time /= 1e3  # [ms] -> [s]
    return ip_time

east_times ¤

east_times(params: TimeSettingParams) -> np.ndarray

Retrieve the Ip timebase for the EAST tokamak.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def east_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the Ip timebase for the EAST tokamak.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    (ip_time,) = params.mds_conn.get_dims(r"\pcrl01", tree_name="pcs_east")
    # For shots before year 2014, the PCRL01 timebase needs to be shifted
    # by 17.0 ms
    if params.shot_id < 44432:
        ip_time -= 0.0170
    return ip_time

disruption_py.settings.time_setting.ListTimeSetting ¤

Bases: TimeSetting

Time setting for using a pre-defined list of times.

Used when a list, numpy array, or pandas series is passed as the time_setting parameter in RetrievalSettings.

Source code in disruption_py/settings/time_setting.py

class ListTimeSetting(TimeSetting):
    """
    Time setting for using a pre-defined list of times.

    Used when a list, numpy array, or pandas series is passed as the `time_setting`
    parameter in `RetrievalSettings`.
    """

    def __init__(self, times):
        """
        Initialize with a list of times.

        Parameters
        ----------
        times : list, np.ndarray
            List or array of times to use as the timebase.
        """
        self.times = times

    def _get_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Return the pre-defined list of times.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        return self.times

init ¤

__init__(times)

PARAMETER	DESCRIPTION
`times`	List or array of times to use as the timebase. TYPE: `(list, ndarray)`

Source code in disruption_py/settings/time_setting.py

def __init__(self, times):
    """
    Initialize with a list of times.

    Parameters
    ----------
    times : list, np.ndarray
        List or array of times to use as the timebase.
    """
    self.times = times

_get_times ¤

_get_times(params: TimeSettingParams) -> np.ndarray

Return the pre-defined list of times.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def _get_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Return the pre-defined list of times.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    return self.times

disruption_py.settings.time_setting.SharedTimeSetting ¤

Bases: TimeSetting

Time setting for using a "shared" time among several time settings. The first time setting in the list is used to determine the timebase, while the other time settings in the list are used to select the starting and ending points, so that the final time base focuses on a time window shared among all the given items.

Source code in disruption_py/settings/time_setting.py

class SharedTimeSetting(TimeSetting):
    """
    Time setting for using a "shared" time among several time settings.
    The first time setting in the list is used to determine the timebase, while the other
    time settings in the list are used to select the starting and ending points, so that
    the final time base focuses on a time window shared among all the given items.
    """

    def __init__(self, time_setting_list: list[TimeSetting]):
        """
        Initialize with a list of time settings.

        Parameters
        ----------
        time_setting_list : list[TimeSetting]
            List of time settings.
        """
        if len(time_setting_list) < 2:
            raise ValueError("SharedTimeSetting requires at least 2 time settings.")
        self.time_setting_list = time_setting_list

    def _get_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the shared timebase for the given settings.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        times, *others = [ts.get_times(params) for ts in self.time_setting_list]
        tmin = np.max([np.min(t) for t in others])
        tmax = np.min([np.max(t) for t in others])
        return times[np.where((times >= tmin) & (times <= tmax))]

init ¤

__init__(time_setting_list: list[TimeSetting])

PARAMETER	DESCRIPTION
`time_setting_list`	List of time settings. TYPE: `list[TimeSetting]`

Source code in disruption_py/settings/time_setting.py

def __init__(self, time_setting_list: list[TimeSetting]):
    """
    Initialize with a list of time settings.

    Parameters
    ----------
    time_setting_list : list[TimeSetting]
        List of time settings.
    """
    if len(time_setting_list) < 2:
        raise ValueError("SharedTimeSetting requires at least 2 time settings.")
    self.time_setting_list = time_setting_list

_get_times ¤

_get_times(params: TimeSettingParams) -> np.ndarray

Retrieve the shared timebase for the given settings.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def _get_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the shared timebase for the given settings.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    times, *others = [ts.get_times(params) for ts in self.time_setting_list]
    tmin = np.max([np.min(t) for t in others])
    tmax = np.min([np.max(t) for t in others])
    return times[np.where((times >= tmin) & (times <= tmax))]

disruption_py.settings.time_setting.SignalTimeSetting ¤

Bases: TimeSetting

Time setting for using the timebase of a specific signal.

Source code in disruption_py/settings/time_setting.py

class SignalTimeSetting(TimeSetting):
    """
    Time setting for using the timebase of a specific signal.
    """

    def __init__(self, tree_name: str, signal_path: str):
        """
        Initialize with the tree name and signal path.

        Parameters
        ----------
        tree_name : str
            Name of the tree containing the signal.
        signal_path : str
            Path to the signal within the tree.
        """
        self.tree_name = tree_name
        self.signal_path = signal_path
        self.tokamak_overrides = {
            Tokamak.D3D: self._get_times_d3d_wrapper,
        }

    def _get_times_d3d_wrapper(self, params: TimeSettingParams) -> np.ndarray:
        """
        Wrapper function for the DIII-D tokamak
        - Allow using tree_name="ptdata" to call PTDATA signals
        - Convert time unit from ms to s
        """
        if self.tree_name.lower() == "ptdata":
            self.signal_path = f'ptdata("{self.signal_path}", {params.shot_id})'
            self.tree_name = None
        return self._get_times(params)

    def _get_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the timebase for the specified signal.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters needed to retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        try:
            (signal_time,) = params.mds_conn.get_dims(
                self.signal_path, tree_name=self.tree_name, astype="float64"
            )
        except mdsExceptions.MdsException:
            params.logger.error(
                "Failed to set up timebase for signal {signal_path}",
                signal_path=self.signal_path,
            )
            raise
        signal_unit = params.mds_conn.get_data(
            f"units_of(dim_of({self.signal_path}))",
            tree_name=self.tree_name,
            astype="str",
        )
        if (
            not signal_unit.strip()
            and self.signal_path.startswith("ptdata")
            and params.tokamak == Tokamak.D3D
        ):
            # timebase of PTDATA signal defaults to [ms]
            signal_unit = "ms"
        elif signal_unit != "s":
            params.logger.warning(
                "Failed to get the time unit of signal `{path}`, assuming seconds.",
                path=self.signal_path,
            )
        return _postprocess(times=signal_time, units=signal_unit)

init ¤

__init__(tree_name: str, signal_path: str)

PARAMETER	DESCRIPTION
`tree_name`	Name of the tree containing the signal. TYPE: `str`
`signal_path`	Path to the signal within the tree. TYPE: `str`

Source code in disruption_py/settings/time_setting.py

def __init__(self, tree_name: str, signal_path: str):
    """
    Initialize with the tree name and signal path.

    Parameters
    ----------
    tree_name : str
        Name of the tree containing the signal.
    signal_path : str
        Path to the signal within the tree.
    """
    self.tree_name = tree_name
    self.signal_path = signal_path
    self.tokamak_overrides = {
        Tokamak.D3D: self._get_times_d3d_wrapper,
    }

_get_times ¤

_get_times(params: TimeSettingParams) -> np.ndarray

Retrieve the timebase for the specified signal.

PARAMETER	DESCRIPTION
`params`	Parameters needed to retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def _get_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the timebase for the specified signal.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters needed to retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    try:
        (signal_time,) = params.mds_conn.get_dims(
            self.signal_path, tree_name=self.tree_name, astype="float64"
        )
    except mdsExceptions.MdsException:
        params.logger.error(
            "Failed to set up timebase for signal {signal_path}",
            signal_path=self.signal_path,
        )
        raise
    signal_unit = params.mds_conn.get_data(
        f"units_of(dim_of({self.signal_path}))",
        tree_name=self.tree_name,
        astype="str",
    )
    if (
        not signal_unit.strip()
        and self.signal_path.startswith("ptdata")
        and params.tokamak == Tokamak.D3D
    ):
        # timebase of PTDATA signal defaults to [ms]
        signal_unit = "ms"
    elif signal_unit != "s":
        params.logger.warning(
            "Failed to get the time unit of signal `{path}`, assuming seconds.",
            path=self.signal_path,
        )
    return _postprocess(times=signal_time, units=signal_unit)

_get_times_d3d_wrapper ¤

_get_times_d3d_wrapper(
    params: TimeSettingParams,
) -> np.ndarray

Wrapper function for the DIII-D tokamak - Allow using tree_name="ptdata" to call PTDATA signals - Convert time unit from ms to s

Source code in disruption_py/settings/time_setting.py

def _get_times_d3d_wrapper(self, params: TimeSettingParams) -> np.ndarray:
    """
    Wrapper function for the DIII-D tokamak
    - Allow using tree_name="ptdata" to call PTDATA signals
    - Convert time unit from ms to s
    """
    if self.tree_name.lower() == "ptdata":
        self.signal_path = f'ptdata("{self.signal_path}", {params.shot_id})'
        self.tree_name = None
    return self._get_times(params)

disruption_py.settings.time_setting._postprocess ¤

_postprocess(times: ndarray, units: str = '') -> np.ndarray

Return a deduplicated, sorted, rescaled time array.

PARAMETER	DESCRIPTION
`times`	Array of times. TYPE: `ndarray`
`units`	Unit of measure for the array of times. TYPE: `str` DEFAULT: `''`

RETURNS	DESCRIPTION
`ndarray`	Post-processed array of times.

Source code in disruption_py/settings/time_setting.py

def _postprocess(times: np.ndarray, units: str = "") -> np.ndarray:
    """
    Return a deduplicated, sorted, rescaled time array.

    Parameters
    ----------
    times : np.ndarray
        Array of times.
    units : str
        Unit of measure for the array of times.

    Returns
    -------
    np.ndarray
        Post-processed array of times.
    """
    times = np.unique(times)
    units = units.lower().strip()
    if units == "ms":
        return times * 1e-3
    if units == "us":
        return times * 1e-6
    return times

disruption_py.settings.time_setting.resolve_time_setting ¤

resolve_time_setting(
    time_setting: TimeSettingType,
) -> TimeSetting

Resolve a time setting to a TimeSetting object.

PARAMETER	DESCRIPTION
`time_setting`	The time setting, which can be a string, list, or TimeSetting instance. TYPE: `TimeSettingType`

RETURNS	DESCRIPTION
`TimeSetting`	The resolved TimeSetting object.

Source code in disruption_py/settings/time_setting.py

def resolve_time_setting(
    time_setting: TimeSettingType,
) -> TimeSetting:
    """
    Resolve a time setting to a TimeSetting object.

    Parameters
    ----------
    time_setting : TimeSettingType
        The time setting, which can be a string, list, or TimeSetting instance.

    Returns
    -------
    TimeSetting
        The resolved TimeSetting object.
    """
    if isinstance(time_setting, TimeSetting):
        return time_setting

    if isinstance(time_setting, str):
        time_setting_object = _time_setting_mappings.get(time_setting)
        if time_setting_object is not None:
            return time_setting_object

    if isinstance(time_setting, list):
        if all(isinstance(ts, TimeSetting) for ts in time_setting):
            return SharedTimeSetting(time_setting)
        return ListTimeSetting(time_setting)

    if isinstance(time_setting, np.ndarray):
        return ListTimeSetting(time_setting)

    if isinstance(time_setting, dict):
        return TimeSettingDict(time_setting)

    raise ValueError("Invalid time setting")

Custom Implementations¤

Custom implementations of time settings must inherit from the TimeSetting abstract class, implementing the abstract methods.

disruption_py.settings.time_setting ¤

This module defines classes for time settings, used to manage the timebase for retrieving data in disruption_py for various tokamaks and shot configurations.

TimeSetting ¤

Bases: ABC

Abstract base class for managing time settings to retrieve the timebase for shots.

METHOD	DESCRIPTION
`get_times`	Retrieve the timebase as a numpy array.

Source code in disruption_py/settings/time_setting.py

class TimeSetting(ABC):
    """
    Abstract base class for managing time settings to retrieve the timebase for shots.

    Methods
    -------
    get_times(params)
        Retrieve the timebase as a numpy array.
    """

    def get_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Retrieve the timebase using the provided parameters.

        Parameters
        ----------
        params : TimeSettingParams
            The parameters used to determine and retrieve the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """
        if hasattr(self, "tokamak_overrides"):
            if params.tokamak in self.tokamak_overrides:
                return _postprocess(
                    times=self.tokamak_overrides[params.tokamak](params)
                )
        return _postprocess(times=self._get_times(params))

    @abstractmethod
    def _get_times(self, params: TimeSettingParams) -> np.ndarray:
        """
        Abstract method for subclasses to implement timebase retrieval.

        Parameters
        ----------
        params : TimeSettingParams
            Parameters used to determine the timebase.

        Returns
        -------
        np.ndarray
            Array of times in the timebase.
        """

_get_times `abstractmethod` ¤

_get_times(params: TimeSettingParams) -> np.ndarray

Abstract method for subclasses to implement timebase retrieval.

PARAMETER	DESCRIPTION
`params`	Parameters used to determine the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

@abstractmethod
def _get_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Abstract method for subclasses to implement timebase retrieval.

    Parameters
    ----------
    params : TimeSettingParams
        Parameters used to determine the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """

get_times ¤

get_times(params: TimeSettingParams) -> np.ndarray

Retrieve the timebase using the provided parameters.

PARAMETER	DESCRIPTION
`params`	The parameters used to determine and retrieve the timebase. TYPE: `TimeSettingParams`

RETURNS	DESCRIPTION
`ndarray`	Array of times in the timebase.

Source code in disruption_py/settings/time_setting.py

def get_times(self, params: TimeSettingParams) -> np.ndarray:
    """
    Retrieve the timebase using the provided parameters.

    Parameters
    ----------
    params : TimeSettingParams
        The parameters used to determine and retrieve the timebase.

    Returns
    -------
    np.ndarray
        Array of times in the timebase.
    """
    if hasattr(self, "tokamak_overrides"):
        if params.tokamak in self.tokamak_overrides:
            return _postprocess(
                times=self.tokamak_overrides[params.tokamak](params)
            )
    return _postprocess(times=self._get_times(params))

TimeSettingParams `dataclass` ¤

Parameters passed by disruption_py to the _get_times() method.

ATTRIBUTE	DESCRIPTION
`shot_id`	Shot ID for the timebase being created. TYPE: `int`
`mds_conn`	Connection to MDSPlus for retrieving MDSPlus data. TYPE: `MDSConnection`
`database`	Database object with connection to the SQL database. TYPE: `ShotDatabase`
`disruption_time`	Time when the shot disrupted in seconds (or None if no disruption occurred). TYPE: `float`
`tokamak`	Tokamak for which the time setting is applied. TYPE: `Tokamak`

Source code in disruption_py/settings/time_setting.py

@dataclass
class TimeSettingParams:
    """
    Parameters passed by disruption_py to the _get_times() method.

    Attributes
    ----------
    shot_id : int
        Shot ID for the timebase being created.
    mds_conn : MDSConnection
        Connection to MDSPlus for retrieving MDSPlus data.
    database : ShotDatabase
        Database object with connection to the SQL database.
    disruption_time : float
        Time when the shot disrupted in seconds (or None if no disruption occurred).
    tokamak : Tokamak
        Tokamak for which the time setting is applied.
    """

    shot_id: int
    mds_conn: MDSConnection
    database: ShotDatabase
    disruption_time: float
    tokamak: Tokamak

    def __post_init__(self):
        self.logger = shot_msg_patch(logger, self.shot_id)

    @property
    def disrupted(self) -> bool:
        """
        Check if the shot disrupted.

        Returns
        -------
        bool
            True if the shot was disrupted, False otherwise.
        """
        return self.disruption_time is not None

disrupted `property` ¤

disrupted: bool

Check if the shot disrupted.

RETURNS	DESCRIPTION
`bool`	True if the shot was disrupted, False otherwise.

Custom time settings example¤

docs.examples.custom_time_setting ¤

Example usage of get_shots_data demonstrating using a custom time setting.

PRadTime ¤

Bases: TimeSetting

Class for retrieving prad times

Source code in docs/examples/custom_time_setting.py

class PRadTime(TimeSetting):
    """Class for retrieving prad times"""

    def _get_times(self, params: TimeSettingParams):
        """Return prad times"""
        (time_array,) = params.mds_conn.get_dims(
            r"\twopi_diode", tree_name="spectroscopy"
        )
        time_array = time_array[time_array > 0]
        return time_array

_get_times ¤

_get_times(params: TimeSettingParams)

Return prad times

Source code in docs/examples/custom_time_setting.py

def _get_times(self, params: TimeSettingParams):
    """Return prad times"""
    (time_array,) = params.mds_conn.get_dims(
        r"\twopi_diode", tree_name="spectroscopy"
    )
    time_array = time_array[time_array > 0]
    return time_array

Time Setting

Overview¤

Usage¤

Built-in Implemenations¤

disruption_py.settings.time_setting.DisruptionTimeSetting ¤

__init__ ¤

_calculate_disruption_times classmethod ¤

_get_end_of_shot classmethod ¤

_get_times ¤

d3d_times ¤

east_times ¤

disruption_py.settings.time_setting.EfitTimeSetting ¤

_get_times ¤

disruption_py.settings.time_setting.IpTimeSetting ¤

__init__ ¤

_get_times ¤

cmod_times ¤

d3d_times ¤

east_times ¤

disruption_py.settings.time_setting.ListTimeSetting ¤

__init__ ¤

_get_times ¤

disruption_py.settings.time_setting.SharedTimeSetting ¤

__init__ ¤

_get_times ¤

disruption_py.settings.time_setting.SignalTimeSetting ¤

__init__ ¤

_get_times ¤

_get_times_d3d_wrapper ¤

disruption_py.settings.time_setting._postprocess ¤

disruption_py.settings.time_setting.resolve_time_setting ¤

Custom Implementations¤

disruption_py.settings.time_setting ¤

TimeSetting ¤

_get_times abstractmethod ¤

get_times ¤

TimeSettingParams dataclass ¤

disrupted property ¤

Custom time settings example¤

docs.examples.custom_time_setting ¤

PRadTime ¤

_get_times ¤

init ¤

_calculate_disruption_times `classmethod` ¤

_get_end_of_shot `classmethod` ¤

init ¤

init ¤

init ¤

init ¤

_get_times `abstractmethod` ¤

TimeSettingParams `dataclass` ¤

disrupted `property` ¤