Domain Setting

Overview

DisruptionPy uses the time domain setting to specify the portion of the shot (ie rampup, flattop, full shot) for which to retrieve data.

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

Usage

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

• An instance of a subclass of DomainSetting
• A string identifier in the _domain_setting_mappings dictionary:

  _domain_setting_mappings: Dict[str, DomainSetting] = {
      "full": FullDomainSetting(),
      "flattop": FlattopDomainSetting(),
      "rampup_and_flattop": RampupAndFlattopDomainSetting(),
  }
  

• A dictionary mapping tokamak type strings to the desired DomainSetting for that tokamak. E.g. {'cmod': 'flattop'}. Currently supported tokamak type strings are: "cmod", "d3d"

Built-in Implemenations

This module provides classes and methods to specify different domain settings for the timebase used, including full, flattop, and ramp-up domains.

disruption_py.settings.domain_setting.FlattopDomainSetting

Bases: DomainSetting

Specifies that the flattop domain of the timebase should be used.

Source code in disruption_py/settings/domain_setting.py

class FlattopDomainSetting(DomainSetting):
    """
    Specifies that the flattop domain of the timebase should be used.
    """

    def __init__(self):
        """
        Initialize the FlattopDomainSetting with tokamak-specific overrides.
        """
        self.tokamak_overrides = {
            Tokamak.CMOD: self._get_domain_cmod,
            Tokamak.D3D: self._get_domain_d3d,
        }

    def _get_domain(self, params: DomainSettingParams) -> np.ndarray:
        """
        Get the flattop domain for the given tokamak.

        Parameters
        ----------
        params : DomainSettingParams
            Parameters containing physics method and tokamak information.

        Returns
        -------
        np.ndarray
            The flattop timebase domain.
        """
        raise ValueError(f"flattop domain not defined for tokamak: {params.tokamak}")

    def _get_domain_cmod(self, params: DomainSettingParams) -> np.ndarray:
        """
        Get the flattop domain for CMOD tokamak.

        Parameters
        ----------
        params : DomainSettingParams
            Parameters containing physics method and tokamak information.

        Returns
        -------
        np.ndarray
            The flattop timebase domain for CMOD.
        """
        ip_parameters = CmodPhysicsMethods.get_ip_parameters(
            params=params.physics_method_params
        )
        ipprog, dipprog_dt = ip_parameters["ip_prog"], ip_parameters["dipprog_dt"]
        indices_flattop_1 = np.where(np.abs(dipprog_dt) <= 1e3)[0]
        indices_flattop_2 = np.where(np.abs(ipprog) > 1.0e5)[0]
        indices_flattop = np.intersect1d(indices_flattop_1, indices_flattop_2)
        if len(indices_flattop) == 0:
            params.logger.warning(
                (
                    "[Shot %s]: Could not find flattop timebase. "
                    "Defaulting to full shot(efit) timebase."
                ),
                params.physics_method_params.shot_id,
            )
            return None
        return params.physics_method_params.times[indices_flattop]

    def _get_domain_d3d(self, params: DomainSettingParams) -> np.ndarray:
        """
        Get the flattop domain for D3D tokamak.

        Parameters
        ----------
        params : DomainSettingParams
            Parameters containing physics method and tokamak information.

        Returns
        -------
        np.ndarray
            The flattop timebase domain for D3D.
        """
        try:
            (
                ip_prog,
                t_ip_prog,
            ) = params.physics_method_params.mds_conn.get_data_with_dims(
                f"ptdata('iptipp', {params.physics_method_params.shot_id})",
                tree_name="d3d",
            )
            t_ip_prog = t_ip_prog / 1.0e3  # [ms] -> [s]
            polarity = np.unique(
                params.physics_method_params.mds_conn.get_data(
                    f"ptdata('iptdirect', {params.physics_method_params.shot_id})",
                    tree_name="d3d",
                )
            )
            if len(polarity) > 1:
                params.logger.info(
                    (
                        "[Shot %s]: Polarity of Ip target is not constant. "
                        "Using value at first timestep."
                    ),
                    params.physics_method_params.shot_id,
                )
                params.logger.debug(
                    "[Shot %s]: Polarity array %s",
                    params.physics_method_params.shot_id,
                    polarity,
                )
                polarity = polarity[0]
            ip_prog = ip_prog * polarity
            dipprog_dt = np.gradient(ip_prog, t_ip_prog)
            ip_prog = interp1(
                t_ip_prog, ip_prog, params.physics_method_params.times, "linear"
            )
            dipprog_dt = interp1(
                t_ip_prog, dipprog_dt, params.physics_method_params.times, "linear"
            )
        except mdsExceptions.MdsException:
            params.logger.warning(
                "[Shot %s]: Could not find flattop timebase. Defaulting to full timebase.",
                params.physics_method_params.shot_id,
            )
            params.logger.debug(
                "[Shot %s]: %s",
                params.physics_method_params.shot_id,
                traceback.format_exc(),
            )
            return None
        epsoff, t_epsoff = params.physics_method_params.mds_conn.get_data_with_dims(
            f"ptdata('epsoff', {params.physics_method_params.shot_id})", tree_name="d3d"
        )
        # [ms] -> [s] # Avoid problem with simultaneity of epsoff being triggered
        # exactly on the last time sample
        t_epsoff = t_epsoff / 1.0e3 + 0.001
        epsoff = interp1(t_epsoff, epsoff, params.physics_method_params.times, "linear")
        railed_indices = np.where(np.abs(epsoff) > 0.5)
        power_supply_railed = np.zeros(len(params.physics_method_params.times))
        power_supply_railed[railed_indices] = 1
        indices_flattop = np.where(
            (np.abs(dipprog_dt) <= 2.0e3)
            & (np.abs(ip_prog) > 100e3)
            & (power_supply_railed != 1)
        )
        return params.physics_method_params.times[indices_flattop]

__init__

__init__()

Source code in disruption_py/settings/domain_setting.py

def __init__(self):
    """
    Initialize the FlattopDomainSetting with tokamak-specific overrides.
    """
    self.tokamak_overrides = {
        Tokamak.CMOD: self._get_domain_cmod,
        Tokamak.D3D: self._get_domain_d3d,
    }

_get_domain

_get_domain(params: DomainSettingParams) -> np.ndarray

Get the flattop domain for the given tokamak.

PARAMETER	DESCRIPTION
params	Parameters containing physics method and tokamak information. TYPE: DomainSettingParams

RETURNS	DESCRIPTION
ndarray	The flattop timebase domain.

Source code in disruption_py/settings/domain_setting.py

def _get_domain(self, params: DomainSettingParams) -> np.ndarray:
    """
    Get the flattop domain for the given tokamak.

    Parameters
    ----------
    params : DomainSettingParams
        Parameters containing physics method and tokamak information.

    Returns
    -------
    np.ndarray
        The flattop timebase domain.
    """
    raise ValueError(f"flattop domain not defined for tokamak: {params.tokamak}")

_get_domain_cmod

_get_domain_cmod(params: DomainSettingParams) -> np.ndarray

Get the flattop domain for CMOD tokamak.

PARAMETER	DESCRIPTION
params	Parameters containing physics method and tokamak information. TYPE: DomainSettingParams

RETURNS	DESCRIPTION
ndarray	The flattop timebase domain for CMOD.

Source code in disruption_py/settings/domain_setting.py

def _get_domain_cmod(self, params: DomainSettingParams) -> np.ndarray:
    """
    Get the flattop domain for CMOD tokamak.

    Parameters
    ----------
    params : DomainSettingParams
        Parameters containing physics method and tokamak information.

    Returns
    -------
    np.ndarray
        The flattop timebase domain for CMOD.
    """
    ip_parameters = CmodPhysicsMethods.get_ip_parameters(
        params=params.physics_method_params
    )
    ipprog, dipprog_dt = ip_parameters["ip_prog"], ip_parameters["dipprog_dt"]
    indices_flattop_1 = np.where(np.abs(dipprog_dt) <= 1e3)[0]
    indices_flattop_2 = np.where(np.abs(ipprog) > 1.0e5)[0]
    indices_flattop = np.intersect1d(indices_flattop_1, indices_flattop_2)
    if len(indices_flattop) == 0:
        params.logger.warning(
            (
                "[Shot %s]: Could not find flattop timebase. "
                "Defaulting to full shot(efit) timebase."
            ),
            params.physics_method_params.shot_id,
        )
        return None
    return params.physics_method_params.times[indices_flattop]

_get_domain_d3d

_get_domain_d3d(params: DomainSettingParams) -> np.ndarray

Get the flattop domain for D3D tokamak.

PARAMETER	DESCRIPTION
params	Parameters containing physics method and tokamak information. TYPE: DomainSettingParams

RETURNS	DESCRIPTION
ndarray	The flattop timebase domain for D3D.

Source code in disruption_py/settings/domain_setting.py

def _get_domain_d3d(self, params: DomainSettingParams) -> np.ndarray:
    """
    Get the flattop domain for D3D tokamak.

    Parameters
    ----------
    params : DomainSettingParams
        Parameters containing physics method and tokamak information.

    Returns
    -------
    np.ndarray
        The flattop timebase domain for D3D.
    """
    try:
        (
            ip_prog,
            t_ip_prog,
        ) = params.physics_method_params.mds_conn.get_data_with_dims(
            f"ptdata('iptipp', {params.physics_method_params.shot_id})",
            tree_name="d3d",
        )
        t_ip_prog = t_ip_prog / 1.0e3  # [ms] -> [s]
        polarity = np.unique(
            params.physics_method_params.mds_conn.get_data(
                f"ptdata('iptdirect', {params.physics_method_params.shot_id})",
                tree_name="d3d",
            )
        )
        if len(polarity) > 1:
            params.logger.info(
                (
                    "[Shot %s]: Polarity of Ip target is not constant. "
                    "Using value at first timestep."
                ),
                params.physics_method_params.shot_id,
            )
            params.logger.debug(
                "[Shot %s]: Polarity array %s",
                params.physics_method_params.shot_id,
                polarity,
            )
            polarity = polarity[0]
        ip_prog = ip_prog * polarity
        dipprog_dt = np.gradient(ip_prog, t_ip_prog)
        ip_prog = interp1(
            t_ip_prog, ip_prog, params.physics_method_params.times, "linear"
        )
        dipprog_dt = interp1(
            t_ip_prog, dipprog_dt, params.physics_method_params.times, "linear"
        )
    except mdsExceptions.MdsException:
        params.logger.warning(
            "[Shot %s]: Could not find flattop timebase. Defaulting to full timebase.",
            params.physics_method_params.shot_id,
        )
        params.logger.debug(
            "[Shot %s]: %s",
            params.physics_method_params.shot_id,
            traceback.format_exc(),
        )
        return None
    epsoff, t_epsoff = params.physics_method_params.mds_conn.get_data_with_dims(
        f"ptdata('epsoff', {params.physics_method_params.shot_id})", tree_name="d3d"
    )
    # [ms] -> [s] # Avoid problem with simultaneity of epsoff being triggered
    # exactly on the last time sample
    t_epsoff = t_epsoff / 1.0e3 + 0.001
    epsoff = interp1(t_epsoff, epsoff, params.physics_method_params.times, "linear")
    railed_indices = np.where(np.abs(epsoff) > 0.5)
    power_supply_railed = np.zeros(len(params.physics_method_params.times))
    power_supply_railed[railed_indices] = 1
    indices_flattop = np.where(
        (np.abs(dipprog_dt) <= 2.0e3)
        & (np.abs(ip_prog) > 100e3)
        & (power_supply_railed != 1)
    )
    return params.physics_method_params.times[indices_flattop]

disruption_py.settings.domain_setting.FullDomainSetting

Bases: DomainSetting

Domain setting that uses the full timebase.

Source code in disruption_py/settings/domain_setting.py

class FullDomainSetting(DomainSetting):
    """
    Domain setting that uses the full timebase.
    """

    def _get_domain(self, params: DomainSettingParams) -> np.ndarray:
        """
        Get the full timebase domain.

        Parameters
        ----------
        params : DomainSettingParams
            Parameters containing physics method and tokamak information.

        Returns
        -------
        np.ndarray
            The full timebase.
        """
        return params.physics_method_params.times

_get_domain

_get_domain(params: DomainSettingParams) -> np.ndarray

Get the full timebase domain.

PARAMETER	DESCRIPTION
params	Parameters containing physics method and tokamak information. TYPE: DomainSettingParams

RETURNS	DESCRIPTION
ndarray	The full timebase.

Source code in disruption_py/settings/domain_setting.py

def _get_domain(self, params: DomainSettingParams) -> np.ndarray:
    """
    Get the full timebase domain.

    Parameters
    ----------
    params : DomainSettingParams
        Parameters containing physics method and tokamak information.

    Returns
    -------
    np.ndarray
        The full timebase.
    """
    return params.physics_method_params.times

disruption_py.settings.domain_setting.RampupAndFlattopDomainSetting

Bases: DomainSetting

Specifies that the ramp up and flattop domain of the timebase should be used.

Source code in disruption_py/settings/domain_setting.py

class RampupAndFlattopDomainSetting(DomainSetting):
    """
    Specifies that the ramp up and flattop domain of the timebase should be used.
    """

    def __init__(self):
        """
        Initialize the RampupAndFlattopDomainSetting with tokamak-specific overrides.
        """
        self.tokamak_overrides = {
            Tokamak.CMOD: self._get_domain_cmod,
        }

    def _get_domain(self, params: DomainSettingParams) -> np.ndarray:
        """
        Get the ramp-up and flattop domain for the given tokamak.

        Parameters
        ----------
        params : DomainSettingParams
            Parameters containing physics method and tokamak information.

        Returns
        -------
        np.ndarray
            The ramp-up and flattop timebase domain.
        """
        raise ValueError(
            f"ramp up and flattop domain not defined for tokamak: {params.tokamak}"
        )

    def _get_domain_cmod(self, params: DomainSettingParams) -> np.ndarray:
        """
        Get the ramp-up and flattop domain for CMOD tokamak.

        Parameters
        ----------
        params : DomainSettingParams
            Parameters containing physics method and tokamak information.

        Returns
        -------
        np.ndarray
            The ramp-up and flattop timebase domain for CMOD.
        """
        ip_parameters = CmodPhysicsMethods.get_ip_parameters(
            params=params.physics_method_params
        )
        ipprog, dipprog_dt = ip_parameters["ip_prog"], ip_parameters["dipprog_dt"]
        indices_flattop_1 = np.where(np.abs(dipprog_dt) <= 6e4)[0]
        indices_flattop_2 = np.where(np.abs(ipprog) > 1.0e5)[0]
        indices_flattop = np.intersect1d(indices_flattop_1, indices_flattop_2)
        if len(indices_flattop) == 0:
            params.logger.warning(
                "[Shot %s]: Could not find flattop timebase. Defaulting to full timebase.",
                params.physics_method_params.shot_id,
            )
            return None
        end_index = np.max(indices_flattop)
        return params.physics_method_params.times[:end_index]

__init__

__init__()

Source code in disruption_py/settings/domain_setting.py

def __init__(self):
    """
    Initialize the RampupAndFlattopDomainSetting with tokamak-specific overrides.
    """
    self.tokamak_overrides = {
        Tokamak.CMOD: self._get_domain_cmod,
    }

_get_domain

_get_domain(params: DomainSettingParams) -> np.ndarray

Get the ramp-up and flattop domain for the given tokamak.

PARAMETER	DESCRIPTION
params	Parameters containing physics method and tokamak information. TYPE: DomainSettingParams

RETURNS	DESCRIPTION
ndarray	The ramp-up and flattop timebase domain.

Source code in disruption_py/settings/domain_setting.py

def _get_domain(self, params: DomainSettingParams) -> np.ndarray:
    """
    Get the ramp-up and flattop domain for the given tokamak.

    Parameters
    ----------
    params : DomainSettingParams
        Parameters containing physics method and tokamak information.

    Returns
    -------
    np.ndarray
        The ramp-up and flattop timebase domain.
    """
    raise ValueError(
        f"ramp up and flattop domain not defined for tokamak: {params.tokamak}"
    )

_get_domain_cmod

_get_domain_cmod(params: DomainSettingParams) -> np.ndarray

Get the ramp-up and flattop domain for CMOD tokamak.

PARAMETER	DESCRIPTION
params	Parameters containing physics method and tokamak information. TYPE: DomainSettingParams

RETURNS	DESCRIPTION
ndarray	The ramp-up and flattop timebase domain for CMOD.

Source code in disruption_py/settings/domain_setting.py

def _get_domain_cmod(self, params: DomainSettingParams) -> np.ndarray:
    """
    Get the ramp-up and flattop domain for CMOD tokamak.

    Parameters
    ----------
    params : DomainSettingParams
        Parameters containing physics method and tokamak information.

    Returns
    -------
    np.ndarray
        The ramp-up and flattop timebase domain for CMOD.
    """
    ip_parameters = CmodPhysicsMethods.get_ip_parameters(
        params=params.physics_method_params
    )
    ipprog, dipprog_dt = ip_parameters["ip_prog"], ip_parameters["dipprog_dt"]
    indices_flattop_1 = np.where(np.abs(dipprog_dt) <= 6e4)[0]
    indices_flattop_2 = np.where(np.abs(ipprog) > 1.0e5)[0]
    indices_flattop = np.intersect1d(indices_flattop_1, indices_flattop_2)
    if len(indices_flattop) == 0:
        params.logger.warning(
            "[Shot %s]: Could not find flattop timebase. Defaulting to full timebase.",
            params.physics_method_params.shot_id,
        )
        return None
    end_index = np.max(indices_flattop)
    return params.physics_method_params.times[:end_index]

disruption_py.settings.domain_setting.resolve_domain_setting

resolve_domain_setting(
    domain_setting: DomainSettingType,
) -> DomainSetting

Resolve the given domain setting into a DomainSetting object.

PARAMETER	DESCRIPTION
domain_setting	The domain setting to resolve, which can be a string, DomainSetting, or dictionary. TYPE: DomainSettingType

RETURNS	DESCRIPTION
DomainSetting	The resolved domain setting.

Source code in disruption_py/settings/domain_setting.py

def resolve_domain_setting(
    domain_setting: DomainSettingType,
) -> DomainSetting:
    """
    Resolve the given domain setting into a DomainSetting object.

    Parameters
    ----------
    domain_setting : DomainSettingType
        The domain setting to resolve, which can be a string, DomainSetting, or dictionary.

    Returns
    -------
    DomainSetting
        The resolved domain setting.
    """
    if domain_setting is None:
        return FullDomainSetting()

    if isinstance(domain_setting, DomainSetting):
        return domain_setting

    if isinstance(domain_setting, str):
        domain_setting_object = _domain_setting_mappings.get(domain_setting, None)
        if domain_setting_object is not None:
            return domain_setting_object

    if isinstance(domain_setting, dict):
        return DomainSettingDict(domain_setting)

    raise ValueError(f"Invalid domain setting: {domain_setting}")

Custom Implementations

Custom implementations of a domain setting must inherit from the DomainSetting abstract class, implementing the abstract methods.

This module provides classes and methods to specify different domain settings for the timebase used, including full, flattop, and ramp-up domains.

disruption_py.settings.domain_setting.DomainSetting

Bases: ABC

Abstract base class for domain settings that modify the timebase domain in physics method parameters.

Source code in disruption_py/settings/domain_setting.py

class DomainSetting(ABC):
    """
    Abstract base class for domain settings that modify the timebase
    domain in physics method parameters.
    """

    def get_domain(self, params: DomainSettingParams) -> np.ndarray:
        """
        Get the domain of the timebase to use.

        Parameters
        ----------
        params : DomainSettingParams
            The parameters including the physics method and tokamak information.

        Returns
        -------
        np.ndarray
            Array representing the timebase domain.
        """
        if hasattr(self, "tokamak_overrides"):
            if params.tokamak in self.tokamak_overrides:
                return self.tokamak_overrides[params.tokamak](params)
        return self._get_domain(params)

    @abstractmethod
    def _get_domain(self, params: DomainSettingParams) -> np.ndarray:
        """
        Abstract method to get the modified times attribute of the physics method
        parameters based on the domain.

        Parameters
        ----------
        params : DomainSettingParams
            The parameters for domain modification.

        Returns
        -------
        np.ndarray
            Array of modified times.
        """

_get_domain abstractmethod

_get_domain(params: DomainSettingParams) -> np.ndarray

Abstract method to get the modified times attribute of the physics method parameters based on the domain.

PARAMETER	DESCRIPTION
params	The parameters for domain modification. TYPE: DomainSettingParams

RETURNS	DESCRIPTION
ndarray	Array of modified times.

Source code in disruption_py/settings/domain_setting.py

@abstractmethod
def _get_domain(self, params: DomainSettingParams) -> np.ndarray:
    """
    Abstract method to get the modified times attribute of the physics method
    parameters based on the domain.

    Parameters
    ----------
    params : DomainSettingParams
        The parameters for domain modification.

    Returns
    -------
    np.ndarray
        Array of modified times.
    """

get_domain

get_domain(params: DomainSettingParams) -> np.ndarray

Get the domain of the timebase to use.

PARAMETER	DESCRIPTION
params	The parameters including the physics method and tokamak information. TYPE: DomainSettingParams

RETURNS	DESCRIPTION
ndarray	Array representing the timebase domain.

Source code in disruption_py/settings/domain_setting.py

def get_domain(self, params: DomainSettingParams) -> np.ndarray:
    """
    Get the domain of the timebase to use.

    Parameters
    ----------
    params : DomainSettingParams
        The parameters including the physics method and tokamak information.

    Returns
    -------
    np.ndarray
        Array representing the timebase domain.
    """
    if hasattr(self, "tokamak_overrides"):
        if params.tokamak in self.tokamak_overrides:
            return self.tokamak_overrides[params.tokamak](params)
    return self._get_domain(params)

disruption_py.settings.domain_setting.DomainSettingParams dataclass

Parameters passed to the get_domain method.

ATTRIBUTE	DESCRIPTION
physics_method_params	The physics method parameters that include timebase information. TYPE: PhysicsMethodParams
tokamak	The tokamak instance for which the domain is set. TYPE: Tokamak
logger	Logger instance for logging relevant messages. TYPE: Logger

Source code in disruption_py/settings/domain_setting.py

@dataclass
class DomainSettingParams:
    """
    Parameters passed to the `get_domain` method.

    Attributes
    ----------
    physics_method_params : PhysicsMethodParams
        The physics method parameters that include timebase information.
    tokamak : Tokamak
        The tokamak instance for which the domain is set.
    logger : Logger
        Logger instance for logging relevant messages.
    """

    physics_method_params: PhysicsMethodParams
    tokamak: Tokamak
    logger: Logger