Energy consumption & Usage from history_stats

[davidrapan]

Hi @sparky8512, first of all great job and thanks for your work!

I have recently added power consumption (PowerDict) to the Home Assistant and latest_power works great but we noticed that

history_stats(parse_samples=-1, context=ChannelContext(target="..."))[-1]["total_energy"]

• w/ parse_samples=-1 which (if I understood correctly) should compute stats above all available data

seems odd and doesn't really "add up" and increase w/ time.

And it seems it's the same for UsageDict too (does not increase w/ time since reboot).

I did take your word for it that it should compute energy since reboot, or am I just misunderstanding this feature completely?

Cheers!

[sparky8512]

Well... you're not completely misunderstanding it.

Perhaps I should not have called that item total_energy, as it evokes similarly named fields in other devices which report cumulative energy since reboot. You are correct that calling history_stats in the way you posted will compute energy across all available data. The problem is that "all available data" is limited to the past 15 minutes of samples, because that's all the dish keeps around.

If you want to compute cumulative energy since some point of time, you'll need to call history_stats periodically, keeping track of the end_counter returned from each call (in the first dict in the returned tuple) in order to pass that into the next call's start parameter, and then add up the total_energy values returned each time. I recommend a 1 minute period, but you can use anything from 1 second to a little less than 15 minutes without losing data.

One more thing. While this:

history_stats(parse_samples=-1, context=ChannelContext(target="..."))[-1]["total_energy"]
                                                                     ^^^^

will currently work to pull the power stats dict out of the return value, I would not recommend that exact usage, as it will break if there is ever another dict added to the history_stats return value. It would be better to use [6] in its place. This is somewhat moot, though, since if you want to track cumulative energy over time, you'll need the first dict, too, so that exact pattern won't work.

[davidrapan]

Ah I see! I should have read through the code more carefully!

Could be then used periodical call of get_history + concatenate_history and then history_stats(..., history=concatenated_history)? As I'd like to find a solution w/ least boilerplate possible (+ it also involves number of download and upload bytes too).

It would be better to use [6] in its place.

Sure! Will do. 😉

[sparky8512]

Could be then used periodical call of get_history + concatenate_history and then history_stats(..., history=concatenated_history)? As I'd like to find a solution w/ least boilerplate possible (+ it also involves number of download and upload bytes too).

Yes, that will work, too, it will just use a bit more memory, as you'd be keeping around an ever-growing history buffer, and be a bit less performant, as the code in history_stats would be computing totals over that whole ever-growing buffer on each call, rather than just the new data. I doubt either would be all that significant, though, unless you kept it running for months on end.

Note that the same approach of tracking the end counter and keeping cumulate totals could be applied to the upload and download usage data, too, but I suppose just keeping a concatenated history buffer is a somewhat simpler.

[davidrapan]

it will just use a bit more memory, as you'd be keeping around an ever-growing history buffer, and be a bit less performant, as the code in history_stats would be computing totals over that whole ever-growing buffer on each call, rather than just the new data

Yeah, that's what I thought too... maybe caching of history_stats(...)[0] and cumulation of history_stats(..., start=result[0]["end_counter"])[6]["total_energy"] (same for usage) won't be that much extra code...

I'll compare both approaches and will see, thanks for your insight. 👍

[davidrapan]

So for anyone wondering I did chose this method (it's not full & working code):

self.history_stats_start = None

def _get_starlink_cumulative_stats(self) -> tuple[UsageDict, PowerDict]:
    """Retrieve Starlink cumulative stats."""
    context = self.channel_context
    index, _, _, _, _, usage, consumption, *_ = history_stats(
        parse_samples=-1, start=self.history_stats_start, context=context
    )
    self.history_stats_start = index["end_counter"]
    if self.data:
        if index["samples"] > 0:
            usage["download_usage"] += self.data.usage["download_usage"]
            usage["upload_usage"] += self.data.usage["upload_usage"]
            consumption["total_energy"] += self.data.consumption["total_energy"]
        else:
            usage = self.data.usage
            consumption = self.data.consumption
    return usage, consumption