Output Setting

Overview¤

A module for handling the output setting passed in the get_shots_data method. The output setting is used to handle the output of data from DisruptionPy as it is retrieved. This may include collecting all the data from a request and returning it as a list or streaming outputted data to a file as it is retrieved.

This module defines the abstract class OutputSetting that can have subclasses passed as the output_setting argument to the get_shots_data method. It also provides built-in classes and mappings to easily set the output type for common use cases.

Usage¤

Currently, these are the options that can be passed as the output_setting argument to get_shots_data:

An instance of a subclass of OutputSetting

A string identifier in the _output_setting_mappings dictionary:

_output_setting_mappings: Dict[str, OutputSetting] = {
    "list": ListOutputSetting(),
    "dataframe": DataFrameOutputSetting(),
    "dict": DictOutputSetting(),
}

A file path as a string with its suffix mapped to an OutputSetting type in the _file_suffix_to_output_setting dictionary:

_file_suffix_to_output_setting: Dict[str, Type[OutputSetting]] = {
    ".h5": HDF5OutputSetting,
    ".hdf5": HDF5OutputSetting,
    ".csv": BatchedCSVOutputSetting,
}

A dictionary mapping tokamak type strings to the desired OutputSetting for that tokamak. E.g. {'cmod': 'list'}.
A Python list of any other output type request option that can be passed as the OutputSetting argument to get_shots_data (all options listed previously). See ListOutputSetting for more details.

Built-in Implemenations¤

Handles output settings for retrieving and saving shot data.

This module provides classes and methods to manage various output settings for shot data, including saving to files, databases, lists, dictionaries, and dataframes.

disruption_py.settings.output_setting.BatchedCSVOutputSetting ¤

Bases: OutputSetting

Stream outputted data to a single CSV file in batches.

Source code in disruption_py/settings/output_setting.py

class BatchedCSVOutputSetting(OutputSetting):
    """
    Stream outputted data to a single CSV file in batches.
    """

    def __init__(self, filepath, batch_size=100, clear_file=True):
        """
        Initialize the BatchedCSVOutputSetting.

        Parameters
        ----------
        filepath : str
            The path to the CSV file where data will be written.
        batch_size : int, optional
            The number of records to write to the CSV file in one batch (default is 100).
        clear_file : bool, optional
            Whether to clear the file at the beginning (default is True).
        """
        self.filepath = filepath
        self.batch_size = batch_size
        self.clear_file = clear_file
        self.batch_data = []  # Initialize an empty list to hold batched data
        self.output_shot_count = 0

        # Clear the file at the beginning if required
        if self.clear_file and os.path.exists(filepath):
            os.remove(filepath)

        self.results: pd.DataFrame = pd.DataFrame()

    def _output_shot(self, params: OutputSettingParams):
        """
        Append the current result to the batch data list and write to CSV if
        batch size is reached.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters containing the result to be outputted.
        """
        # Append the current result to the batch data list
        self.batch_data.append(params.result)

        # Check if the batch size has been reached
        if len(self.batch_data) >= self.batch_size:
            self._write_batch_to_csv()

        self.output_shot_count += 1
        self.results = safe_df_concat(self.results, [params.result])

    def _write_batch_to_csv(self):
        """
        Write the current batch of data to the CSV file.
        """
        file_exists = os.path.isfile(self.filepath)
        combined_df = safe_df_concat(pd.DataFrame(), self.batch_data)
        combined_df.to_csv(
            self.filepath, mode="a", index=False, header=(not file_exists)
        )
        self.batch_data.clear()

    def get_results(self, params: CompleteOutputSettingParams):
        """
        Write any remaining batched data to the CSV file before returning results.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for retrieving results.

        Returns
        -------
        pd.DataFrame
            The DataFrame containing the results.
        """
        # Write any remaining batched data to the CSV file before returning results
        if self.batch_data:
            self._write_batch_to_csv()
        return self.results

    def stream_output_cleanup(self, params: CompleteOutputSettingParams):
        """
        Clean up the output stream by resetting results.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for cleaning up the output stream.
        """
        self.results = pd.DataFrame()

init ¤

__init__(filepath, batch_size=100, clear_file=True)

PARAMETER	DESCRIPTION
`filepath`	The path to the CSV file where data will be written. TYPE: `str`
`batch_size`	The number of records to write to the CSV file in one batch (default is 100). TYPE: `int` DEFAULT: `100`
`clear_file`	Whether to clear the file at the beginning (default is True). TYPE: `bool` DEFAULT: `True`

Source code in disruption_py/settings/output_setting.py

def __init__(self, filepath, batch_size=100, clear_file=True):
    """
    Initialize the BatchedCSVOutputSetting.

    Parameters
    ----------
    filepath : str
        The path to the CSV file where data will be written.
    batch_size : int, optional
        The number of records to write to the CSV file in one batch (default is 100).
    clear_file : bool, optional
        Whether to clear the file at the beginning (default is True).
    """
    self.filepath = filepath
    self.batch_size = batch_size
    self.clear_file = clear_file
    self.batch_data = []  # Initialize an empty list to hold batched data
    self.output_shot_count = 0

    # Clear the file at the beginning if required
    if self.clear_file and os.path.exists(filepath):
        os.remove(filepath)

    self.results: pd.DataFrame = pd.DataFrame()

_output_shot ¤

_output_shot(params: OutputSettingParams)

Append the current result to the batch data list and write to CSV if batch size is reached.

PARAMETER	DESCRIPTION
`params`	The parameters containing the result to be outputted. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def _output_shot(self, params: OutputSettingParams):
    """
    Append the current result to the batch data list and write to CSV if
    batch size is reached.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters containing the result to be outputted.
    """
    # Append the current result to the batch data list
    self.batch_data.append(params.result)

    # Check if the batch size has been reached
    if len(self.batch_data) >= self.batch_size:
        self._write_batch_to_csv()

    self.output_shot_count += 1
    self.results = safe_df_concat(self.results, [params.result])

_write_batch_to_csv ¤

_write_batch_to_csv()

Write the current batch of data to the CSV file.

Source code in disruption_py/settings/output_setting.py

def _write_batch_to_csv(self):
    """
    Write the current batch of data to the CSV file.
    """
    file_exists = os.path.isfile(self.filepath)
    combined_df = safe_df_concat(pd.DataFrame(), self.batch_data)
    combined_df.to_csv(
        self.filepath, mode="a", index=False, header=(not file_exists)
    )
    self.batch_data.clear()

get_results ¤

get_results(params: CompleteOutputSettingParams)

Write any remaining batched data to the CSV file before returning results.

PARAMETER	DESCRIPTION
`params`	The parameters for retrieving results. TYPE: `CompleteOutputSettingParams`

RETURNS	DESCRIPTION
`DataFrame`	The DataFrame containing the results.

Source code in disruption_py/settings/output_setting.py

def get_results(self, params: CompleteOutputSettingParams):
    """
    Write any remaining batched data to the CSV file before returning results.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for retrieving results.

    Returns
    -------
    pd.DataFrame
        The DataFrame containing the results.
    """
    # Write any remaining batched data to the CSV file before returning results
    if self.batch_data:
        self._write_batch_to_csv()
    return self.results

stream_output_cleanup ¤

stream_output_cleanup(params: CompleteOutputSettingParams)

Clean up the output stream by resetting results.

PARAMETER	DESCRIPTION
`params`	The parameters for cleaning up the output stream. TYPE: `CompleteOutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def stream_output_cleanup(self, params: CompleteOutputSettingParams):
    """
    Clean up the output stream by resetting results.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for cleaning up the output stream.
    """
    self.results = pd.DataFrame()

disruption_py.settings.output_setting.CSVOutputSetting ¤

Bases: OutputSetting

Outputs shot data to a single CSV file. Not recommended when retrieving a large number of shots.

Source code in disruption_py/settings/output_setting.py

class CSVOutputSetting(OutputSetting):
    """
    Outputs shot data to a single CSV file.
    Not recommended when retrieving a large number of shots.
    """

    def __init__(
        self, filepath: str, flexible_columns: bool = True, clear_file: bool = True
    ):
        """
        Initialize CSVOutputSetting with a file path and options.

        Parameters
        ----------
        filepath : str
            The path to the CSV file.
        flexible_columns : bool, optional
            If True, allows for flexible columns in the CSV (default is True).
        clear_file : bool, optional
            If True, clears the file if it exists (default is True).
        """
        self.filepath = filepath
        self.flexible_columns = flexible_columns
        self.output_shot_count = 0
        if clear_file and os.path.exists(filepath):
            os.remove(filepath)
        self.results: pd.DataFrame = pd.DataFrame()

    def _output_shot(self, params: OutputSettingParams):
        """
        Output a single shot to the CSV file.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters for outputting shot results.
        """
        file_exists = os.path.isfile(self.filepath)
        if self.flexible_columns:
            if file_exists:
                existing_df = pd.read_csv(self.filepath)
                combined_df = safe_df_concat(existing_df, [params.result])
            else:
                combined_df = params.result

            combined_df.to_csv(self.filepath, index=False)
        else:
            params.result.to_csv(
                self.filepath, mode="a", index=False, header=(not file_exists)
            )
        self.output_shot_count += 1
        self.results = safe_df_concat(self.results, [params.result])

    def get_results(self, params: CompleteOutputSettingParams):
        """
        Get the accumulated results.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.

        Returns
        -------
        pd.DataFrame
            The combined DataFrame of results.
        """
        return self.results

    def stream_output_cleanup(self, params: CompleteOutputSettingParams):
        """
        Cleanup the results DataFrame.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.
        """
        self.results = pd.DataFrame()

init ¤

__init__(
    filepath: str,
    flexible_columns: bool = True,
    clear_file: bool = True,
)

PARAMETER	DESCRIPTION
`filepath`	The path to the CSV file. TYPE: `str`
`flexible_columns`	If True, allows for flexible columns in the CSV (default is True). TYPE: `bool` DEFAULT: `True`
`clear_file`	If True, clears the file if it exists (default is True). TYPE: `bool` DEFAULT: `True`

Source code in disruption_py/settings/output_setting.py

def __init__(
    self, filepath: str, flexible_columns: bool = True, clear_file: bool = True
):
    """
    Initialize CSVOutputSetting with a file path and options.

    Parameters
    ----------
    filepath : str
        The path to the CSV file.
    flexible_columns : bool, optional
        If True, allows for flexible columns in the CSV (default is True).
    clear_file : bool, optional
        If True, clears the file if it exists (default is True).
    """
    self.filepath = filepath
    self.flexible_columns = flexible_columns
    self.output_shot_count = 0
    if clear_file and os.path.exists(filepath):
        os.remove(filepath)
    self.results: pd.DataFrame = pd.DataFrame()

_output_shot ¤

_output_shot(params: OutputSettingParams)

Output a single shot to the CSV file.

PARAMETER	DESCRIPTION
`params`	The parameters for outputting shot results. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def _output_shot(self, params: OutputSettingParams):
    """
    Output a single shot to the CSV file.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters for outputting shot results.
    """
    file_exists = os.path.isfile(self.filepath)
    if self.flexible_columns:
        if file_exists:
            existing_df = pd.read_csv(self.filepath)
            combined_df = safe_df_concat(existing_df, [params.result])
        else:
            combined_df = params.result

        combined_df.to_csv(self.filepath, index=False)
    else:
        params.result.to_csv(
            self.filepath, mode="a", index=False, header=(not file_exists)
        )
    self.output_shot_count += 1
    self.results = safe_df_concat(self.results, [params.result])

get_results ¤

get_results(params: CompleteOutputSettingParams)

Get the accumulated results.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

RETURNS	DESCRIPTION
`DataFrame`	The combined DataFrame of results.

Source code in disruption_py/settings/output_setting.py

def get_results(self, params: CompleteOutputSettingParams):
    """
    Get the accumulated results.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.

    Returns
    -------
    pd.DataFrame
        The combined DataFrame of results.
    """
    return self.results

stream_output_cleanup ¤

stream_output_cleanup(params: CompleteOutputSettingParams)

Cleanup the results DataFrame.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def stream_output_cleanup(self, params: CompleteOutputSettingParams):
    """
    Cleanup the results DataFrame.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.
    """
    self.results = pd.DataFrame()

disruption_py.settings.output_setting.DataFrameOutputSetting ¤

Bases: OutputSetting

Outputs all shot data as a single DataFrame.

Source code in disruption_py/settings/output_setting.py

class DataFrameOutputSetting(OutputSetting):
    """
    Outputs all shot data as a single DataFrame.
    """

    def __init__(self):
        """Initialize DataFrameOutputSetting with an empty DataFrame."""
        self.results: pd.DataFrame = pd.DataFrame()

    def _output_shot(self, params: OutputSettingParams):
        """
        Output a single shot by concatenating the result to the DataFrame.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters for outputting shot results.
        """
        self.results = safe_df_concat(self.results, [params.result])

    def get_results(self, params: CompleteOutputSettingParams):
        """
        Get the accumulated results.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.

        Returns
        -------
        pd.DataFrame
            The combined DataFrame of results.
        """
        return self.results

    def stream_output_cleanup(self, params: CompleteOutputSettingParams):
        """
        Cleanup the results DataFrame.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.
        """
        self.results = pd.DataFrame()

init ¤

__init__()

Source code in disruption_py/settings/output_setting.py

def __init__(self):
    """Initialize DataFrameOutputSetting with an empty DataFrame."""
    self.results: pd.DataFrame = pd.DataFrame()

_output_shot ¤

_output_shot(params: OutputSettingParams)

Output a single shot by concatenating the result to the DataFrame.

PARAMETER	DESCRIPTION
`params`	The parameters for outputting shot results. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def _output_shot(self, params: OutputSettingParams):
    """
    Output a single shot by concatenating the result to the DataFrame.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters for outputting shot results.
    """
    self.results = safe_df_concat(self.results, [params.result])

get_results ¤

get_results(params: CompleteOutputSettingParams)

Get the accumulated results.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

RETURNS	DESCRIPTION
`DataFrame`	The combined DataFrame of results.

Source code in disruption_py/settings/output_setting.py

def get_results(self, params: CompleteOutputSettingParams):
    """
    Get the accumulated results.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.

    Returns
    -------
    pd.DataFrame
        The combined DataFrame of results.
    """
    return self.results

stream_output_cleanup ¤

stream_output_cleanup(params: CompleteOutputSettingParams)

Cleanup the results DataFrame.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def stream_output_cleanup(self, params: CompleteOutputSettingParams):
    """
    Cleanup the results DataFrame.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.
    """
    self.results = pd.DataFrame()

disruption_py.settings.output_setting.DictOutputSetting ¤

Bases: OutputSetting

Outputs shot data as a dict of DataFrames keyed by shot number.

Source code in disruption_py/settings/output_setting.py

class DictOutputSetting(OutputSetting):
    """
    Outputs shot data as a dict of DataFrames keyed by shot number.
    """

    def __init__(self):
        """Initialize DictOutputSetting with an empty results dictionary."""
        self.results = {}

    def _output_shot(self, params: OutputSettingParams):
        """
        Output a single shot by storing the result in the dictionary.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters for outputting shot results.
        """
        self.results[params.shot_id] = params.result

    def get_results(self, params: CompleteOutputSettingParams):
        """
        Get the accumulated results.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.

        Returns
        -------
        Dict[int, pd.DataFrame]
            The dictionary of results keyed by shot number.
        """
        return self.results

    def stream_output_cleanup(self, params: CompleteOutputSettingParams):
        """
        Cleanup the results dictionary.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.
        """
        self.results = {}

init ¤

__init__()

Source code in disruption_py/settings/output_setting.py

def __init__(self):
    """Initialize DictOutputSetting with an empty results dictionary."""
    self.results = {}

_output_shot ¤

_output_shot(params: OutputSettingParams)

Output a single shot by storing the result in the dictionary.

PARAMETER	DESCRIPTION
`params`	The parameters for outputting shot results. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def _output_shot(self, params: OutputSettingParams):
    """
    Output a single shot by storing the result in the dictionary.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters for outputting shot results.
    """
    self.results[params.shot_id] = params.result

get_results ¤

get_results(params: CompleteOutputSettingParams)

Get the accumulated results.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

RETURNS	DESCRIPTION
`Dict[int, DataFrame]`	The dictionary of results keyed by shot number.

Source code in disruption_py/settings/output_setting.py

def get_results(self, params: CompleteOutputSettingParams):
    """
    Get the accumulated results.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.

    Returns
    -------
    Dict[int, pd.DataFrame]
        The dictionary of results keyed by shot number.
    """
    return self.results

stream_output_cleanup ¤

stream_output_cleanup(params: CompleteOutputSettingParams)

Cleanup the results dictionary.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def stream_output_cleanup(self, params: CompleteOutputSettingParams):
    """
    Cleanup the results dictionary.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.
    """
    self.results = {}

disruption_py.settings.output_setting.HDF5OutputSetting ¤

Bases: OutputSetting

Stream outputted data to an HDF5 file. Data for each shot is stored in a table under the key df_SHOTID.

Source code in disruption_py/settings/output_setting.py

class HDF5OutputSetting(OutputSetting):
    """
    Stream outputted data to an HDF5 file. Data for each shot is stored in a table
    under the key `df_SHOTID`.
    """

    def __init__(self, filepath: str, only_output_numeric: bool = False):
        """
        Initialize HDF5OutputSetting with a file path and numeric output option.

        Parameters
        ----------
        filepath : str
            The path to the HDF5 file.
        only_output_numeric : bool, optional
            If True, only numeric data will be outputted (default is False) and
            non-numeric quantities like commit hash will be excluded.
        """
        self.filepath = filepath
        self.output_shot_count = 0
        self.only_output_numeric = only_output_numeric
        self.results: pd.DataFrame = pd.DataFrame()

    def _output_shot(self, params: OutputSettingParams):
        """
        Output a single shot to the HDF5 file.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters for outputting shot results.
        """
        mode = "a" if self.output_shot_count > 0 else "w"

        if self.only_output_numeric:
            output_result = params.result.select_dtypes([np.number])
        else:
            output_result = params.result

        output_result.to_hdf(
            self.filepath,
            key=f"df_{params.shot_id}",
            format="table",
            complib="blosc",
            mode=mode,
        )
        self.output_shot_count += 1
        self.results = safe_df_concat(self.results, [params.result])

    def get_results(self, params: CompleteOutputSettingParams):
        """
        Get the accumulated results.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.

        Returns
        -------
        pd.DataFrame
            The combined DataFrame of results.
        """
        return self.results

    def stream_output_cleanup(self, params: CompleteOutputSettingParams):
        """
        Cleanup the results DataFrame.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.
        """
        self.results = pd.DataFrame()

init ¤

__init__(filepath: str, only_output_numeric: bool = False)

PARAMETER	DESCRIPTION
`filepath`	The path to the HDF5 file. TYPE: `str`
`only_output_numeric`	If True, only numeric data will be outputted (default is False) and non-numeric quantities like commit hash will be excluded. TYPE: `bool` DEFAULT: `False`

Source code in disruption_py/settings/output_setting.py

def __init__(self, filepath: str, only_output_numeric: bool = False):
    """
    Initialize HDF5OutputSetting with a file path and numeric output option.

    Parameters
    ----------
    filepath : str
        The path to the HDF5 file.
    only_output_numeric : bool, optional
        If True, only numeric data will be outputted (default is False) and
        non-numeric quantities like commit hash will be excluded.
    """
    self.filepath = filepath
    self.output_shot_count = 0
    self.only_output_numeric = only_output_numeric
    self.results: pd.DataFrame = pd.DataFrame()

_output_shot ¤

_output_shot(params: OutputSettingParams)

Output a single shot to the HDF5 file.

PARAMETER	DESCRIPTION
`params`	The parameters for outputting shot results. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def _output_shot(self, params: OutputSettingParams):
    """
    Output a single shot to the HDF5 file.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters for outputting shot results.
    """
    mode = "a" if self.output_shot_count > 0 else "w"

    if self.only_output_numeric:
        output_result = params.result.select_dtypes([np.number])
    else:
        output_result = params.result

    output_result.to_hdf(
        self.filepath,
        key=f"df_{params.shot_id}",
        format="table",
        complib="blosc",
        mode=mode,
    )
    self.output_shot_count += 1
    self.results = safe_df_concat(self.results, [params.result])

get_results ¤

get_results(params: CompleteOutputSettingParams)

Get the accumulated results.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

RETURNS	DESCRIPTION
`DataFrame`	The combined DataFrame of results.

Source code in disruption_py/settings/output_setting.py

def get_results(self, params: CompleteOutputSettingParams):
    """
    Get the accumulated results.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.

    Returns
    -------
    pd.DataFrame
        The combined DataFrame of results.
    """
    return self.results

stream_output_cleanup ¤

stream_output_cleanup(params: CompleteOutputSettingParams)

Cleanup the results DataFrame.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def stream_output_cleanup(self, params: CompleteOutputSettingParams):
    """
    Cleanup the results DataFrame.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.
    """
    self.results = pd.DataFrame()

disruption_py.settings.output_setting.ListOutputSetting ¤

Bases: OutputSetting

Outputs shot data as a list of DataFrames.

Source code in disruption_py/settings/output_setting.py

class ListOutputSetting(OutputSetting):
    """
    Outputs shot data as a list of DataFrames.
    """

    def __init__(self):
        """Initialize ListOutputSetting with an empty results list."""
        self.results = []

    def _output_shot(self, params: OutputSettingParams):
        """
        Output a single shot by appending the result to the list.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters for outputting shot results.
        """
        self.results.append(params.result)

    def get_results(self, params: CompleteOutputSettingParams):
        """
        Get the accumulated results.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.

        Returns
        -------
        List[pd.DataFrame]
            The list of results.
        """
        return self.results

    def stream_output_cleanup(self, params: CompleteOutputSettingParams):
        """
        Cleanup the results list.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.
        """
        self.results = []

init ¤

__init__()

Source code in disruption_py/settings/output_setting.py

def __init__(self):
    """Initialize ListOutputSetting with an empty results list."""
    self.results = []

_output_shot ¤

_output_shot(params: OutputSettingParams)

Output a single shot by appending the result to the list.

PARAMETER	DESCRIPTION
`params`	The parameters for outputting shot results. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def _output_shot(self, params: OutputSettingParams):
    """
    Output a single shot by appending the result to the list.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters for outputting shot results.
    """
    self.results.append(params.result)

get_results ¤

get_results(params: CompleteOutputSettingParams)

Get the accumulated results.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

RETURNS	DESCRIPTION
`List[DataFrame]`	The list of results.

Source code in disruption_py/settings/output_setting.py

def get_results(self, params: CompleteOutputSettingParams):
    """
    Get the accumulated results.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.

    Returns
    -------
    List[pd.DataFrame]
        The list of results.
    """
    return self.results

stream_output_cleanup ¤

stream_output_cleanup(params: CompleteOutputSettingParams)

Cleanup the results list.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def stream_output_cleanup(self, params: CompleteOutputSettingParams):
    """
    Cleanup the results list.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.
    """
    self.results = []

disruption_py.settings.output_setting.SQLOutputSetting ¤

Bases: OutputSetting

Stream outputted data to a SQL table. By default, stream to the test table: disruption_warning_test.

Source code in disruption_py/settings/output_setting.py

class SQLOutputSetting(OutputSetting):
    """
    Stream outputted data to a SQL table. By default, stream to the test table:
    disruption_warning_test.
    """

    def __init__(
        self,
        should_update=False,
        should_override_columns: List[str] = None,
        table_name="disruption_warning_test",
    ):
        """
        Initialize the SQLOutputSetting.

        Parameters
        ----------
        should_update : bool, optional
            Whether to update existing records (default is False).
        should_override_columns : List[str], optional
            List of columns to override in the SQL table (default is None).
        table_name : str, optional
            The name of the SQL table to stream data to (default is
            "disruption_warning_test").
        """
        self.should_update = should_update
        self.should_override_columns = should_override_columns
        self.table_name = table_name
        self.results: pd.DataFrame = pd.DataFrame()

    def _output_shot(self, params: OutputSettingParams):
        """
        Output the current shot data to the SQL table.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters containing the result to be outputted.
        """
        if not params.result.empty and ("shot" in params.result.columns):
            shot_id = params.result["shot"].iloc[0]
            params.database.add_shot_data(
                shot_id=shot_id,
                shot_data=params.result,
                update=self.should_update,
                override_columns=self.should_override_columns,
            )
        else:
            params.logger.warning("No shot id found in result DataFrame")
        self.results = safe_df_concat(self.results, [params.result])

    def get_results(self, params: CompleteOutputSettingParams) -> Any:
        """
        Retrieve the results stored in the SQL output setting.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for retrieving results.

        Returns
        -------
        pd.DataFrame
            The DataFrame containing the results.
        """
        return self.results

    def stream_output_cleanup(self, params: CompleteOutputSettingParams):
        """
        Clean up the output stream by resetting results.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for cleaning up the output stream.
        """
        self.results = pd.DataFrame()

init ¤

__init__(
    should_update=False,
    should_override_columns: List[str] = None,
    table_name="disruption_warning_test",
)

PARAMETER	DESCRIPTION
`should_update`	Whether to update existing records (default is False). TYPE: `bool` DEFAULT: `False`
`should_override_columns`	List of columns to override in the SQL table (default is None). TYPE: `List[str]` DEFAULT: `None`
`table_name`	The name of the SQL table to stream data to (default is "disruption_warning_test"). TYPE: `str` DEFAULT: `'disruption_warning_test'`

Source code in disruption_py/settings/output_setting.py

def __init__(
    self,
    should_update=False,
    should_override_columns: List[str] = None,
    table_name="disruption_warning_test",
):
    """
    Initialize the SQLOutputSetting.

    Parameters
    ----------
    should_update : bool, optional
        Whether to update existing records (default is False).
    should_override_columns : List[str], optional
        List of columns to override in the SQL table (default is None).
    table_name : str, optional
        The name of the SQL table to stream data to (default is
        "disruption_warning_test").
    """
    self.should_update = should_update
    self.should_override_columns = should_override_columns
    self.table_name = table_name
    self.results: pd.DataFrame = pd.DataFrame()

_output_shot ¤

_output_shot(params: OutputSettingParams)

Output the current shot data to the SQL table.

PARAMETER	DESCRIPTION
`params`	The parameters containing the result to be outputted. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def _output_shot(self, params: OutputSettingParams):
    """
    Output the current shot data to the SQL table.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters containing the result to be outputted.
    """
    if not params.result.empty and ("shot" in params.result.columns):
        shot_id = params.result["shot"].iloc[0]
        params.database.add_shot_data(
            shot_id=shot_id,
            shot_data=params.result,
            update=self.should_update,
            override_columns=self.should_override_columns,
        )
    else:
        params.logger.warning("No shot id found in result DataFrame")
    self.results = safe_df_concat(self.results, [params.result])

get_results ¤

get_results(params: CompleteOutputSettingParams) -> Any

Retrieve the results stored in the SQL output setting.

PARAMETER	DESCRIPTION
`params`	The parameters for retrieving results. TYPE: `CompleteOutputSettingParams`

RETURNS	DESCRIPTION
`DataFrame`	The DataFrame containing the results.

Source code in disruption_py/settings/output_setting.py

def get_results(self, params: CompleteOutputSettingParams) -> Any:
    """
    Retrieve the results stored in the SQL output setting.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for retrieving results.

    Returns
    -------
    pd.DataFrame
        The DataFrame containing the results.
    """
    return self.results

stream_output_cleanup ¤

stream_output_cleanup(params: CompleteOutputSettingParams)

Clean up the output stream by resetting results.

PARAMETER	DESCRIPTION
`params`	The parameters for cleaning up the output stream. TYPE: `CompleteOutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def stream_output_cleanup(self, params: CompleteOutputSettingParams):
    """
    Clean up the output stream by resetting results.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for cleaning up the output stream.
    """
    self.results = pd.DataFrame()

disruption_py.settings.output_setting.resolve_output_setting ¤

resolve_output_setting(
    output_setting: OutputSettingType,
) -> OutputSetting

Resolve the output setting to an OutputSetting instance.

PARAMETER	DESCRIPTION
`output_setting`	The output setting to resolve, which can be an instance of OutputSetting, a string, a dictionary, or a list. TYPE: `OutputSettingType`

RETURNS	DESCRIPTION
`OutputSetting`	The resolved OutputSetting instance.

Source code in disruption_py/settings/output_setting.py

def resolve_output_setting(
    output_setting: OutputSettingType,
) -> OutputSetting:
    """
    Resolve the output setting to an OutputSetting instance.

    Parameters
    ----------
    output_setting : OutputSettingType
        The output setting to resolve, which can be an instance of OutputSetting,
        a string, a dictionary, or a list.

    Returns
    -------
    OutputSetting
        The resolved OutputSetting instance.
    """
    if isinstance(output_setting, OutputSetting):
        return output_setting

    if isinstance(output_setting, str):
        output_setting_object = _output_setting_mappings.get(output_setting, None)
        if output_setting_object is not None:
            return output_setting_object

    if isinstance(output_setting, str):
        # assume that it is a file path
        for (
            suffix,
            output_setting_type,
        ) in _file_suffix_to_output_setting.items():
            if output_setting.endswith(suffix):
                return output_setting_type(output_setting)

    if isinstance(output_setting, dict):
        return OutputSettingDict(output_setting)

    if isinstance(output_setting, list):
        return OutputSettingList(output_setting)

    raise ValueError(f"Invalid output processor {output_setting}")

Custom Implementations¤

Custom implementations of output type requests must inherit from the OutputTypeRequest abstract class, implementing the abstract methods.

Handles output settings for retrieving and saving shot data.

This module provides classes and methods to manage various output settings for shot data, including saving to files, databases, lists, dictionaries, and dataframes.

disruption_py.settings.output_setting.OutputSetting ¤

Bases: ABC

OutputSetting abstract class that should be inherited by all output setting classes.

Source code in disruption_py/settings/output_setting.py

class OutputSetting(ABC):
    """
    OutputSetting abstract class that should be inherited by all output setting classes.
    """

    def output_shot(self, params: OutputSettingParams):
        """
        Output a single shot based on the provided parameters.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters for outputting shot results.
        """
        if hasattr(self, "tokamak_overrides"):
            if params.tokamak in self.tokamak_overrides:
                return self.tokamak_overrides[params.tokamak](params)
        return self._output_shot(params)

    @abstractmethod
    def _output_shot(self, params: OutputSettingParams):
        """
        Abstract method implemented by subclasses to handle data output for a
        single shot.

        Parameters
        ----------
        params : OutputSettingParams
            The parameters for outputting shot results.
        """

    def stream_output_cleanup(self, params: CompleteOutputSettingParams):
        """
        Empty method optionally overridden by subclasses to handle cleanup after
        all shots have been output. This may include closing files or other cleanup.
        Subclasses should implement this method so multiple output types can be
        used for the same data without appending to the other's outputted dataframe.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.
        """

    @abstractmethod
    def get_results(self, params: CompleteOutputSettingParams) -> Any:
        """
        Return final output after all shots are processed.

        Parameters
        ----------
        params : CompleteOutputSettingParams
            The parameters for output cleanup and result fetching.

        Returns
        -------
        Any
            The final output results.
        """

_output_shot `abstractmethod` ¤

_output_shot(params: OutputSettingParams)

Abstract method implemented by subclasses to handle data output for a single shot.

PARAMETER	DESCRIPTION
`params`	The parameters for outputting shot results. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

@abstractmethod
def _output_shot(self, params: OutputSettingParams):
    """
    Abstract method implemented by subclasses to handle data output for a
    single shot.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters for outputting shot results.
    """

get_results `abstractmethod` ¤

get_results(params: CompleteOutputSettingParams) -> Any

Return final output after all shots are processed.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

RETURNS	DESCRIPTION
`Any`	The final output results.

Source code in disruption_py/settings/output_setting.py

@abstractmethod
def get_results(self, params: CompleteOutputSettingParams) -> Any:
    """
    Return final output after all shots are processed.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.

    Returns
    -------
    Any
        The final output results.
    """

output_shot ¤

output_shot(params: OutputSettingParams)

Output a single shot based on the provided parameters.

PARAMETER	DESCRIPTION
`params`	The parameters for outputting shot results. TYPE: `OutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def output_shot(self, params: OutputSettingParams):
    """
    Output a single shot based on the provided parameters.

    Parameters
    ----------
    params : OutputSettingParams
        The parameters for outputting shot results.
    """
    if hasattr(self, "tokamak_overrides"):
        if params.tokamak in self.tokamak_overrides:
            return self.tokamak_overrides[params.tokamak](params)
    return self._output_shot(params)

stream_output_cleanup ¤

stream_output_cleanup(params: CompleteOutputSettingParams)

Empty method optionally overridden by subclasses to handle cleanup after all shots have been output. This may include closing files or other cleanup. Subclasses should implement this method so multiple output types can be used for the same data without appending to the other's outputted dataframe.

PARAMETER	DESCRIPTION
`params`	The parameters for output cleanup and result fetching. TYPE: `CompleteOutputSettingParams`

Source code in disruption_py/settings/output_setting.py

def stream_output_cleanup(self, params: CompleteOutputSettingParams):
    """
    Empty method optionally overridden by subclasses to handle cleanup after
    all shots have been output. This may include closing files or other cleanup.
    Subclasses should implement this method so multiple output types can be
    used for the same data without appending to the other's outputted dataframe.

    Parameters
    ----------
    params : CompleteOutputSettingParams
        The parameters for output cleanup and result fetching.
    """

disruption_py.settings.output_setting.OutputSettingParams `dataclass` ¤

Parameters for outputting shot results.

ATTRIBUTE	DESCRIPTION
`shot_id`	Shot ID. TYPE: `int`
`result`	DataFrame of shot results. TYPE: `DataFrame`
`database`	Database connection for retrieving cache data. TYPE: `ShotDatabase`
`tokamak`	The tokamak for which results are being outputted. TYPE: `Tokamak`
`logger`	Logger instance. TYPE: `Logger`

Source code in disruption_py/settings/output_setting.py

@dataclass
class OutputSettingParams:
    """
    Parameters for outputting shot results.

    Attributes
    ----------
    shot_id : int
        Shot ID.
    result : pd.DataFrame
        DataFrame of shot results.
    database : ShotDatabase
        Database connection for retrieving cache data.
    tokamak : Tokamak
        The tokamak for which results are being outputted.
    logger : Logger
        Logger instance.
    """

    shot_id: int
    result: pd.DataFrame
    database: ShotDatabase
    tokamak: Tokamak
    logger: Logger

disruption_py.settings.output_setting.CompleteOutputSettingParams `dataclass` ¤

Parameters for output cleanup and result fetching.

ATTRIBUTE	DESCRIPTION
`tokamak`	The tokamak for which results are being outputted. TYPE: `Tokamak`
`logger`	Logger instance. TYPE: `Logger`

Source code in disruption_py/settings/output_setting.py

@dataclass
class CompleteOutputSettingParams:
    """
    Parameters for output cleanup and result fetching.

    Attributes
    ----------
    tokamak : Tokamak
        The tokamak for which results are being outputted.
    logger : Logger
        Logger instance.
    """

    tokamak: Tokamak
    logger: Logger

Output Setting

Overview¤

Usage¤

Built-in Implemenations¤

disruption_py.settings.output_setting.BatchedCSVOutputSetting ¤

__init__ ¤

_output_shot ¤

_write_batch_to_csv ¤

get_results ¤

stream_output_cleanup ¤

disruption_py.settings.output_setting.CSVOutputSetting ¤

__init__ ¤

_output_shot ¤

get_results ¤

stream_output_cleanup ¤

disruption_py.settings.output_setting.DataFrameOutputSetting ¤

__init__ ¤

_output_shot ¤

get_results ¤

stream_output_cleanup ¤

disruption_py.settings.output_setting.DictOutputSetting ¤

__init__ ¤

_output_shot ¤

get_results ¤

stream_output_cleanup ¤

disruption_py.settings.output_setting.HDF5OutputSetting ¤

__init__ ¤

_output_shot ¤

get_results ¤

stream_output_cleanup ¤

disruption_py.settings.output_setting.ListOutputSetting ¤

__init__ ¤

_output_shot ¤

get_results ¤

stream_output_cleanup ¤

disruption_py.settings.output_setting.SQLOutputSetting ¤

__init__ ¤

_output_shot ¤

get_results ¤

stream_output_cleanup ¤

disruption_py.settings.output_setting.resolve_output_setting ¤

Custom Implementations¤

disruption_py.settings.output_setting.OutputSetting ¤

_output_shot abstractmethod ¤

get_results abstractmethod ¤

output_shot ¤

stream_output_cleanup ¤

disruption_py.settings.output_setting.OutputSettingParams dataclass ¤

disruption_py.settings.output_setting.CompleteOutputSettingParams dataclass ¤

init ¤

init ¤

init ¤

init ¤

init ¤

init ¤

init ¤

_output_shot `abstractmethod` ¤

get_results `abstractmethod` ¤

disruption_py.settings.output_setting.OutputSettingParams `dataclass` ¤

disruption_py.settings.output_setting.CompleteOutputSettingParams `dataclass` ¤