bcontrolpy

bcontrolpy is an asynchronous Python client for the TQ-Systems EM300 energy meter, providing authentication, session management, and data retrieval via the MUM webservice.

Table of Contents

• Installation

• Quickstart

• Usage

  • Python API
    • Home Assistant Integration
  • Command Line Example

• Configuration

• License

Installation

Install the latest release from PyPI:

pip install bcontrolpy

Or install the development version directly from GitHub:

pip install git+https://github.com/ITTV-tools/bcontrolpy.git

Quickstart

import 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())

Usage

Python API

Example:

async with BControl(ip="192.168.1.100", password="your_password") as bc:
    info = await bc.login()
    values = await bc.get_data()

Home Assistant Integration

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 err

This aligns with Home Assistant best practices:

• Reuse Home Assistant's shared aiohttp session.
• Poll through a single DataUpdateCoordinator.
• Map auth errors to ConfigEntryAuthFailed and transport errors to UpdateFailed.

Command Line Example

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.

Configuration

• Reuse aiohttp.ClientSession: Pass an existing session (e.g., from Home Assistant) to BControl to take advantage of connection pooling. An external session will not be closed by BControl.close().
• Handle Exceptions: Catch AuthenticationError to trigger re-authentication flows.
• Customize Mapping: The OBIS code mapping resides in key_mapping.py and can be extended for additional measurements.

License

bcontrolpy is distributed under the Apache 2.0 License. See LICENSE for details.