jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-20 11:47:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reduce complexity of _sorted_statistics_to_dict (#123936)

Browse Source 
* comp stats queries

* tweak

* reduce
This commit is contained in:
J. Nick Koston 2024-08-26 23:22:10 -10:00 committed by GitHub
parent 4bc19876ca
commit 0d2f22838a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 20 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
homeassistant/components/recorder/statistics.py
View File

@ -2088,71 +2088,38 @@ def _build_stats(
db_rows: list[Row],
table_duration_seconds: float,
start_ts_idx: int,
mean_idx: int | None,
min_idx: int | None,
max_idx: int | None,
last_reset_ts_idx: int | None,
state_idx: int | None,
sum_idx: int | None,
row_mapping: tuple[tuple[str, int], ...],
) -> list[StatisticsRow]:
"""Build a list of statistics without unit conversion."""
result: list[StatisticsRow] = []
ent_results_append = result.append
for db_row in db_rows:
row: StatisticsRow = {
return [
{
"start": (start_ts := db_row[start_ts_idx]),
"end": start_ts + table_duration_seconds,
**{key: db_row[idx] for key, idx in row_mapping}, # type: ignore[typeddict-item]
}
if last_reset_ts_idx is not None:
row["last_reset"] = db_row[last_reset_ts_idx]
if mean_idx is not None:
row["mean"] = db_row[mean_idx]
if min_idx is not None:
row["min"] = db_row[min_idx]
if max_idx is not None:
row["max"] = db_row[max_idx]
if state_idx is not None:
row["state"] = db_row[state_idx]
if sum_idx is not None:
row["sum"] = db_row[sum_idx]
ent_results_append(row)
return result
for db_row in db_rows
]
def _build_converted_stats(
db_rows: list[Row],
table_duration_seconds: float,
start_ts_idx: int,
mean_idx: int | None,
min_idx: int | None,
max_idx: int | None,
last_reset_ts_idx: int | None,
state_idx: int | None,
sum_idx: int | None,
row_mapping: tuple[tuple[str, int], ...],
convert: Callable[[float | None], float | None] | Callable[[float], float],
) -> list[StatisticsRow]:
"""Build a list of statistics with unit conversion."""
result: list[StatisticsRow] = []
ent_results_append = result.append
for db_row in db_rows:
row: StatisticsRow = {
return [
{
"start": (start_ts := db_row[start_ts_idx]),
"end": start_ts + table_duration_seconds,
**{
key: None if (v := db_row[idx]) is None else convert(v) # type: ignore[typeddict-item]
for key, idx in row_mapping
},
}
if last_reset_ts_idx is not None:
row["last_reset"] = db_row[last_reset_ts_idx]
if mean_idx is not None:
row["mean"] = None if (v := db_row[mean_idx]) is None else convert(v)
if min_idx is not None:
row["min"] = None if (v := db_row[min_idx]) is None else convert(v)
if max_idx is not None:
row["max"] = None if (v := db_row[max_idx]) is None else convert(v)
if state_idx is not None:
row["state"] = None if (v := db_row[state_idx]) is None else convert(v)
if sum_idx is not None:
row["sum"] = None if (v := db_row[sum_idx]) is None else convert(v)
ent_results_append(row)
return result
for db_row in db_rows
]
def _sorted_statistics_to_dict(
@ -2192,14 +2159,11 @@ def _sorted_statistics_to_dict(
# Figure out which fields we need to extract from the SQL result
# and which indices they have in the result so we can avoid the overhead
# of doing a dict lookup for each row
mean_idx = field_map["mean"] if "mean" in types else None
min_idx = field_map["min"] if "min" in types else None
max_idx = field_map["max"] if "max" in types else None
last_reset_ts_idx = field_map["last_reset_ts"] if "last_reset" in types else None
state_idx = field_map["state"] if "state" in types else None
if "last_reset_ts" in field_map:
field_map["last_reset"] = field_map.pop("last_reset_ts")
sum_idx = field_map["sum"] if "sum" in types else None
sum_only = len(types) == 1 and sum_idx is not None
row_idxes = (mean_idx, min_idx, max_idx, last_reset_ts_idx, state_idx, sum_idx)
row_mapping = tuple((key, field_map[key]) for key in types if key in field_map)
# Append all statistic entries, and optionally do unit conversion
table_duration_seconds = table.duration.total_seconds()
for meta_id, db_rows in stats_by_meta_id.items():
@ -2228,9 +2192,9 @@ def _sorted_statistics_to_dict(
else:
_stats = _build_sum_stats(*build_args, sum_idx)
elif convert:
_stats = _build_converted_stats(*build_args, *row_idxes, convert)
_stats = _build_converted_stats(*build_args, row_mapping, convert)
else:
_stats = _build_stats(*build_args, *row_idxes)
_stats = _build_stats(*build_args, row_mapping)
result[statistic_id] = _stats