Skip to content

Reference

Top-level module for importing the Ecos class.

AsyncEcos(email=None, password=None, country=None, datacenter=None, url=None, access_token=None, refresh_token=None)

Asynchronous ECOS API client class.

This class provides methods for interacting with the ECOS API, including authentication, retrieving user information, and managing homes. It uses the aiohttp library to make asynchronous HTTP requests to the API.

Parameters:

Name Type Description Default
email str | None

The user's email to use for authentication.

 None
password str | None

The user's password to use for authentication.

 None
country str | None

Reserved for future.

 None
datacenter str | None

The location of the ECOS API datacenter. Can be one of CN, EU, or AU.

 None
url str | None

The URL of the ECOS API. If specified, datacenter is ignored.

 None
access_token str | None

The access token for authentication with the ECOS API.

 None
refresh_token str | None

The refresh token for authentication with the ECOS API.

 None

Raises:

Type Description
InitializationError

If datacenter is not one of CN, EU, or AU and url is not provided.
Source code in src/ecactus/base.py 
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
    self,
    email: str | None = None,
    password: str | None = None,
    country: str | None = None,
    datacenter: str | None = None,
    url: str | None = None,
    access_token: str | None = None,
    refresh_token: str | None = None,
) -> None:
    """Initialize a session with ECOS API.

    Args:
        email: The user's email to use for authentication.
        password: The user's password to use for authentication.
        country: _Reserved for future_.
        datacenter: The location of the ECOS API datacenter.
            Can be one of `CN`, `EU`, or `AU`.
        url: The URL of the ECOS API. If specified, `datacenter` is ignored.
        access_token: The access token for authentication with the ECOS API.
        refresh_token: The refresh token for authentication with the ECOS API.

    Raises:
        InitializationError: If `datacenter` is not one of `CN`, `EU`, or `AU` and `url` is not provided.

    """
    logger.info("Initializing session")
    self.email = email
    self.password = password
    self.country = country
    self.access_token = access_token
    self.refresh_token = refresh_token
    # TODO: get datacenters from https://dcdn-config.weiheng-tech.com/prod/config.json
    datacenters = {
        "CN": "https://api-ecos-hu.weiheng-tech.com",
        "EU": "https://api-ecos-eu.weiheng-tech.com",
        "AU": "https://api-ecos-au.weiheng-tech.com",
    }
    if url is None:
        if datacenter is None:
            raise InitializationError("url or datacenter not specified")
        if datacenter not in datacenters:
            raise InitializationError(
                "datacenter must be one of {}".format(", ".join(datacenters.keys()))
            )
        self.url = datacenters[datacenter]
    else:  # url specified, ignore datacenter
        self.url = url.rstrip("/")  # remove trailing / from url

login(email=None, password=None) async

Authenticate with the ECOS API using a provided email and password.

Parameters:

Name Type Description Default
email str | None

The user's email to use for authentication.

 None
password str | None

The user's password to use for authentication.

 None

Raises:

Type Description
AuthenticationError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def login(
    self, email: str | None = None, password: str | None = None
) -> None:
    """Authenticate with the ECOS API using a provided email and password.

    Args:
        email: The user's email to use for authentication.
        password: The user's password to use for authentication.

    Raises:
        AuthenticationError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Login")
    if email is not None:
        self.email = email
    if password is not None:
        self.password = password
    payload = {
        "_t": int(time.time()),
        "clientType": "BROWSER",
        "clientVersion": "1.0",
        "email": self.email,
        "password": self.password,
    }
    try:
        data = await self._async_post("/api/client/guide/login", payload=payload)
    except ApiResponseError as err:
        if err.code == 20414:
            raise AuthenticationError from err
        if err.code == 20000:
            raise AuthenticationError("Missing Account or Password") from err
        raise
    self.access_token = data["accessToken"]
    self.refresh_token = data["refreshToken"]

get_user() async

Get user details.

Returns:

Type Description
User

A User object.

Raises:

Type Description
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
async def get_user(self) -> User:
    """Get user details.

    Returns:
        A User object.

    Raises:
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get user")
    await self._ensure_login()
    return User(**await self._async_get("/api/client/settings/user/info"))

get_homes() async

Get a list of homes.

Returns:

Type Description
list[Home]

A list of Home objects.

Raises:

Type Description
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def get_homes(self) -> list[Home]:
    """Get a list of homes.

    Returns:
        A list of Home objects.

    Raises:
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get home list")
    await self._ensure_login()
    return [
        Home(**home_data)
        for home_data in await self._async_get("/api/client/v2/home/family/query")
    ]

get_devices(home_id) async

Get a list of devices for a home.

Parameters:

Name Type Description Default
home_id str

The home ID to get devices for.

 required

Returns:

Type Description
list[Device]

A list of Device objects.

Raises:

Type Description
HomeDoesNotExistError

If the home id is not correct.
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def get_devices(self, home_id: str) -> list[Device]:
    """Get a list of devices for a home.

    Args:
        home_id: The home ID to get devices for.

    Returns:
        A list of Device objects.

    Raises:
        HomeDoesNotExistError: If the home id is not correct.
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get devices for home %s", home_id)
    await self._ensure_login()
    try:
        return [
            Device(**device_data)
            for device_data in await self._async_get(
                "/api/client/v2/home/device/query", payload={"homeId": home_id}
            )
        ]
    except ApiResponseError as err:
        if err.code == 20450:
            raise HomeDoesNotExistError(home_id) from err
        raise

get_all_devices() async

Get a list of all the devices.

Returns:

Type Description
list[Device]

A list of Device objects.

Raises:

Type Description
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
async def get_all_devices(self) -> list[Device]:
    """Get a list of all the devices.

    Returns:
        A list of Device objects.

    Raises:
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get devices for every homes")
    await self._ensure_login()
    return [
        Device(**device_data)
        for device_data in await self._async_get("/api/client/home/device/list")
    ]

get_today_device_data(device_id) async

Get power metrics of the current day until now.

Parameters:

Name Type Description Default
device_id str

The device ID to get power metrics for.

 required

Returns:

Type Description
PowerTimeSeries

Metrics of the current day until now.

Raises:

Type Description
UnauthorizedDeviceError

If the device is not authorized or unknown.
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
async def get_today_device_data(self, device_id: str) -> PowerTimeSeries:
    """Get power metrics of the current day until now.

    Args:
        device_id: The device ID to get power metrics for.

    Returns:
        Metrics of the current day until now.

    Raises:
        UnauthorizedDeviceError: If the device is not authorized or unknown.
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get current day data for device %s", device_id)
    await self._ensure_login()
    try:
        return PowerTimeSeries(**await self._async_post(
            "/api/client/home/now/device/realtime", payload={"deviceId": device_id}
        ))
    except ApiResponseError as err:
        if err.code == 20424:
            raise UnauthorizedDeviceError(device_id) from err
        raise

get_realtime_home_data(home_id) async

Get current power for the home.

Parameters:

Name Type Description Default
home_id str

The home ID to get current power for.

 required

Returns:

Type Description
PowerMetrics

Current metrics for the home.

Raises:

Type Description
HomeDoesNotExistError

If the home id is not correct.
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
async def get_realtime_home_data(self, home_id: str) -> PowerMetrics:
    """Get current power for the home.

    Args:
        home_id: The home ID to get current power for.

    Returns:
        Current metrics for the home.

    Raises:
        HomeDoesNotExistError: If the home id is not correct.
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get realtime data for home %s", home_id)
    try:
        return PowerMetrics(**await self._async_get(
            "/api/client/v2/home/device/runData", payload={"homeId": home_id}
        ))
    except ApiResponseError as err:
        if err.code == 20450:
            raise HomeDoesNotExistError(home_id) from err
        raise

get_realtime_device_data(device_id) async

Get current power for a device.

Parameters:

Name Type Description Default
device_id str

The device ID to get current power for.

 required

Returns:

Type Description
PowerMetrics

Current metrics for the device.

Raises:

Type Description
UnauthorizedDeviceError

If the device is not authorized or unknown.
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
async def get_realtime_device_data(self, device_id: str) -> PowerMetrics:
    """Get current power for a device.

    Args:
        device_id: The device ID to get current power for.

    Returns:
        Current metrics for the device.

    Raises:
        UnauthorizedDeviceError: If the device is not authorized or unknown.
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get realtime data for device %s", device_id)
    try:
        return PowerMetrics(**await self._async_post(
            "/api/client/home/now/device/runData", payload={"deviceId": device_id}
        ))
    except ApiResponseError as err:
        if err.code == 20424:
            raise UnauthorizedDeviceError(device_id) from err
        raise

get_history(device_id, period_type, start_date=None) async

Get aggregated energy for a period.

Parameters:

Name Type Description Default
device_id str

The device ID to get history for.

 required
period_type int

Possible value:

  • 0: daily values of the calendar month corresponding to start_date
  • 1: today daily values (start_date is ignored) (?)
  • 2: daily values of the current month (start_date is ignored)
  • 3: same than 2 ?
  • 4: total for the current month (start_date is ignored)
 required
start_date datetime | None

The start date.

 None

Returns:

Type Description
EnergyHistory

Data and metrics corresponding to the defined period.

Raises:

Type Description
UnauthorizedDeviceError

If the device is not authorized or unknown.
ParameterVerificationFailedError

If a parameter is not valid (period_type number for example)
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
async def get_history(
    self, device_id: str, period_type: int, start_date: datetime | None = None
) -> EnergyHistory:
    """Get aggregated energy for a period.

    Args:
        device_id: The device ID to get history for.
        period_type: Possible value:

            - `0`: daily values of the calendar month corresponding to `start_date`
            - `1`: today daily values (`start_date` is ignored) (?)
            - `2`: daily values of the current month (`start_date` is ignored)
            - `3`: same than 2 ?
            - `4`: total for the current month (`start_date` is ignored)
        start_date: The start date.

    Returns:
        Data and metrics corresponding to the defined period.

    Raises:
        UnauthorizedDeviceError: If the device is not authorized or unknown.
        ParameterVerificationFailedError: If a parameter is not valid (`period_type` number for example)
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get history for device %s", device_id)
    if start_date is None:
        if period_type in (1, 2, 4):
            start_ts = 0
        else:
            raise ParameterVerificationFailedError(f"start_date is required for period_type {period_type}")
    else:
        start_ts = int(start_date.timestamp()) if start_date is not None else 0
    try:
        return EnergyHistory(**await self._async_post(
            "/api/client/home/history/home",
            payload={
                "deviceId": device_id,
                "timestamp": start_ts,
                "periodType": period_type,
            },
        ))
    except ApiResponseError as err:
        if err.code == 20424:
            raise UnauthorizedDeviceError(device_id) from err
        if err.code == 20404:
            raise ParameterVerificationFailedError from err
        raise

get_insight(device_id, period_type, start_date=None) async

Get energy metrics and statistics of a device for a period.

Parameters:

Name Type Description Default
device_id str

The device ID to get data for.

 required
period_type int

Possible value:

  • 0: 5-minute power measurement for the calendar day corresponding to start_date (insightConsumptionDataDto is None)
  • 1: (not implemented)
  • 2: daily energy for the calendar month corresponding to start_date (deviceRealtimeDto is None)
  • 3: (not implemented)
  • 4: monthly energy for the calendar year corresponding to start_date (deviceRealtimeDto is None)
  • 5: yearly energy, start_date is ignored (?) (deviceRealtimeDto is None)
 required
start_date datetime | None

The start date.

 None

Returns:

Type Description
DeviceInsight

Statistics and metrics corresponding to the defined period.

Raises:

Type Description
UnauthorizedDeviceError

If the device is not authorized or unknown.
ParameterVerificationFailedError

If a parameter is not valid (period_type number for example)
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/async_client.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
async def get_insight(
    self, device_id: str, period_type: int, start_date: datetime | None = None
) -> DeviceInsight:
    """Get energy metrics and statistics of a device for a period.

    Args:
        device_id: The device ID to get data for.
        period_type: Possible value:

            - `0`: 5-minute power measurement for the calendar day corresponding to `start_date` (`insightConsumptionDataDto` is `None`)
            - `1`: (not implemented)
            - `2`: daily energy for the calendar month corresponding to `start_date` (`deviceRealtimeDto` is `None`)
            - `3`: (not implemented)
            - `4`: monthly energy for the calendar year corresponding to `start_date` (`deviceRealtimeDto` is `None`)
            - `5`: yearly energy, `start_date` is ignored (?) (`deviceRealtimeDto` is `None`)
        start_date: The start date.

    Returns:
        Statistics and metrics corresponding to the defined period.

    Raises:
        UnauthorizedDeviceError: If the device is not authorized or unknown.
        ParameterVerificationFailedError: If a parameter is not valid (`period_type` number for example)
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get insight for device %s", device_id)
    if start_date is None:
        if period_type == 5:
            start_ts = 0
        else:
            raise ParameterVerificationFailedError(f"start_date is required for period_type {period_type}")
    else:
        start_ts = int(start_date.timestamp() * 1000)  # timestamp in milliseconds
    try:
        return DeviceInsight(**await self._async_post(
            "/api/client/v2/device/three/device/insight",
            payload={
                "deviceId": device_id,
                "timestamp": start_ts,
                "periodType": period_type,
            },
        ))
    except ApiResponseError as err:
        if err.code == 20424:
            raise UnauthorizedDeviceError(device_id) from err
        if err.code == 20404:
            raise ParameterVerificationFailedError from err
        raise

Ecos(email=None, password=None, country=None, datacenter=None, url=None, access_token=None, refresh_token=None)

Synchronous ECOS API client class.

This class provides methods for interacting with the ECOS API, including authentication, retrieving user information, and managing homes. It uses the requests library to make HTTP requests to the API.

Parameters:

Name Type Description Default
email str | None

The user's email to use for authentication.

 None
password str | None

The user's password to use for authentication.

 None
country str | None

Reserved for future.

 None
datacenter str | None

The location of the ECOS API datacenter. Can be one of CN, EU, or AU.

 None
url str | None

The URL of the ECOS API. If specified, datacenter is ignored.

 None
access_token str | None

The access token for authentication with the ECOS API.

 None
refresh_token str | None

The refresh token for authentication with the ECOS API.

 None

Raises:

Type Description
InitializationError

If datacenter is not one of CN, EU, or AU and url is not provided.
Source code in src/ecactus/base.py 
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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
    self,
    email: str | None = None,
    password: str | None = None,
    country: str | None = None,
    datacenter: str | None = None,
    url: str | None = None,
    access_token: str | None = None,
    refresh_token: str | None = None,
) -> None:
    """Initialize a session with ECOS API.

    Args:
        email: The user's email to use for authentication.
        password: The user's password to use for authentication.
        country: _Reserved for future_.
        datacenter: The location of the ECOS API datacenter.
            Can be one of `CN`, `EU`, or `AU`.
        url: The URL of the ECOS API. If specified, `datacenter` is ignored.
        access_token: The access token for authentication with the ECOS API.
        refresh_token: The refresh token for authentication with the ECOS API.

    Raises:
        InitializationError: If `datacenter` is not one of `CN`, `EU`, or `AU` and `url` is not provided.

    """
    logger.info("Initializing session")
    self.email = email
    self.password = password
    self.country = country
    self.access_token = access_token
    self.refresh_token = refresh_token
    # TODO: get datacenters from https://dcdn-config.weiheng-tech.com/prod/config.json
    datacenters = {
        "CN": "https://api-ecos-hu.weiheng-tech.com",
        "EU": "https://api-ecos-eu.weiheng-tech.com",
        "AU": "https://api-ecos-au.weiheng-tech.com",
    }
    if url is None:
        if datacenter is None:
            raise InitializationError("url or datacenter not specified")
        if datacenter not in datacenters:
            raise InitializationError(
                "datacenter must be one of {}".format(", ".join(datacenters.keys()))
            )
        self.url = datacenters[datacenter]
    else:  # url specified, ignore datacenter
        self.url = url.rstrip("/")  # remove trailing / from url

login(email=None, password=None)

Authenticate with the ECOS API using a provided email and password.

Parameters:

Name Type Description Default
email str | None

The user's email to use for authentication.

 None
password str | None

The user's password to use for authentication.

 None

Raises:

Type Description
AuthenticationError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def login(
    self, email: str | None = None, password: str | None = None
) -> None:
    """Authenticate with the ECOS API using a provided email and password.

    Args:
        email: The user's email to use for authentication.
        password: The user's password to use for authentication.

    Raises:
        AuthenticationError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Login")
    if email is not None:
        self.email = email
    if password is not None:
        self.password = password
    payload = {
        "_t": int(time.time()),
        "clientType": "BROWSER",
        "clientVersion": "1.0",
        "email": self.email,
        "password": self.password,
    }
    try:
        data = self._post("/api/client/guide/login", payload=payload)
    except ApiResponseError as err:
        if err.code == 20414:
            raise AuthenticationError from err
        if err.code == 20000:
            raise AuthenticationError("Missing Account or Password") from err
        raise
    self.access_token = data["accessToken"]
    self.refresh_token = data["refreshToken"]

get_user()

Get user details.

Returns:

Type Description
User

A User object.

Raises:

Type Description
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def get_user(self) -> User:
    """Get user details.

    Returns:
        A User object.

    Raises:
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get user")
    self._ensure_login()
    return User(**self._get("/api/client/settings/user/info"))

get_homes()

Get a list of homes.

Returns:

Type Description
list[Home]

A list of Home objects.

Raises:

Type Description
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def get_homes(self) -> list[Home]:
    """Get a list of homes.

    Returns:
        A list of Home objects.

    Raises:
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get home list")
    self._ensure_login()
    return [
        Home(**home_data)
        for home_data in self._get("/api/client/v2/home/family/query")
    ]

get_devices(home_id)

Get a list of devices for a home.

Parameters:

Name Type Description Default
home_id str

The home ID to get devices for.

 required

Returns:

Type Description
list[Device]

A list of Device objects.

Raises:

Type Description
HomeDoesNotExistError

If the home id is not correct.
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def get_devices(self, home_id: str) -> list[Device]:
    """Get a list of devices for a home.

    Args:
        home_id: The home ID to get devices for.

    Returns:
        A list of Device objects.

    Raises:
        HomeDoesNotExistError: If the home id is not correct.
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get devices for home %s", home_id)
    self._ensure_login()
    try:
        return [
            Device(**device_data)
            for device_data in self._get(
                "/api/client/v2/home/device/query", payload={"homeId": home_id}
            )
        ]
    except ApiResponseError as err:
        if err.code == 20450:
            raise HomeDoesNotExistError(home_id) from err
        raise

get_all_devices()

Get a list of all the devices.

Returns:

Type Description
list[Device]

A list of Device objects.

Raises:

Type Description
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def get_all_devices(self) -> list[Device]:
    """Get a list of all the devices.

    Returns:
        A list of Device objects.

    Raises:
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get devices for every homes")
    self._ensure_login()
    return [
        Device(**device_data)
        for device_data in self._get("/api/client/home/device/list")
    ]

get_today_device_data(device_id)

Get power metrics of the current day until now.

Parameters:

Name Type Description Default
device_id str

The device ID to get power metrics for.

 required

Returns:

Type Description
PowerTimeSeries

Metrics of the current day until now.

Raises:

Type Description
UnauthorizedDeviceError

If the device is not authorized or unknown.
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def get_today_device_data(self, device_id: str) -> PowerTimeSeries:
    """Get power metrics of the current day until now.

    Args:
        device_id: The device ID to get power metrics for.

    Returns:
        Metrics of the current day until now.

    Raises:
        UnauthorizedDeviceError: If the device is not authorized or unknown.
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get current day data for device %s", device_id)
    self._ensure_login()
    try:
        return PowerTimeSeries(**self._post(
            "/api/client/home/now/device/realtime", payload={"deviceId": device_id}
        ))
    except ApiResponseError as err:
        if err.code == 20424:
            raise UnauthorizedDeviceError(device_id) from err
        raise

get_realtime_home_data(home_id)

Get current power for the home.

Parameters:

Name Type Description Default
home_id str

The home ID to get current power for.

 required

Returns:

Type Description
PowerMetrics

Current metrics for the home.

Raises:

Type Description
HomeDoesNotExistError

If the home id is not correct.
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def get_realtime_home_data(self, home_id: str) -> PowerMetrics:
    """Get current power for the home.

    Args:
        home_id: The home ID to get current power for.

    Returns:
        Current metrics for the home.

    Raises:
        HomeDoesNotExistError: If the home id is not correct.
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get realtime data for home %s", home_id)
    try:
        return PowerMetrics(**self._get(
            "/api/client/v2/home/device/runData", payload={"homeId": home_id}
        ))
    except ApiResponseError as err:
        if err.code == 20450:
            raise HomeDoesNotExistError(home_id) from err
        raise

get_realtime_device_data(device_id)

Get current power for a device.

Parameters:

Name Type Description Default
device_id str

The device ID to get current power for.

 required

Returns:

Type Description
PowerMetrics

Current metrics for the device.

Raises:

Type Description
UnauthorizedDeviceError

If the device is not authorized or unknown.
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def get_realtime_device_data(self, device_id: str) -> PowerMetrics:
    """Get current power for a device.

    Args:
        device_id: The device ID to get current power for.

    Returns:
        Current metrics for the device.

    Raises:
        UnauthorizedDeviceError: If the device is not authorized or unknown.
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get realtime data for device %s", device_id)
    try:
        return PowerMetrics(**self._post(
            "/api/client/home/now/device/runData", payload={"deviceId": device_id}
        ))
    except ApiResponseError as err:
        if err.code == 20424:
            raise UnauthorizedDeviceError(device_id) from err
        raise

get_history(device_id, period_type, start_date=None)

Get aggregated energy for a period.

Parameters:

Name Type Description Default
device_id str

The device ID to get history for.

 required
period_type int

Possible value:

  • 0: daily values of the calendar month corresponding to start_date
  • 1: today daily values (start_date is ignored) (?)
  • 2: daily values of the current month (start_date is ignored)
  • 3: same than 2 ?
  • 4: total for the current month (start_date is ignored)
 required
start_date datetime | None

The start date.

 None

Returns:

Type Description
EnergyHistory

Data and metrics corresponding to the defined period.

Raises:

Type Description
UnauthorizedDeviceError

If the device is not authorized or unknown.
ParameterVerificationFailedError

If a parameter is not valid (period_type number for example)
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def get_history(
    self, device_id: str, period_type: int, start_date: datetime | None = None
) -> EnergyHistory:
    """Get aggregated energy for a period.

    Args:
        device_id: The device ID to get history for.
        period_type: Possible value:

            - `0`: daily values of the calendar month corresponding to `start_date`
            - `1`: today daily values (`start_date` is ignored) (?)
            - `2`: daily values of the current month (`start_date` is ignored)
            - `3`: same than 2 ?
            - `4`: total for the current month (`start_date` is ignored)
        start_date: The start date.

    Returns:
        Data and metrics corresponding to the defined period.

    Raises:
        UnauthorizedDeviceError: If the device is not authorized or unknown.
        ParameterVerificationFailedError: If a parameter is not valid (`period_type` number for example)
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get history for device %s", device_id)
    if start_date is None:
        if period_type in (1, 2, 4):
            start_ts = 0
        else:
            raise ParameterVerificationFailedError(f"start_date is required for period_type {period_type}")
    else:
        start_ts = int(start_date.timestamp()) if start_date is not None else 0
    try:
        return EnergyHistory(**self._post(
            "/api/client/home/history/home",
            payload={
                "deviceId": device_id,
                "timestamp": start_ts,
                "periodType": period_type,
            },
        ))
    except ApiResponseError as err:
        if err.code == 20424:
            raise UnauthorizedDeviceError(device_id) from err
        if err.code == 20404:
            raise ParameterVerificationFailedError from err
        raise

get_insight(device_id, period_type, start_date=None)

Get energy metrics and statistics of a device for a period.

Parameters:

Name Type Description Default
device_id str

The device ID to get data for.

 required
period_type int

Possible value:

  • 0: 5-minute power measurement for the calendar day corresponding to start_date (insightConsumptionDataDto is None)
  • 1: (not implemented)
  • 2: daily energy for the calendar month corresponding to start_date (deviceRealtimeDto is None)
  • 3: (not implemented)
  • 4: monthly energy for the calendar year corresponding to start_date (deviceRealtimeDto is None)
  • 5: yearly energy, start_date is ignored (?) (deviceRealtimeDto is None)
 required
start_date datetime | None

The start date.

 None

Returns:

Type Description
DeviceInsight

Statistics and metrics corresponding to the defined period.

Raises:

Type Description
UnauthorizedDeviceError

If the device is not authorized or unknown.
ParameterVerificationFailedError

If a parameter is not valid (period_type number for example)
UnauthorizedError

If the Authorization token is not valid.
ApiResponseError

If the API returns a non-successful response.
Source code in src/ecactus/client.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def get_insight(
    self, device_id: str, period_type: int, start_date: datetime | None = None
) -> DeviceInsight:
    """Get energy metrics and statistics of a device for a period.

    Args:
        device_id: The device ID to get data for.
        period_type: Possible value:

            - `0`: 5-minute power measurement for the calendar day corresponding to `start_date` (`insightConsumptionDataDto` is `None`)
            - `1`: (not implemented)
            - `2`: daily energy for the calendar month corresponding to `start_date` (`deviceRealtimeDto` is `None`)
            - `3`: (not implemented)
            - `4`: monthly energy for the calendar year corresponding to `start_date` (`deviceRealtimeDto` is `None`)
            - `5`: yearly energy, `start_date` is ignored (?) (`deviceRealtimeDto` is `None`)
        start_date: The start date.

    Returns:
        Statistics and metrics corresponding to the defined period.

    Raises:
        UnauthorizedDeviceError: If the device is not authorized or unknown.
        ParameterVerificationFailedError: If a parameter is not valid (`period_type` number for example)
        UnauthorizedError: If the Authorization token is not valid.
        ApiResponseError: If the API returns a non-successful response.

    """
    logger.info("Get insight for device %s", device_id)
    if start_date is None:
        if period_type == 5:
            start_ts = 0
        else:
            raise ParameterVerificationFailedError(f"start_date is required for period_type {period_type}")
    else:
        start_ts = int(start_date.timestamp() * 1000)  # timestamp in milliseconds
    try:
        return DeviceInsight(**self._post(
            "/api/client/v2/device/three/device/insight",
            payload={
                "deviceId": device_id,
                "timestamp": start_ts,
                "periodType": period_type,
            },
        ))
    except ApiResponseError as err:
        if err.code == 20424:
            raise UnauthorizedDeviceError(device_id) from err
        if err.code == 20404:
            raise ParameterVerificationFailedError from err
        raise

ecactus.model

Data model for ECOS API.

User

Represents a user.

Attributes:

Name Type Description
username str

The user's username.
nickname str

The user's nickname.
email str

The user's email address.
phone str

The user's phone number.
timezone_id str

The user's time zone ID.
timezone str

The user's time zone offset (e.g., GMT-05:00).
timezone_name str

The user's time zone name (e.g., America/Toronto).
datacenter_phonecode int

The user's datacenter phone code.
datacenter str

The user's datacenter (e.g., EU).
datacenter_host str

The user's datacenter host (e.g., https://api-ecos-eu.weiheng-tech.com).

Home

Represents a home.

Attributes:

Name Type Description
id str

Unique identifier for the home.
name str

Name of the home (or SHARED_DEVICES if the home is shared from another account).
type_int int

Type number of the home.
longitude float | None

Longitude of the home's location, or None if not specified.
latitude float | None

Latitude of the home's location, or None if not specified.
device_number int

Number of devices associated with the home.
relation_type_int int

Type number of relation for the home.
create_time datetime

Time when the home was created.
update_time datetime

Time when the home was last updated.

Device

Represents a device.

Attributes:

Name Type Description
id str

Unique identifier for the device.
alias str

Human-readable name for the device (e.g., "My Device").
state_int int

Current state number of the device.
vpp bool

VPP status.
type_int int

Type number of the device.
serial str

Device serial number.
agent_id str

Unique identifier for the device's agent.
longitude float

Longitude of the device's location.
latitude float

Latitude of the device's location.
device_type str | None

Unknown (e.g., "XX-XXX123").
master_int int

Master status number.

PowerMetrics

Represents a single timestamped power data point.

Attributes:

Name Type Description
timestamp datetime

Timestamp of the data point.
solar float | None

Power generated by the solar panels in watts (W).
grid float | None

To be defined (W).
battery float | None

Power to/from the battery (W).
meter float | None

Power measured (Negative means export to the grid, positive means import from the grid.) (W).
home float | None

Power consumed by the home (W).
eps float | None

Power consumed by the EPS (W).
charge float | None

To be defined (W).

PowerTimeSeries

Represents a series of power metrics over time.

Attributes:

Name Type Description
metrics list[PowerMetrics]

A list of power metrics.

During initialization it can be provided as a dictionary. Example: 

PowerTimeSeries( {
    "solarPowerDps":   { "1740783600":0.0, ... },
    "batteryPowerDps": { "1740783600":0.0, ... },
    "gridPowerDps":    { "1740783600":0.0, ... },
    "meterPowerDps":   { "1740783600":3152.0, ... },
    "homePowerDps":    { "1740783600":3152.0, ... },
    "epsPowerDps":     { "1740783600":0.0, ... }
} )
The model validator automatically transforms this raw dict into a sorted list of PowerMetrics objects.

find_by_timestamp(target, exact=False)

Return the PowerMetrics instance with the timestamp nearest to the target datetime (or with the exact timestamp only).

Parameters:

Name Type Description Default
target datetime

The target datetime to find the PowerMetrics instance for.

 required
exact bool

If True, only return the PowerMetrics instance with the exact timestamp. If False, return the PowerMetrics instance with the timestamp nearest to the target datetime.

 False

Returns:

Type Description
PowerMetrics | None

A PowerMetrics instance corresponding to the target datetime.

find_between(start, end)

Return a list of PowerMetrics instances with timestamps between start and end (inclusive).

Parameters:

Name Type Description Default
start datetime

The start datetime of the range.

 required
end datetime

The end datetime of the range.

 required

Returns:

Type Description
PowerTimeSeries

A list of PowerMetrics instances with timestamps between start and end (inclusive).

EnergyMetric

Represents a single energy data point.

Attributes:

Name Type Description
timestamp datetime

Timestamp of the data point.
energy float | None

The measured energy value at the given timestamp.

EnergyHistory

Represents a series of energy metrics over a period.

Attributes:

Name Type Description
energy_consumption float

The total energy consumption in kWh.
solar_percent float

The percentage of energy produced by solar panels.
metrics list[EnergyMetric]

A list of energy measurement points.

During initialization it can be provided as a dictionary from the "homeEnergyDps" field. Example: 

EnergyHistory( {
"energyConsumption": 12.3,
"solarPercent": 45.6,
"homeEnergyDps": {
    "1733112000": 39.6,
    ...
    "1735707599": 41.3,
    }
} )
The model validator automatically transforms this raw dict into a sorted list of EnergyMetric objects.

EnergyStatistics

Represents energy statistics.

Attributes:

Name Type Description
consumption float

The total energy consumption (kWh).
from_battery float

The energy consumed from battery (kWh).
to_battery float

The energy sent to battery (kWh).
from_grid float

The energy consumed from the grid (kWh).
to_grid float

The energy sent to the grid (kWh).
from_solar float

The energy produced from solar (kWh).
eps float

To be defined.

ConsumptionMetrics

Represents a single timestamped consumption data point.

Attributes:

Name Type Description
timestamp datetime

Timestamp of the data point.
from_battery Annotated[float | None, Field(strict=True, ge=0, alias=fromBatteryDps)]

Energy consumed from battery (kWh).
to_battery Annotated[float | None, Field(strict=True, ge=0, alias=toBatteryDps)]

Energy sent to battery (kWh).
from_grid Annotated[float | None, Field(strict=True, ge=0, alias=fromGridDps)]

Energy consumed from the grid (kWh).
to_grid Annotated[float | None, Field(strict=True, ge=0, alias=toGridDps)]

Energy sent to the grid (kWh).
from_solar Annotated[float | None, Field(strict=True, ge=0, alias=fromSolarDps)]

Energy produced from solar (kWh).
home Annotated[float | None, Field(strict=True, ge=0, alias=homeEnergyDps)]

Home energy consumption (kWh).
eps Annotated[float | None, Field(strict=True, ge=0, alias=epsDps)]

To be defined.
self_powered Annotated[float | None, Field(strict=True, ge=0, alias=selfPoweredDps)]

Autonomous operation (%).

ConsumptionTimeSeries

Represents energy time series.

Attributes:

Name Type Description
metrics list[ConsumptionMetrics]

A list of consumption metrics.

During initialization it can be provided as a dictionary. Example: 

ConsumptionTimeSeries( {
    "fromBatteryDps": {
        "1733976000": 0.0,
        "1733889600": 0.0,
        ...
        "1734062400": 0.0,
    },
    "toBatteryDps": {...},
    "fromGridDps": {...},
    "toGridDps": {...},
    "fromSolarDps": {...},
    "homeEnergyDps": {...},
    "epsDps": {...},
    "selfPoweredDps": {...},
} )
The model validator automatically transforms this raw dict into a sorted list of ConsumptionMetrics objects.

DeviceInsight

Represents various statistics and metrics.

Attributes:

Name Type Description
self_powered int

Autonomous operation (%).
power_timeseries PowerTimeSeries | None

A list of power metrics.
energy_statistics EnergyStatistics | None

Statistics of energy usage.
energy_timeseries ConsumptionTimeSeries | None

A list of energy consumption metrics.

ecactus.exceptions

Ecos client custom exceptions.

EcosApiError

Base exception class for all ECOS API-related errors.

InitializationError(message=None)

Raised when there is an initialization error.

Parameters:

Name Type Description Default
message str | None

The error message.

 None

AuthenticationError(message=None)

Raised when there is an authentication error.

Parameters:

Name Type Description Default
message str | None

The error message.

 None

UnauthorizedError(message=None)

Raised when there is an unauthorized error.

Parameters:

Name Type Description Default
message str | None

The error message.

 None

HomeDoesNotExistError(home_id=None)

Raised when a home does not exist.

Parameters:

Name Type Description Default
home_id str | None

The home ID the error occurred for.

 None

UnauthorizedDeviceError(device_id=None)

Raised when a device is not authorized or unknown.

Parameters:

Name Type Description Default
device_id str | None

The device ID the error occurred for.

 None

ParameterVerificationFailedError(message=None)

Raised when a parameter verification fails.

InvalidJsonError()

Raised when the API returns invalid JSON.

ApiResponseError(code, message)

Raised when the API returns a non-successful response.

Parameters:

Name Type Description Default
code int

The API status code.

 required
message str

The error message.

 required

HttpError(status_code, message)

Raised when an HTTP error occurs while making an API request.

Parameters:

Name Type Description Default
status_code int

The HTTP status code.

 required
message str

The error message.

 required