Point Values - Latest¶
PointValuesLatest(baze)
¶
Class used for handling the latest value 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(points, time_zone='local', output_type='DataFrame')
¶
Gets the last value of the given points.
Parameters:
-
(points¶dict[str, list[str]]) –Dict in the format {object_name: [point_name, ...], ...}
-
(time_zone¶TimeZone, default:'local') –Used to define in which time zone the output is. There are three options:
- If "UTC" is used, we assume time already is in UTC.
- If local is used, the default time zone defined in echo_baze will be used.
- If an int, must be between -12 and +12
By default "local"
-
(output_type¶Literal['dict', 'DataFrame'], default:'DataFrame') –Output type of the data. Can be one of ["dict", "DataFrame"] By default "DataFrame"
Returns:
-
dict[str, dict[str, dict[str, Any]]]–In case output_type == "dict" it will return a dict with the following format: {object_name: {point_name: {t: time, v: value, q: quality}, ...}, ...}
-
DataFrame–In case output_type == "DataFrame" it will return a DataFrame with the following format: index = MultiIndex with levels ["object_name", "point"], columns = [time, value, quality]
Source code in echo_baze/point_values_latest.py
@validate_call
def get(
self,
points: dict[str, list[str]],
time_zone: TimeZone = "local",
output_type: Literal["dict", "DataFrame"] = "DataFrame",
) -> dict[str, dict[str, dict[str, Any]]] | DataFrame:
"""Gets the last value of the given points.
Parameters
----------
points : dict[str, list[str]]
Dict in the format {object_name: [point_name, ...], ...}
time_zone : TimeZone, optional
Used to define in which time zone the output is. There are three options:
- If "UTC" is used, we assume time already is in UTC.
- If local is used, the default time zone defined in echo_baze will be used.
- If an int, must be between -12 and +12
By default "local"
output_type : Literal["dict", "DataFrame"], optional
Output type of the data. Can be one of ["dict", "DataFrame"]
By default "DataFrame"
Returns
-------
dict[str, dict[str, dict[str, Any]]]
In case output_type == "dict" it will return a dict with the following format: {object_name: {point_name: {t: time, v: value, q: quality}, ...}, ...}
DataFrame
In case output_type == "DataFrame" it will return a DataFrame with the following format: index = MultiIndex with levels ["object_name", "point"], columns = [time, value, quality]
"""
# getting the models of each object
object_names = list(points.keys())
objects_def = self.baze.objects.instances.get(object_names=object_names)
object_models = {
object_name: object_attrs["attributes"]["domainName"]
for object_name, object_attrs in objects_def.items()
if object_name in points
}
if len(object_models) != len(points):
raise ValueError(
f"The following objects were not found: {set(points) - set(object_models)}",
)
# dict that maps object names to object ids
object_name_to_id = {object_name: object_attrs["objectId"] for object_name, object_attrs in objects_def.items()}
# dict that maps object ids to object names
object_id_to_name = {v: k for k, v in object_name_to_id.items()}
# getting the points for each model
object_model_points = self.baze.points.definitions.get_ids(
object_models=list(set(object_models.values())),
)
results = {}
# getting the values for each object
for object_name, object_points in points.items():
# checking if all points are available for the model
not_found_points = set(object_points) - set(
object_model_points[object_models[object_name]],
)
if len(not_found_points) > 0:
raise ValueError(
f"The following points were not found for object '{object_name}': {not_found_points}",
)
# getting the data
endpoint = "objects/points"
payload = {
"ObjectIds": [object_name_to_id[object_name]],
"Points": object_points,
}
# getting the data
result = self.baze.conn.get(endpoint, json=payload)
self._handle_http_errors(result)
# converting to dict
result: dict[str, dict[str, dict[str, dict[str, Any]]]] = result.json()["result"]
# converting to a better format
result = {
object_id_to_name[object_id]: {
point_name: {
"t": datetime_from_timestamp(timestamp=value["t"], time_zone=time_zone),
"v": value["v"],
"q": cast_quality(value["q"], return_type="int"),
}
for point_name, value in point_values["points"].items()
if "v" in value and value["t"] > 0
}
for object_id, point_values in result.items()
}
# adding to results
results |= result
# returning on the desired format
match output_type:
case "dict":
return results
case "DataFrame":
# types for the DataFrame
df_dtypes = {
"object_name": "string[pyarrow]",
"point": "string[pyarrow]",
"value": "double[pyarrow]",
"quality": "int32[pyarrow]",
"time": "datetime64[s]",
}
if len(results) == 0:
# if no results, return an empty DataFrame
df = DataFrame(
columns=["time", "value", "quality", "object_name", "point"],
)
# making sure timestamp columns are rounded to second level
df["time"] = df["time"].dt.round("s")
# casting to correct types
df = df.astype(df_dtypes)
# setting the index
df = df.set_index(["object_name", "point"])
return df
# converting to a DataFrame with index containing level 0 = object_name and level 1 = point and columns = [time, value, quality]
df = json_normalize(results, max_level=1).T
df.columns = ["dict"]
# splitting dict in three columns (t, v and q)
df = df["dict"].apply(Series)
# splitting index in two columns (object_name and point)
df = df.reset_index(drop=False)
df.loc[:, ["object_name", "point"]] = df["index"].str.split(".", expand=True).values
# dropping the old index
df = df.drop(columns=["index"])
# renaming
df = df.rename(columns={"t": "time", "q": "quality", "v": "value"})
# casting to correct types
df = df.astype(df_dtypes)
# setting the index
df = df.set_index(["object_name", "point"])
# reorder columns
df = df[["time", "value", "quality"]]
return df
case _:
raise ValueError(
f"output_type must be one of ['dict', 'DataFrame'], got '{output_type}'",
)