API documentation¶
pyecotrend_ista.PyEcotrendIsta ¶
PyEcotrendIsta(email: str , password: str , logger: Logger | None = None, hass_dir: str | None = None, totp: str | None = None, session: Session | None = None)
A Python client for interacting with the ista EcoTrend API.
This class provides methods to authenticate and interact with the ista EcoTrend API.
Attributes:
-
(_account
) –AccountResponse The account information.
-
(_uuid
) –str The UUID of the consumption unit.
-
(_access_token
) –str | NoneThe access token for API authentication.
-
(_refresh_token
) –str | NoneThe refresh token for obtaining new access tokens.
-
(_access_token_expires_in
) –int The expiration time of the access token.
-
(_header
) –dict [str ,str ]The headers used in HTTP requests.
-
(_support_code
) –str | NoneThe support code for the account.
-
(_start_timer
) –float The start time for tracking elapsed time.
Examples:
Initialize the client and log in:
>>> client = PyEcotrendIsta(email="user@example.com", password="password")
>>> client.login()
Parameters:
-
email
(
) –str The email address used to log in to the ista EcoTrend API.
-
password
(
) –str The password used to log in to the ista EcoTrend API.
-
logger
(
, default:Logger None
) –[DEPRECATED] An optional logger instance for logging messages. Default is None.
-
hass_dir
(
, default:str None
) –[DEPRECATED] An optional directory for Home Assistant configuration. Default is None.
-
totp
(
, default:str None
) –An optional TOTP (Time-based One-Time Password) for two-factor authentication. Default is None.
-
session
(
, default:Session None
) –An optional requests session for making HTTP requests. Default is None.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def __init__(
self,
email: str,
password: str,
logger: logging.Logger | None = None,
hass_dir: str | None = None,
totp: str | None = None,
session: requests.Session | None = None,
) -> None: # numpydoc ignore=ES01,EX01
"""Initialize the PyEcotrendIsta client.
Parameters
----------
email : str
The email address used to log in to the ista EcoTrend API.
password : str
The password used to log in to the ista EcoTrend API.
logger : logging.Logger, optional
[DEPRECATED] An optional logger instance for logging messages. Default is None.
hass_dir : str, optional
[DEPRECATED] An optional directory for Home Assistant configuration. Default is None.
totp : str, optional
An optional TOTP (Time-based One-Time Password) for two-factor authentication. Default is None.
session : requests.Session, optional
An optional requests session for making HTTP requests. Default is None.
"""
if hass_dir:
warnings.warn(
"The 'hass_dir' parameter is deprecated and will be removed in a future release.",
DeprecationWarning,
stacklevel=2,
)
if logger:
warnings.warn(
"The 'logger' parameter is deprecated and will be removed in a future release.",
DeprecationWarning,
stacklevel=2,
)
self._email: str = email.strip()
self._password: str = password
self.loginhelper = LoginHelper(
username=self._email,
password=self._password,
totp=totp,
session=session,
logger=_LOGGER,
)
self.session: requests.Session = self.loginhelper.session
access_token property
writable
¶
access_token
Retrieve the access token, refreshing it if necessary.
This property checks if the access token is still valid. If the token has expired and the client is connected, it refreshes the token. The token is considered expired if the current time minus the start time exceeds the token's expiration period.
Returns:
-
–str The current access token.
Notes
This method will automatically refresh the access token if it has expired.
__login ¶
__login() -> str | None
Perform the login process to obtain an access token.
If the email is a demo account, it logs in using a demo user login function. For other accounts, it retrieves a token using a login helper.
Returns:
-
–str or NoneThe access token if login is successful, None otherwise.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def __login(self) -> str | None: # numpydoc ignore=ES01,EX01
"""
Perform the login process to obtain an access token.
If the email is a demo account, it logs in using a demo user login function.
For other accounts, it retrieves a token using a login helper.
Returns
-------
str or None
The access token if login is successful, None otherwise.
"""
if self._email == DEMO_USER_ACCOUNT:
_LOGGER.debug("Logging in as demo user")
token = self.demo_user_login()
else:
token = self.loginhelper.get_token()
if token:
self.access_token = token["access_token"]
self._access_token_expires_in = token["expires_in"]
self._refresh_token = token["refresh_token"]
return self.access_token
return None
__refresh ¶
__refresh() -> None
Refresh the access token using the refresh token.
This method retrieves a new access token, updates internal variables, and resets the token expiration timer.
Raises:
-
–ParserError If there is an error parsing the request response.
-
–LoginError If there is an authorization failure.
-
–ServerError If there is a server error, connection timeout, or request exception.
Notes
This method assumes self._refresh_token
is already set.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def __refresh(self) -> None: # numpydoc ignore=ES01,EX01
"""
Refresh the access token using the refresh token.
This method retrieves a new access token, updates internal variables,
and resets the token expiration timer.
Raises
------
ParserError
If there is an error parsing the request response.
LoginError
If there is an authorization failure.
ServerError
If there is a server error, connection timeout, or request exception.
Notes
-----
This method assumes `self._refresh_token` is already set.
"""
(
self.access_token,
self._access_token_expires_in,
self._refresh_token,
) = self.loginhelper.refresh_token(self._refresh_token)
self._header["Authorization"] = f"Bearer {self.access_token}"
__set_account ¶
__set_account() -> None
Fetch and set account information from the API.
This method performs an API request to retrieve account information, handles various potential errors that might occur during the request, and sets instance variables accordingly using the response data.
Raises:
-
–ParserError If there is an error parsing the JSON response.
-
–LoginError If the request fails due to an authorization error.
-
–ServerError If the request fails due to a server error, timeout, or other request exceptions.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def __set_account(self) -> None: # numpydoc ignore=ES01,EX01
"""
Fetch and set account information from the API.
This method performs an API request to retrieve account information,
handles various potential errors that might occur during the request,
and sets instance variables accordingly using the response data.
Raises
------
ParserError
If there is an error parsing the JSON response.
LoginError
If the request fails due to an authorization error.
ServerError
If the request fails due to a server error, timeout, or other request exceptions.
"""
self._header = {
"Content-Type": "application/json",
"User-Agent": self.get_user_agent(),
"Authorization": f"Bearer {self.access_token}",
}
url = f"{API_BASE_URL}account"
try:
with self.session.get(url, headers=self._header) as r:
_LOGGER.debug("Performed GET request: %s [%s]:\n%s", url, r.status_code, r.text)
r.raise_for_status()
try:
data = r.json()
except requests.JSONDecodeError as exc:
raise ParserError(
"Loading account information failed due to an error parsing the request response"
) from exc
except requests.HTTPError as exc:
if exc.response.status_code == HTTPStatus.UNAUTHORIZED:
raise LoginError("Loading account information failed due to an authorization failure") from exc
raise ServerError(
"Loading account information failed due to a server error "
f"[{exc.response.status_code}: {exc.response.reason}]"
) from exc
except requests.Timeout as exc:
raise ServerError("Loading account information failed due a connection timeout") from exc
except requests.RequestException as exc:
raise ServerError("Loading account information failed due to a request exception") from exc
self._account = cast(AccountResponse, data)
self._uuid = data["activeConsumptionUnit"]
get_version ¶
get_version() -> str
Get the version of the PyEcotrendIsta client.
Returns:
-
–str The version number of the PyEcotrendIsta client.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def get_version(self) -> str: # numpydoc ignore=EX01,ES01
"""
Get the version of the PyEcotrendIsta client.
Returns
-------
str
The version number of the PyEcotrendIsta client.
"""
return VERSION
login ¶
login(force_login: bool = False, debug: bool = False, **kwargs) -> str | None
Perform the login process if not already connected or forced.
Parameters:
-
force_login
(
, default:bool False
) –If True, forces a fresh login attempt even if already connected. Default is False.
-
debug
(
, default:bool False
) –[DEPRECATED] Flag indicating whether to enable debug logging. Default is False.
-
forceLogin
(
) –bool [DEPRECATED] Use
force_login
instead.
Returns:
-
–str or NoneThe access token if login is successful, None otherwise.
Raises:
-
–LoginError If the login process fails due to an error.
-
–ServerError If a server error occurs during login attempts.
-
–InternalServerError If an internal server error occurs during login attempts.
-
–Exception For any other unexpected errors during the login process.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def login(self, force_login: bool = False, debug: bool = False, **kwargs) -> str | None: # numpydoc ignore=ES01,EX01,PR01,PR02
"""
Perform the login process if not already connected or forced.
Parameters
----------
force_login : bool, optional
If True, forces a fresh login attempt even if already connected. Default is False.
debug : bool, optional
[DEPRECATED] Flag indicating whether to enable debug logging. Default is False.
forceLogin : bool, optional
[DEPRECATED] Use `force_login` instead.
Returns
-------
str or None
The access token if login is successful, None otherwise.
Raises
------
LoginError
If the login process fails due to an error.
ServerError
If a server error occurs during login attempts.
InternalServerError
If an internal server error occurs during login attempts.
Exception
For any other unexpected errors during the login process.
"""
if debug:
warnings.warn(
"The 'debug' parameter is deprecated and will be removed in a future release.",
DeprecationWarning,
stacklevel=2,
)
if "forceLogin" in kwargs:
warnings.warn(
"The 'forceLogin' keyword parameter is deprecated and will be removed in a future release. "
"Use force_login instead.",
DeprecationWarning,
stacklevel=2,
)
force_login = kwargs["forceLogin"]
if not self._is_connected() or force_login:
try:
self.__login()
self.__set_account()
except (KeycloakError, LoginError) as exc:
# Login failed
self._access_token = None
raise LoginError(
"Login failed due to an authorization failure, please verify your email and password"
) from exc
except ServerError as exc:
raise ServerError("Login failed due to a request exception, please try again later") from exc
return self.access_token
userinfo ¶
userinfo(token)
Retrieve user information using the provided access token.
This method constructs an authorization header using the provided access token and sends a GET request to the userinfo endpoint of the provider API. It expects a JSON response with user information.
Parameters:
-
token
(
) –str The access token used for authentication.
Returns:
-
–Any JSON response containing user information.
Raises:
-
–RequestException If an error occurs while making the HTTP request.
Notes
This method constructs an authorization header using the provided access token and sends a GET request to the userinfo endpoint of the provider API. It expects a JSON response with user information.
Examples:
>>> client = PyEcotrendIsta(email="user@example.com", password="password")
>>> token = client.login()
>>> user_info = client.userinfo(token)
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def userinfo(self, token):
"""
Retrieve user information using the provided access token.
This method constructs an authorization header using the provided access token
and sends a GET request to the userinfo endpoint of the provider API. It expects
a JSON response with user information.
Parameters
----------
token : str
The access token used for authentication.
Returns
-------
Any
JSON response containing user information.
Raises
------
requests.exceptions.RequestException
If an error occurs while making the HTTP request.
Notes
-----
This method constructs an authorization header using the provided access token
and sends a GET request to the userinfo endpoint of the provider API.
It expects a JSON response with user information.
Examples
--------
>>> client = PyEcotrendIsta(email="user@example.com", password="password")
>>> token = client.login()
>>> user_info = client.userinfo(token)
"""
return self.loginhelper.userinfo(token=token)
logout ¶
logout() -> None
Perform logout operation by invalidating the current session.
This method invokes the logout functionality in the loginhelper module, passing the current refresh token for session invalidation.
Raises:
-
–KeycloakPostError -
If an error occurs during the logout process. This error is raised based on the response from the logout request.
–
Notes
This method assumes self._refresh_token
is already set.
Examples:
>>> client = PyEcotrendIsta(email="user@example.com", password="password")
>>> client.login()
>>> client.logout()
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def logout(self) -> None:
"""
Perform logout operation by invalidating the current session.
This method invokes the logout functionality in the loginhelper module,
passing the current refresh token for session invalidation.
Raises
------
KeycloakPostError
If an error occurs during the logout process. This error is raised based on the response from the logout request.
Notes
-----
This method assumes `self._refresh_token` is already set.
Examples
--------
>>> client = PyEcotrendIsta(email="user@example.com", password="password")
>>> client.login()
>>> client.logout()
"""
if self.loginhelper.username != DEMO_USER_ACCOUNT:
self.loginhelper.logout(self._refresh_token)
get_uuids ¶
get_uuids() -> list [str ]
Retrieve UUIDs of consumption units registered in the account.
Returns:
-
–list [str ]A list containing UUIDs of consumption units. Each UUID represents a consumption unit, which could be a flat or a house for which consumption readings are provided.
Notes
A consumption unit represents a residence or building where consumption readings are recorded. The UUIDs are extracted from the _residentAndConsumptionUuidsMap
attribute.
Examples:
>>> client = PyEcotrendIsta(email="user@example.com", password="password")
>>> client.login()
>>> uuids = client.get_uuids()
>>> print(uuids)
['uuid1', 'uuid2', 'uuid3']
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def get_uuids(self) -> list[str]: # numpydoc ignore=ES01
"""
Retrieve UUIDs of consumption units registered in the account.
Returns
-------
list[str]
A list containing UUIDs of consumption units. Each UUID represents a consumption unit,
which could be a flat or a house for which consumption readings are provided.
Notes
-----
A consumption unit represents a residence or building where consumption readings are recorded.
The UUIDs are extracted from the `_residentAndConsumptionUuidsMap` attribute.
Examples
--------
>>> client = PyEcotrendIsta(email="user@example.com", password="password")
>>> client.login()
>>> uuids = client.get_uuids()
>>> print(uuids)
['uuid1', 'uuid2', 'uuid3']
"""
return list(self._account.get("residentAndConsumptionUuidsMap", {}).values())
consum_raw ¶
consum_raw(select_year: list [int ] | None = None, select_month: list [int ] | None = None, filter_none: bool = True, obj_uuid: str | None = None) -> dict [str , Any ] | ConsumptionsResponse
Process and filter consumption and cost data for a given consumption unit.
This method processes consumption and cost data obtained from the get_consumption_data
method. It filters and aggregates data based on the parameters provided.
Parameters:
-
select_year
(
, default:list [int ] | NoneNone
) –List of years to filter data by year, default is None.
-
select_month
(
, default:list [int ] | NoneNone
) –List of months to filter data by month, default is None.
-
filter_none
(
, default:bool True
) –Whether to filter out None values in readings, default is True.
-
obj_uuid
(
, default:str | NoneNone
) –UUID of the consumption unit to fetch data for, default is None.
Returns:
-
–dict [str ,Any ] |ConsumptionsResponse Processed data including consumption types, total additional values, last values, last costs, sum by year, and last year compared consumption.
Raises:
-
–Exception If there is an unexpected error during data processing.
Notes
This method processes consumption and cost data obtained from the get_consumption_data
method. It filters and aggregates data based on the parameters provided.
Examples:
>>> api = PyEcotrendIsta()
>>> result = api.consum_raw(select_year=[2023], select_month=[7], filter_none=True, obj_uuid="uuid")
>>> print(result)
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def consum_raw( # noqa: C901
self,
select_year: list[int] | None = None,
select_month: list[int] | None = None,
filter_none: bool = True,
obj_uuid: str | None = None,
) -> dict[str, Any] | ConsumptionsResponse: # noqa: C901
"""
Process and filter consumption and cost data for a given consumption unit.
This method processes consumption and cost data obtained from the `get_consumption_data` method.
It filters and aggregates data based on the parameters provided.
Parameters
----------
select_year : list[int] | None, optional
List of years to filter data by year, default is None.
select_month : list[int] | None, optional
List of months to filter data by month, default is None.
filter_none : bool, optional
Whether to filter out None values in readings, default is True.
obj_uuid : str | None, optional
UUID of the consumption unit to fetch data for, default is None.
Returns
-------
dict[str, Any] | ConsumptionsResponse
Processed data including consumption types, total additional values, last values,
last costs, sum by year, and last year compared consumption.
Raises
------
Exception
If there is an unexpected error during data processing.
Notes
-----
This method processes consumption and cost data obtained from the `get_consumption_data` method.
It filters and aggregates data based on the parameters provided.
Examples
--------
>>> api = PyEcotrendIsta()
>>> result = api.consum_raw(select_year=[2023], select_month=[7], filter_none=True, obj_uuid="uuid")
>>> print(result)
"""
# Fetch raw consumption data for the specified UUID
c_raw: ConsumptionsResponse = self.get_consumption_data(obj_uuid)
if not isinstance(c_raw, dict) or (c_raw.get("consumptions") is None and c_raw.get("costs") is None):
return c_raw
if "consumptions" not in c_raw or not isinstance(c_raw.get("consumptions"), list):
c_raw["consumptions"] = []
consum_types = []
all_dates = []
indices_to_delete_consumption = []
for i, consumption in enumerate(c_raw.get("consumptions", [])):
if (
not isinstance(consumption, dict)
or "readings" not in consumption
or consumption.get("readings") is None
or not isinstance(consumption.get("readings"), list)
):
consumption = {}
continue
for reading in consumption.get("readings", []):
if reading["additionalValue"] is not None or reading["value"] is not None:
consum_types.append(reading["type"])
consum_types = list({consum_type for consum_type in consum_types if i is not None})
new_readings = []
if "date" in consumption:
all_dates.append(consumption["date"])
if select_month is None and select_year is None:
for reading in consumption.get("readings", []):
if filter_none and reading["type"] is not None:
new_readings.append(reading)
elif not filter_none:
new_readings.append(reading)
elif (
select_year is not None
and select_month is not None
and consumption["date"]["year"] in select_year
and consumption["date"]["month"] in select_month
):
for reading in consumption.get("readings", []):
if filter_none and reading["type"] is not None:
new_readings.append(reading)
elif not filter_none:
new_readings.append(reading)
elif select_year is not None and consumption["date"]["year"] in select_year and select_month is None:
for reading in consumption.get("readings", []):
if filter_none and reading["type"] is not None:
new_readings.append(reading)
elif not filter_none:
new_readings.append(reading)
elif select_month is not None and consumption["date"]["month"] in select_month and select_year is None:
for reading in consumption.get("readings", []):
if filter_none and reading["type"] is not None:
new_readings.append(reading)
elif not filter_none:
new_readings.append(reading)
if new_readings:
consumption["readings"] = new_readings
else:
indices_to_delete_consumption.append(i)
for index in sorted(indices_to_delete_consumption, reverse=True):
if index < len(c_raw["consumptions"]):
del c_raw["consumptions"][index]
_all_date = all_dates
new_date = []
sum_by_year = {}
for date in _all_date:
if select_year is None or date["year"] in select_year:
new_date.append(date["year"])
new_date = list(dict.fromkeys(new_date))
cost_consum_types = consum_types
sum_by_year = {typ: {year: 0.0 for year in new_date} for typ in cost_consum_types}
# pylint: disable=too-many-nested-blocks
for item in c_raw.get("consumptions", []):
if "readings" not in item or not item["readings"]:
continue
for reading in item.get("readings", []):
if reading.get("type", None) is None:
continue
for typ in cost_consum_types:
for year in new_date:
if reading["type"] == typ and item["date"]["year"] == year:
if reading["value"]:
sum_by_year[typ][year] += round(
float(reading["value"].replace(",", ".")),
1,
)
else:
sum_by_year[typ][year] += round(
(
float(reading["additionalValue"].replace(",", "."))
if reading["additionalValue"] is not None
else 0.0
),
1,
)
if reading["type"] == "warmwater":
sum_by_year["ww"] = reading["unit"]
elif reading["type"] == "water":
sum_by_year["w"] = reading["unit"]
elif reading["type"] == "heating" and reading["unit"]:
sum_by_year["h"] = reading["unit"]
elif reading["type"] == "heating":
sum_by_year["h"] = reading["additionalUnit"]
indices_to_delete_costs = []
if "costs" not in c_raw or not isinstance(c_raw.get("costs"), list):
c_raw["costs"] = []
for i, costs in enumerate(c_raw.get("costs", [])):
new_readings = []
if "costsByEnergyType" in costs:
if select_month is None and select_year is None:
for reading in costs.get("costsByEnergyType", []):
if filter_none and reading["type"] is not None:
new_readings.append(reading)
elif not filter_none:
new_readings.append(reading)
elif (
select_year is not None
and select_month is not None
and costs["date"]["year"] in select_year
and costs["date"]["month"] in select_month
):
for reading in costs.get("costsByEnergyType", []):
if filter_none and reading["type"] is not None:
new_readings.append(reading)
elif not filter_none:
new_readings.append(reading)
elif select_year is not None and costs["date"]["year"] in select_year and select_month is None:
for reading in costs.get("costsByEnergyType", []):
if filter_none and reading["type"] is not None:
new_readings.append(reading)
elif not filter_none:
new_readings.append(reading)
elif select_month is not None and costs["date"]["month"] in select_month and select_year is None:
for reading in costs.get("costsByEnergyType", []):
if filter_none and reading["type"] is not None:
new_readings.append(reading)
elif not filter_none:
new_readings.append(reading)
if new_readings:
costs["costsByEnergyType"] = new_readings
else:
indices_to_delete_costs.append(i)
for index in sorted(indices_to_delete_costs, reverse=True):
if "costs" in c_raw and index < len(c_raw["costs"]):
del c_raw["costs"][index]
for key in [
"consumptionsBillingPeriods",
"costsBillingPeriods",
"resident",
"co2Emissions",
"co2EmissionsBillingPeriods",
]:
if key in c_raw:
del c_raw[key]
consumptions: list = c_raw.get("consumptions", [])
costs: list = c_raw.get("costs", [])
combined_data = []
for cost_entry in costs:
for consumption_entry in consumptions:
# Überprüfen, ob die Daten das gleiche Datum haben
if cost_entry["date"] == consumption_entry["date"]:
# Wenn ja, kombiniere die Kosten- und Verbrauchsdaten in einem Eintrag
combined_entry = {
"date": cost_entry["date"],
"consumptions": consumption_entry["readings"],
"costs": cost_entry["costsByEnergyType"],
}
combined_data.append(combined_entry)
total_additional_values = {}
total_additional_custom_values = {}
for consumption_unit in consumptions:
if "readings" not in consumption_unit or not consumption_unit["readings"]:
continue
for reading in consumption_unit.get("readings", []):
if reading["type"] is None or (reading["value"] is None and reading["additionalValue"] is None):
continue
if reading["type"] not in total_additional_custom_values:
total_additional_custom_values[reading["type"]] = 0.0
if reading["additionalValue"]:
total_additional_custom_values[reading["type"]] += round(
float(reading["additionalValue"].replace(",", ".")), 1
)
else:
total_additional_custom_values[reading["type"]] += round(
(float(reading["value"].replace(",", ".")) if reading["value"] is not None else 0.0),
1,
)
if reading["type"] == "warmwater":
total_additional_custom_values["ww"] = reading["additionalUnit"]
elif reading["type"] == "water":
total_additional_custom_values["w"] = reading["additionalUnit"]
elif reading["type"] == "heating" and reading["additionalUnit"]:
total_additional_custom_values["h"] = reading["additionalUnit"]
elif reading["type"] == "heating":
total_additional_custom_values["h"] = reading["unit"]
if reading["type"] not in total_additional_values:
total_additional_values[reading["type"]] = 0.0
if reading["value"]:
total_additional_values[reading["type"]] += round(float(reading["value"].replace(",", ".")), 1)
else:
total_additional_values[reading["type"]] += round(
(
float(reading["additionalValue"].replace(",", "."))
if reading["additionalValue"] is not None
else 0.0
),
1,
)
if reading["type"] == "warmwater":
total_additional_values["ww"] = reading["unit"]
elif reading["type"] == "water":
total_additional_values["w"] = reading["unit"]
elif reading["type"] == "heating" and reading["unit"]:
total_additional_values["h"] = reading["unit"]
elif reading["type"] == "heating":
total_additional_values["h"] = reading["additionalUnit"]
last_value = None
last_custom_value = None
last_year_compared_consumption = None
if consumptions:
last_value = {}
last_custom_value = {}
last_year_compared_consumption = {}
if len(consumptions) > 0 and "readings" in consumptions[0] and consumptions[0]["readings"]:
for reading in consumptions[0]["readings"]:
if reading["type"] is None or (reading["value"] is None and reading["additionalValue"] is None):
continue
if reading["comparedConsumption"]:
last_year_compared_consumption[reading["type"]] = reading["comparedConsumption"]
last_year_compared_consumption[reading["type"]]["comparedValue"] = float(
last_year_compared_consumption[reading["type"]]["comparedValue"].replace(",", ".")
)
if reading["value"]:
last_year_compared_consumption[reading["type"]]["nowYearValue"] = float(
reading["value"].replace(",", ".")
)
elif reading["additionalValue"]:
last_year_compared_consumption[reading["type"]]["nowYearValue"] = float(
reading["additionalValue"].replace(",", ".")
)
if "period" in last_year_compared_consumption[reading["type"]]:
del last_year_compared_consumption[reading["type"]]["period"]
if reading["type"] not in last_custom_value:
last_custom_value[reading["type"]] = 0.0
if reading["additionalValue"]:
last_custom_value[reading["type"]] += float(reading["additionalValue"].replace(",", "."))
else:
last_custom_value[reading["type"]] += (
float(reading["value"].replace(",", ".")) if reading["value"] is not None else 0.0
)
if reading["type"] == "warmwater":
last_custom_value["ww"] = reading["additionalUnit"]
elif reading["type"] == "water":
last_custom_value["w"] = reading["additionalUnit"]
elif reading["type"] == "heating" and reading["additionalUnit"]:
last_custom_value["h"] = reading["additionalUnit"]
elif reading["type"] == "heating":
last_custom_value["h"] = reading["unit"]
if reading["type"] not in last_value:
last_value[reading["type"]] = 0.0
if reading["value"]: # reading["type"] in ("warmwater", "water", "heating") and
last_value[reading["type"]] += float(reading["value"].replace(",", "."))
else:
last_value[reading["type"]] += (
float(reading["additionalValue"].replace(",", "."))
if reading["additionalValue"] is not None
else 0.0
)
if reading["type"] == "warmwater":
last_value["ww"] = reading["unit"]
elif reading["type"] == "water":
last_value["w"] = reading["unit"]
elif reading["type"] == "heating" and reading["additionalUnit"]:
last_value["h"] = reading["unit"]
elif reading["type"] == "heating":
last_value["h"] = reading["additionalUnit"]
last_custom_value["month"] = consumptions[0]["date"]["month"]
last_custom_value["year"] = consumptions[0]["date"]["year"]
last_value["month"] = consumptions[0]["date"]["month"]
last_value["year"] = consumptions[0]["date"]["year"]
last_costs = None
if costs:
if last_costs is None:
last_costs = {}
for costs_by_energy_type in costs[0]["costsByEnergyType"]:
# pylint: disable=too-many-boolean-expressions
if (
costs_by_energy_type is None
or "type" not in costs_by_energy_type
or costs_by_energy_type["type"] is None
or "comparedCost" not in costs_by_energy_type
or costs_by_energy_type["comparedCost"] is None
or "smiley" not in costs_by_energy_type["comparedCost"]
or costs_by_energy_type["comparedCost"]["smiley"] is None
or "comparedPercentage" not in costs_by_energy_type["comparedCost"]
or costs_by_energy_type["comparedCost"]["comparedPercentage"] is None
):
continue
if costs_by_energy_type["type"] not in last_costs:
last_costs[costs_by_energy_type["type"]] = 0.0
last_costs[costs_by_energy_type["type"]] += costs_by_energy_type["value"]
last_costs["unit"] = costs_by_energy_type["unit"]
if costs_by_energy_type["type"] == "warmwater":
if costs_by_energy_type["comparedCost"]["smiley"] == ["MAD", "EQUAL"]:
last_costs["ww"] = costs_by_energy_type["comparedCost"]["comparedPercentage"]
elif costs_by_energy_type["comparedCost"]["smiley"] in ["HAPPY"]:
last_costs["ww"] = costs_by_energy_type["comparedCost"]["comparedPercentage"] * -1
elif costs_by_energy_type["type"] == "water":
if costs_by_energy_type["comparedCost"]["smiley"] == ["MAD", "EQUAL"]:
last_costs["w"] = costs_by_energy_type["comparedCost"]["comparedPercentage"]
elif costs_by_energy_type["comparedCost"]["smiley"] in ["HAPPY"]:
last_costs["w"] = costs_by_energy_type["comparedCost"]["comparedPercentage"] * -1
elif costs_by_energy_type["type"] == "heating":
if costs_by_energy_type["comparedCost"]["smiley"] in ["MAD", "EQUAL"]:
last_costs["h"] = costs_by_energy_type["comparedCost"]["comparedPercentage"]
elif costs_by_energy_type["comparedCost"]["smiley"] in ["HAPPY"]:
last_costs["h"] = costs_by_energy_type["comparedCost"]["comparedPercentage"] * -1
last_costs["month"] = costs[0]["date"]["month"]
last_costs["year"] = costs[0]["date"]["year"]
return CustomRaw.from_dict(
{
"consum_types": consum_types,
"combined_data": None, # combined_data,
"total_additional_values": total_additional_values,
"total_additional_custom_values": total_additional_custom_values,
"last_value": last_value,
"last_custom_value": last_custom_value,
"last_costs": last_costs,
"all_dates": None, # all_dates,
"sum_by_year": sum_by_year,
"last_year_compared_consumption": last_year_compared_consumption,
}
).to_dict()
get_consumption_data ¶
get_consumption_data(obj_uuid: str | None = None) -> ConsumptionsResponse
Fetch consumption data from the API for a specific consumption unit.
This method sends a GET request to the ista EcoTrend API to retrieve consumption data for a specific consumption unit identified by the provided UUID. If no UUID is provided, the method uses the UUID associated with the instance.
Parameters:
-
obj_uuid
(
, default:str None
) –The UUID of the consumption unit. If not provided, defaults to the UUID associated with the instance (
self._uuid
).
Returns:
-
–ConsumptionsResponse A dictionary containing the consumption data fetched from the API.
Raises:
-
–LoginError If the API responds with an error indicating authorization failure.
-
–ParserError If there is an error parsing the request response.
-
–ValueError If the provided UUID is invalid.
-
–ServerError If there is a server error, connection timeout, or request exception.
Examples:
>>> api = PyEcotrendIsta()
>>> data = api.get_consumption_data(obj_uuid="uuid")
>>> print(data)
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def get_consumption_data(self, obj_uuid: str | None = None) -> ConsumptionsResponse:
"""
Fetch consumption data from the API for a specific consumption unit.
This method sends a GET request to the ista EcoTrend API to retrieve consumption data
for a specific consumption unit identified by the provided UUID. If no UUID is provided,
the method uses the UUID associated with the instance.
Parameters
----------
obj_uuid : str, optional
The UUID of the consumption unit. If not provided,
defaults to the UUID associated with the instance (`self._uuid`).
Returns
-------
ConsumptionsResponse
A dictionary containing the consumption data fetched from the API.
Raises
------
LoginError
If the API responds with an error indicating authorization failure.
ParserError
If there is an error parsing the request response.
ValueError
If the provided UUID is invalid.
ServerError
If there is a server error, connection timeout, or request exception.
Examples
--------
>>> api = PyEcotrendIsta()
>>> data = api.get_consumption_data(obj_uuid="uuid")
>>> print(data)
"""
params = {"consumptionUnitUuid": obj_uuid or self._uuid}
url = f"{API_BASE_URL}consumptions"
try:
with self.session.get(
url,
params=params,
headers=self._header,
) as result:
_LOGGER.debug("Performed GET request: %s [%s]:\n%s", url, result.status_code, result.text[:100])
result.raise_for_status()
try:
return cast(ConsumptionsResponse, result.json())
except requests.JSONDecodeError as exc:
raise ParserError("Loading consumption data failed due to an error parsing the request response") from exc
except requests.HTTPError as exc:
if exc.response.status_code == HTTPStatus.UNAUTHORIZED:
raise LoginError("Loading consumption data failed failed due to an authorization failure") from exc
if exc.response.status_code == HTTPStatus.BAD_REQUEST:
raise ValueError(
f"Invalid UUID. Retrieving data for consumption unit {obj_uuid or self._uuid} failed"
) from exc
raise ServerError(
"Loading consumption data failed due to a server error " f"[{exc.response.status_code}: {exc.response.reason}]"
) from exc
except requests.Timeout as exc:
raise ServerError("Loading consumption data failed due a connection timeout") from exc
except requests.RequestException as exc:
raise ServerError("Loading consumption data failed due to a request exception") from exc
get_consumption_unit_details ¶
get_consumption_unit_details() -> ConsumptionUnitDetailsResponse
Retrieve details of the consumption unit from the API.
Returns:
-
–ConsumptionUnitDetailsResponse A dictionary containing the details of the consumption unit.
Raises:
-
–LoginError If the API responds with an authorization failure.
-
–ParserError If there is an issue with decoding the JSON response
-
–ServerError If there is a server error, connection timeout, or request exception.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def get_consumption_unit_details(self) -> ConsumptionUnitDetailsResponse: # numpydoc ignore=ES01,EX01
"""
Retrieve details of the consumption unit from the API.
Returns
-------
ConsumptionUnitDetailsResponse
A dictionary containing the details of the consumption unit.
Raises
------
LoginError
If the API responds with an authorization failure.
ParserError
If there is an issue with decoding the JSON response
ServerError
If there is a server error, connection timeout, or request exception.
"""
url = f"{API_BASE_URL}menu"
try:
with self.session.get(url, headers=self._header) as r:
_LOGGER.debug("Performed GET request: %s [%s]:\n%s", url, r.status_code, r.text)
r.raise_for_status()
try:
return cast(ConsumptionUnitDetailsResponse, r.json())
except requests.JSONDecodeError as exc:
raise ParserError(
"Loading consumption unit details failed due to an error parsing the request response"
) from exc
except requests.HTTPError as exc:
if exc.response.status_code == HTTPStatus.UNAUTHORIZED:
raise LoginError("Loading consumption unit details failed failed due to an authorization failure") from exc
raise ServerError(
"Loading consumption unit details failed due to a server error "
f"[{exc.response.status_code}: {exc.response.reason}]"
) from exc
except requests.Timeout as exc:
raise ServerError("Loading consumption unit details failed due a connection timeout") from exc
except requests.RequestException as exc:
raise ServerError("Loading consumption unit details failed due to a request exception") from exc
get_support_code ¶
get_support_code() -> str | None
Return the support code associated with the instance.
Returns:
-
–str or NoneThe support code associated with the instance, or None if not set.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def get_support_code(self) -> str | None: # numpydoc ignore=ES01,EX01
"""
Return the support code associated with the instance.
Returns
-------
str or None
The support code associated with the instance, or None if not set.
"""
return getattr(self, "_account", {}).get("supportCode")
get_user_agent ¶
get_user_agent() -> str
Return the User-Agent string used for HTTP requests.
Returns:
-
–str The User-Agent string.
Notes
This method provides a static User-Agent string commonly used for web browsers.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def get_user_agent(self) -> str: # numpydoc ignore=ES01,EX01
"""
Return the User-Agent string used for HTTP requests.
Returns
-------
str
The User-Agent string.
Notes
-----
This method provides a static User-Agent string commonly used for web browsers.
"""
return (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.4951.67"
" Safari/537.36"
)
demo_user_login ¶
demo_user_login() -> GetTokenResponse
Retrieve authentication tokens for the demo user.
Returns:
-
–GetTokenResponse A TypedDict containing authentication tokens including 'accessToken', 'accessTokenExpiresIn', and 'refreshToken'.
Raises:
-
–ParserError If there is an error parsing the request response.
-
–ServerError If there is a server error, connection timeout, or request exception.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def demo_user_login(self) -> GetTokenResponse: # numpydoc ignore=ES01,EX01
"""
Retrieve authentication tokens for the demo user.
Returns
-------
GetTokenResponse
A TypedDict containing authentication tokens including 'accessToken',
'accessTokenExpiresIn', and 'refreshToken'.
Raises
------
ParserError
If there is an error parsing the request response.
ServerError
If there is a server error, connection timeout, or request exception.
"""
url = f"{API_BASE_URL}demo-user-token"
try:
self._header["User-Agent"] = self.get_user_agent()
with self.session.get(url, headers=self._header) as r:
_LOGGER.debug("Performed GET request %s [%s]:\n%s", url, r.status_code, r.text)
r.raise_for_status()
try:
data = r.json()
key = iter(GetTokenResponse.__annotations__)
token = {next(key): value for value in data.values()}
return cast(GetTokenResponse, token)
except requests.JSONDecodeError as exc:
raise ParserError("Demo user authentication failed due to an error parsing the request response") from exc
except requests.HTTPError as exc:
raise ServerError(
"Demo user authentication failed due to a server error " f"[{exc.response.status_code}: {exc.response.reason}]"
) from exc
except requests.Timeout as exc:
raise ServerError("Demo user authentication failed due a connection timeout") from exc
except requests.RequestException as exc:
raise ServerError("Demo user authentication failed due to a request exception") from exc
get_account ¶
get_account() -> AccountResponse | None
Retrieve the account information.
Returns the _account
attribute if it exists, otherwise returns None.
Returns:
-
–AccountResponse | NoneAccount information if available, otherwise None.
Source code in src/pyecotrend_ista/pyecotrend_ista.py
def get_account(self) -> AccountResponse | None: # numpydoc ignore=ES01,EX01
"""
Retrieve the account information.
Returns the `_account` attribute if it exists, otherwise returns None.
Returns
-------
AccountResponse | None
Account information if available, otherwise None.
"""
return getattr(self, "_account", None)