Point Details

PointDetails(baze)

Class used for handling point details

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(points, output_type='dict')

Gets the details of object points. Very useful to get the tag names and ids for the points.

The main attributes returned are:

• measurementId: tag id
• measurementName: tag name
• attributes: tag attributes

Parameters:

• points

  (dict[str, list[str]]) –

  Dict in the format {object_name: [point_name, ...], ...}

• 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 format {object_name: {point_name: {attribute_name: value, ...}, ...}, ...}

• DataFrame –

  In case output_type == "DataFrame" it will return a DataFrame with object_name and point as the index (MultiIndex).

Source code in echo_baze/point_details.py

@validate_call
def get(
    self,
    points: dict[str, list[str]],
    output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, dict[str, Any]]] | DataFrame:
    """Gets the details of object points. Very useful to get the tag names and ids for the points.

    The main attributes returned are:

    - measurementId: tag id
    - measurementName: tag name
    - attributes: tag attributes

    Parameters
    ----------
    points : dict[str, list[str]]
        Dict in the format {object_name: [point_name, ...], ...}
    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 format {object_name: {point_name: {attribute_name: value, ...}, ...}, ...}
    DataFrame
        In case output_type == "DataFrame" it will return a DataFrame with object_name and point as the index (MultiIndex).
    """
    #  this will only be used to get the id of the turbines
    _, object_names, _, _ = self.baze.points._get_points_info(points)  # noqa
    endpoint = "/objects/points/detail"

    results = {}
    for object_name, point_names in points.items():
        payload = {"objectIds": object_name, "points": point_names}

        result = self.baze.conn.get(endpoint, json=payload)
        self._handle_http_errors(result)

        result = result.json()
        object_ids = list(result["data"].keys())

        schemas = {object_names[object_id]: result["data"][object_id]["schemas"] for object_id in object_ids}
        result = {key: {} for key in schemas}

        for schema, data in schemas.items():
            for point in data:
                point_name = data[point]["measurementName"]
                point_name_new = point_name.replace(schema + "-", "")
                data[point]["attributes"] = data[point].pop("schema")
                result[schema].update({point_name_new: data[point]})
        results |= result

    # handling output
    match output_type:
        case "dict":
            return results
        case "DataFrame":
            # converting to df with MultiIndex index with levels object_name, point and rest as columns
            dfs = []
            for object_name, point_names in results.items():
                for point_name, point in point_names.items():
                    df = json_normalize(point)
                    df.index = MultiIndex.from_tuples([(object_name, point_name)], names=["object_name", "point"])
                    dfs.append(df)
            df = concat(dfs)

            return df