Alarm Definitions

AlarmDefinitions(baze)

Class used for handling definition of alarms.

Source code in echo_baze/baze_root.py

def __init__(self, baze: e_bz.Baze) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    baze : Baze
        Top level object carrying all functionality and the connection handler.

    """
    # check inputs
    if not isinstance(baze, e_bz.Baze):
        raise ValueError(f"baze must be of type Baze, not {type(baze)}")

    self.baze: e_bz.Baze = baze

get(object_models, output_type='dict')

Gets all alarm definitions for the given object models.

The most useful keys/columns returned are:

• id
• eventName
• eventCode.code
• eventCode.vendorEventType
• brakePrg
• canTriggerAllocation
• description
• isOkAlarm
• eventKey
• resetPrg
• type

Parameters:

• object_models

  (list[str]) –

  Desired object models to get the alarm definitions from.

• output_type

  (Literal['dict', 'DataFrame'], default: 'dict' ) –

  Output type of the data. Can be one of ["dict", "DataFrame"] By default "dict"

Returns:

• dict[str, dict[str, dict[str, Any]]] –

  In case output_type == "dict" it will return a dict with the following format: {model_name: {eventName: {alarm_attribute: value, ...}, ...}, ...}

• DataFrame –

  In case output_type == "DataFrame" it will return a DataFrame with the following format: index = MultiIndex with levels ["object_model", "eventName"], columns = [attribute, ...]

Source code in echo_baze/alarm_definitions.py

@validate_call
def get(
    self,
    object_models: list[str],
    output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, dict[str, Any]]] | DataFrame:
    """Gets all alarm definitions for the given object models.

    The most useful keys/columns returned are:

    - id
    - eventName
    - eventCode.code
    - eventCode.vendorEventType
    - brakePrg
    - canTriggerAllocation
    - description
    - isOkAlarm
    - eventKey
    - resetPrg
    - type

    Parameters
    ----------
    object_models : list[str]
        Desired object models to get the alarm definitions from.
    output_type : Literal["dict", "DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame"]
        By default "dict"

    Returns
    -------
    dict[str, dict[str, dict[str, Any]]]
        In case output_type == "dict" it will return a dict with the following format: {model_name: {eventName: {alarm_attribute: value, ...}, ...}, ...}
    DataFrame
        In case output_type == "DataFrame" it will return a DataFrame with the following format: index = MultiIndex with levels ["object_model", "eventName"], columns = [attribute, ...]

    """
    # getting object models
    object_model_ids = self.baze.objects.models.get_ids()

    if missing_object_models := set(object_models) - set(object_model_ids):
        logger.warning(f"Object models {missing_object_models} do not exist")

    # getting the alarm definitions
    results = {}
    for object_model in object_models:
        # creating an empty list for the results
        results[object_model] = []

        # getting alarm templates
        endpoint = f"configuration/domain/eventtemplates?domainIds={object_model_ids[object_model]}&includeParent=true"

        # getting the data
        request_result = self.baze.conn.get(endpoint)
        self._handle_http_errors(request_result)

        # converting to dict
        result: dict[str, list[dict[str, Any]]] = request_result.json()["data"]

        # now getting alarms for each alarm template
        # iterating models
        for model_alarm_templates in result.values():
            # iterating the templates itself
            for alarm_template in model_alarm_templates:
                # getting alarm template id
                alarm_template_id = alarm_template["eventTemplateId"]

                # getting the alarm definitions for the template
                endpoint = f"configuration/eventtemplates/{alarm_template_id}"

                # getting the data
                request_result = self.baze.conn.get(endpoint)
                self._handle_http_errors(request_result)

                # converting to list of dicts
                result: list[dict[str, Any]] = request_result.json()["events"]

                # appending to the results
                results[object_model] += result

    # returning on the desired format
    match output_type:
        case "dict":
            # making it a list of dicts
            return {object_model: {entry["eventName"]: entry for entry in entries} for object_model, entries in results.items()}
        case "DataFrame":
            # converting result to a list of dicts
            results = [{**entry, "object_model": object_model} for object_model, entries in results.items() for entry in entries]
            alarms_df = json_normalize(results, max_level=1).set_index(["object_model", "eventName"])
            return alarms_df.convert_dtypes(dtype_backend="pyarrow")
        case _:
            raise ValueError(f"Invalid output_type: {output_type}")