Feature Values - When At Condition¶
FeatureValuesWhenAtCondition(perfdb)
¶
Class used for getting timestamps when the feature meets a desired condition. Can be accessed via perfdb.features.values.whenatcondition.
Parameters:
Source code in echo_postgres/perfdb_root.py
def __init__(self, perfdb: e_pg.PerfDB) -> None:
"""Base class that all subclasses should inherit from.
Parameters
----------
perfdb : PerfDB
Top level object carrying all functionality and the connection handler.
"""
self._perfdb: e_pg.PerfDB = perfdb
get(features)
¶
Gets the timestamps when the feature meets a desired condition.
Parameters:
-
(features¶dict[str, dict[str, dict[str, str]]]) –Dictionary in the format (values are examples):
{ object_name: { feature_name: { "condition": "value > 1000 AND timestamp > '2021-01-01'", "order_by": "timestamp DESC", "limit": 1, }, ... }, ... }The keys needed for each feature are:
- condition: The condition to be met. Any valid SQL condition can be used. You must always do the comparisons against "value" and "timestamp" columns.
- order_by: The order by clause to be used. Any valid SQL order by clause can be used. You must always order by "value" or "timestamp" columns.
- limit: The number of timestamps to return. A minimum of 1 is required.
Returns:
-
dict[str, dict[str, list[datetime] | None]]–Dict containing the timestamps when the condition is met. The number of timestamps returned will be dictated by the limit key in the arguments. It is in the following format:
{ object_name: { feature_name: [timestamp1, timestamp2, ...], ... }, ... }If the condition is not met, the value is None.
Source code in echo_postgres/feature_values_whenatcondition.py
@validate_call
def get(
self,
features: dict[str, dict[str, dict[str, str | int]]],
) -> dict[str, dict[str, list[datetime] | None]]:
"""Gets the timestamps when the feature meets a desired condition.
Parameters
----------
features : dict[str, dict[str, dict[str, str]]]
Dictionary in the format (values are examples):
```python
{
object_name: {
feature_name: {
"condition": "value > 1000 AND timestamp > '2021-01-01'",
"order_by": "timestamp DESC",
"limit": 1,
},
...
},
...
}
```
The keys needed for each feature are:
- **condition**: The condition to be met. Any valid SQL condition can be used. You must always do the comparisons against "value" and "timestamp" columns.
- **order_by**: The order by clause to be used. Any valid SQL order by clause can be used. You must always order by "value" or "timestamp" columns.
- **limit**: The number of timestamps to return. A minimum of 1 is required.
Returns
-------
dict[str, dict[str, list[datetime] | None]]
Dict containing the timestamps when the condition is met. The number of timestamps returned will be dictated by the limit key in the arguments. It is in the following format:
```python
{
object_name: {
feature_name: [timestamp1, timestamp2, ...],
...
},
...
}
```
If the condition is not met, the value is None.
"""
# defining json schema to check the input
json_schema = {
"type": "object",
"additionalProperties": {
"type": "object",
"additionalProperties": {
"type": "object",
"properties": {
"condition": {"type": "string"},
"order_by": {"type": "string"},
"limit": {"type": "integer", "minimum": 1},
},
"required": ["condition", "order_by", "limit"],
"additionalProperties": False,
},
},
}
# checking arguments
try:
validate(instance=features, schema=json_schema)
except Exception as e:
raise ValueError("Invalid format for features argument.") from e
# getting the models of all the objects
objs = self._perfdb.objects.instances.get(object_names=list(features.keys()), output_type="DataFrame")
# validating if all objects were found
if objs.shape[0] != len(features):
missing_objs = set(features.keys()) - set(objs["object_name"].tolist())
raise ValueError(f"The following objects were not found in the database: {missing_objs}")
obj_models = sorted(objs["object_model_name"].unique().tolist())
# getting the features for the desired models
features_def = self._perfdb.features.definitions.get(object_models=obj_models, output_type="DataFrame")
# TODO refactor to avoid the loops and reduce everything to a single query
# dict to store the results
results = {}
# iterating over the objects
for object_name, features_dict in features.items():
# dict to store the results for the object
results[object_name] = {}
# getting the object id
obj_id = objs.loc[object_name, "id"]
# getting the model of the object
obj_model = objs.loc[object_name, "object_model_name"]
# getting the features available for the model
model_features = features_def.loc[pd.IndexSlice[obj_model, :], :]
# getting the features available for the object
# iterating over the features
for feature_name, feature_dict in features_dict.items():
# checking if the feature exists for the object
if feature_name not in model_features.index.get_level_values(1):
raise ValueError(f"Feature {feature_name} not found for object {object_name}.")
# getting the feature id
feature_id = model_features.loc[(obj_model, feature_name), "id"]
# making sure "value" or "timestamp" are in the condition and order_by
for field in ["order_by", "condition"]:
if not re.search(r"\bvalue\b", feature_dict[field]) and not re.search(r"\btimestamp\b", feature_dict[field]):
raise ValueError(f'{field} for feature {feature_name} in object {object_name} must contain "value" or "timestamp".')
# getting the query to get the timestamps
query = sql.SQL(
"""
SELECT timestamp::TIMESTAMP FROM performance.feature_values
WHERE object_id = {object_id}
AND feature_id = {feature_id}
AND {condition}
ORDER BY {order_by}
LIMIT {limit}
""",
).format(
object_id=sql.Literal(obj_id),
feature_id=sql.Literal(feature_id),
condition=sql.SQL(feature_dict["condition"]),
order_by=sql.SQL(feature_dict["order_by"]),
limit=sql.Literal(feature_dict["limit"]),
)
# getting the timestamps
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_polars(query=query)
# storing the timestamps
if df.is_empty():
results[object_name][feature_name] = None
else:
results[object_name][feature_name] = df["timestamp"].to_list()
return results