bcontrolpy is an asynchronous Python client for the TQ-Systems EM300 energy meter, providing authentication, session management, and data retrieval via the MUM webservice.
Install the latest release from PyPI:
pip install bcontrolpyOr install the development version directly from GitHub:
pip install git+https://github.com/ITTV-tools/bcontrolpy.gitimport asyncio
from bcontrolpy import BControl, AuthenticationError
async def main():
# Connect to EM300 meter on local network
async with BControl(ip="192.168.1.100", password="your_password") as bc:
try:
info = await bc.login()
print("Login successful:", info)
data = await bc.get_data()
print("Meter readings:", data)
except AuthenticationError:
print("Authentication failed: check your credentials")
asyncio.run(main())| Method | Returns | Raises |
|---|---|---|
login() -> dict |
{'serial', 'app_version', 'authentication'} |
AuthenticationError, CookieRetrievalError, LoginValueError, CookieValueError |
get_data() -> dict |
Measurement values mapped to human-readable keys | NotAuthenticatedError, HTTP errors |
async_get_data() -> dict |
Alias of get_data() for coordinator usage |
Same as get_data() |
async_test_connection() -> None |
Validates credentials and connectivity | AuthenticationError, BControlCommunicationError |
close() -> None |
Example:
async with BControl(ip="192.168.1.100", password="your_password") as bc:
info = await bc.login()
values = await bc.get_data()bcontrolpy is designed to work with Home Assistant's DataUpdateCoordinator pattern.
from datetime import timedelta
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from bcontrolpy import AuthenticationError, BControl, BControlCommunicationError
class BControlCoordinator(DataUpdateCoordinator):
def __init__(self, hass, entry):
super().__init__(
hass,
logger=LOGGER,
name="bcontrolpy",
update_interval=timedelta(seconds=30),
always_update=False,
)
self.client = BControl(
ip=entry.data["ip"],
password=entry.data["password"],
session=async_get_clientsession(hass),
)
async def _async_update_data(self):
try:
return await self.client.async_get_data()
except AuthenticationError as err:
raise ConfigEntryAuthFailed("Authentication failed") from err
except BControlCommunicationError as err:
raise UpdateFailed(f"Communication error: {err}") from errThis aligns with Home Assistant best practices:
- Reuse Home Assistant's shared
aiohttpsession. - Poll through a single
DataUpdateCoordinator. - Map auth errors to
ConfigEntryAuthFailedand transport errors toUpdateFailed.
Run the provided example script:
python example/example.py --ip 192.168.1.100 --password "your_password"This prints the login details and current meter readings.
- Reuse
aiohttp.ClientSession: Pass an existing session (e.g., from Home Assistant) toBControlto take advantage of connection pooling. An external session will not be closed byBControl.close(). - Handle Exceptions: Catch
AuthenticationErrorto trigger re-authentication flows. - Customize Mapping: The OBIS code mapping resides in
key_mapping.pyand can be extended for additional measurements.
bcontrolpy is distributed under the Apache 2.0 License. See LICENSE for details.