Point Definitions

PointDefinitions(baze)

Class used for handling definition of points.

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 point definitions for the given object models.

The most useful keys/columns returned are:

• dataTemplateType
• deadband
• desc
• disabled
• schemaTemplate
• unit

Parameters:

• object_models

  (list[str]) –

  Desired object models to get the point 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: {point_name: {attribute: value, ...}, ...}, ...}

• DataFrame –

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

Source code in echo_baze/point_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 point definitions for the given object models.

    The most useful keys/columns returned are:

    - dataTemplateType
    - deadband
    - desc
    - disabled
    - schemaTemplate
    - unit

    Parameters
    ----------
    object_models : list[str]
        Desired object models to get the point 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: {point_name: {attribute: value, ...}, ...}, ...}
    DataFrame
        In case output_type == "DataFrame" it will return a DataFrame with the following format: index = MultiIndex with levels ["object_model", "point"], columns = [attribute, ...]
    """
    # getting the point templates
    point_templates = self.baze.points.templates.get_ids(object_models=object_models)

    # getting the points
    results = {}
    for object_model in object_models:
        results[object_model] = {}
        for point_template_id in point_templates[object_model].values():
            endpoint = f"configuration/domain/schematemplates/{point_template_id}"

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

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

            # converting to desired format
            result = {entry["schema"]: entry for entry in result["schemas"]}

            # adding to the results
            results[object_model] = results[object_model] | result

    # returning on the desired format
    match output_type:
        case "dict":
            return results
        case "DataFrame":
            df = DataFrame.from_dict(
                {(model, point): attributes for model, points in results.items() for point, attributes in points.items()},
                orient="index",
            )
            if df.empty:
                df.index = MultiIndex.from_product([[], []])
            df.index.names = ["object_model", "point"]
            df = df.reset_index(drop=False)
            # adding columns if they dont exist
            for column in ["object_model", "point", "dataTemplateType", "deadband", "desc", "disabled", "schemaTemplate", "unit"]:
                if column not in df.columns:
                    df[column] = NA
            df = df.astype({"object_model": "string[pyarrow]", "point": "string[pyarrow]"})
            df = df.convert_dtypes(dtype_backend="pyarrow").set_index(["object_model", "point"])
            return df
        case _:
            raise ValueError(f"Invalid output_type: {output_type}")

get_ids(object_models)

Gets a dictionary with all point ids.

Parameters:

• object_models

  (list[str]) –

  List of object models to get the points from.

Returns:

• dict[str, dict[str, int]]: –

  Dictionary with all point in the format {object_model: {name: id, ...}, ...}

Source code in echo_baze/point_definitions.py

@validate_call
def get_ids(self, object_models: list[str]) -> dict[str, dict[str, int]]:
    """
    Gets a dictionary with all point ids.

    Parameters
    ----------
    object_models : list[str]
        List of object models to get the points from.

    Returns
    -------
    dict[str, dict[str, int]]:
        Dictionary with all point in the format {object_model: {name: id, ...}, ...}
    """
    # getting ids of the object models
    object_model_ids = self.baze.objects.models.get_ids()
    object_model_ids = {k: v for k, v in object_model_ids.items() if k in object_models}
    if len(object_model_ids) != len(object_models):
        wrong_models = set(object_models) - set(object_model_ids.keys())
        raise ValueError(f"Invalid object models: {wrong_models}")

    # getting the points
    results = {}
    for object_model, object_model_id in object_model_ids.items():
        endpoint = f"configuration/domain/{object_model_id}/schemas?includeParent=true"

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

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

        # converting to desired format
        result = {entry["schema"]: entry["schemaId"] for entry in result}

        # adding to the results
        results[object_model] = result

    return results