/
recorder_import_statistics.py
55 lines (48 loc) 路 2 KB
/
recorder_import_statistics.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
"""Spook - Not your homie."""
from __future__ import annotations
import voluptuous as vol
from homeassistant.components.recorder import DOMAIN
from homeassistant.components.recorder.statistics import (
async_add_external_statistics,
async_import_statistics,
)
from homeassistant.core import ServiceCall, valid_entity_id
from homeassistant.helpers import config_validation as cv
from . import AbstractSpookAdminService
class SpookService(AbstractSpookAdminService):
"""Recorder integration service to import statistics."""
domain = DOMAIN
service = "import_statistics"
schema = {
vol.Required("has_mean"): bool,
vol.Required("has_sum"): bool,
vol.Optional("name", default=None): str,
vol.Required("source"): str,
vol.Required("statistic_id"): str,
vol.Optional("unit_of_measurement", default=None): str,
vol.Required("stats"): [
{
vol.Required("start"): cv.datetime,
vol.Optional("mean"): vol.Any(float, int),
vol.Optional("min"): vol.Any(float, int),
vol.Optional("max"): vol.Any(float, int),
vol.Optional("last_reset", default=None): cv.datetime,
vol.Optional("state"): vol.Any(float, int),
vol.Optional("sum"): vol.Any(float, int),
},
],
}
async def async_handle_service(self, call: ServiceCall) -> None:
"""Handle the service call."""
metadata = {
"has_mean": call.data["has_mean"],
"has_sum": call.data["has_sum"],
"name": call.data["name"],
"source": call.data["source"],
"statistic_id": call.data["statistic_id"],
"unit_of_measurement": call.data["unit_of_measurement"],
}
if valid_entity_id(call.data["statistic_id"]):
async_import_statistics(self.hass, metadata, call.data["stats"])
else:
async_add_external_statistics(self.hass, metadata, call.data["stats"])