Conincidence or not, just today I wanted to have something similar for my water usage.
I created a not-very-nice (not a dev myself but a pretty OK hacker, my own views
) ... pyscscript for that.
If it helps, I can also have a look if you get stuck and see if I can modify it to your specific situation. The entity can be chosen and it will add the averages as attributes to it.
Note: this is only a few hours old and I have not yet the time to see how to embed this in my cards/graphs
# <config>/pyscript/water_stats.py
from datetime import datetime, timedelta, timezone
@service(supports_response="only")
async def get_statistics_water(
days: int = 180,
entity_id: str = None,
):
"""yaml
description: Get water usage averages over a configurable number of days.
fields:
days:
description: Number of days to look back
required: false
default: 180
selector:
number:
min: 1
max: 365
mode: box
entity_id:
description: The entity ID of the total water sensor
required: false
default: "sensor.water_total_increasing"
selector:
entity:
domain: sensor
"""
target_eid = entity_id
statistic_ids = [target_eid]
days = int(days)
def fmt(dt):
return dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
now = datetime.now(timezone.utc)
start = now - timedelta(days=days)
# --- Daily average ---
resp = await service.call("recorder", "get_statistics",
statistic_ids=statistic_ids,
start_time=fmt(start),
end_time=fmt(now),
period="day",
types=["change"],
return_response=True)
daily_raw = resp["statistics"].get(target_eid, [])
daily_values = [round(e["change"], 3) for e in daily_raw if e.get("change") is not None and e["change"] >= 0]
avg_daily = round(sum(daily_values) / len(daily_values), 3) if daily_values else 0
# --- Weekly average ---
resp = await service.call("recorder", "get_statistics",
statistic_ids=statistic_ids,
start_time=fmt(start),
end_time=fmt(now),
period="week",
types=["change"],
return_response=True)
weekly_raw = resp["statistics"].get(target_eid, [])
weekly_values = [round(e["change"], 3) for e in weekly_raw if e.get("change") is not None and e["change"] >= 0]
avg_weekly = round(sum(weekly_values) / len(weekly_values), 3) if weekly_values else 0
# --- Monthly average ---
resp = await service.call("recorder", "get_statistics",
statistic_ids=statistic_ids,
start_time=fmt(start),
end_time=fmt(now),
period="month",
types=["change"],
return_response=True)
monthly_raw = resp["statistics"].get(target_eid, [])
monthly_values = [round(e["change"], 3) for e in monthly_raw if e.get("change") is not None and e["change"] >= 0]
avg_monthly = round(sum(monthly_values) / len(monthly_values), 3) if monthly_values else 0
log.info("Water stats (%s days): daily=%s, weekly=%s, monthly=%s m³", days, avg_daily, avg_weekly, avg_monthly)
state.setattr(f"{target_eid}.avg_daily_m3", avg_daily)
state.setattr(f"{target_eid}.avg_weekly_m3", avg_weekly)
state.setattr(f"{target_eid}.avg_monthly_m3", avg_monthly)
state.setattr(f"{target_eid}.stats_days", days)
return {
"avg_daily_m3": avg_daily,
"avg_weekly_m3": avg_weekly,
"avg_monthly_m3": avg_monthly,
"days": days,
}