pycsmeter ¶

Python client for interacting with CS water softener valves via BLE.

Modules:

cli –

Command line interface for pycsmeter.
exceptions –

Exceptions for pycsmeter.
valve –

Valve client for interacting with CS valves.

cli ¶

Command line interface for pycsmeter.

Functions:

connect –

Test connection to a valve.
main –

Command line interface for interacting with CS water softener valves.
status –

Get current status from a valve.

connect ¶

connect(address: str, password: str) -> None

Test connection to a valve.

Parameters:

address (str) –

The Bluetooth address of the valve (e.g. 00:11:22:33:44:55)
password (str) –

The valve's connection password

Source code in pycsmeter/cli.py

@main.command()
@click.argument("address")
@click.argument("password")
def connect(address: str, password: str) -> None:
    """Test connection to a valve.

    Parameters:
        address: The Bluetooth address of the valve (e.g. 00:11:22:33:44:55)
        password: The valve's connection password
    """

    async def _run() -> None:
        valve = await _connect_valve(address, password)
        if valve:
            click.echo(f"Successfully connected to valve at {address}")
            await valve.disconnect()

    asyncio.run(_run())

main ¶

main() -> None

Command line interface for interacting with CS water softener valves.

Source code in pycsmeter/cli.py

@click.group()
def main() -> None:
    """Command line interface for interacting with CS water softener valves."""

status ¶

status(address: str, password: str) -> None

Get current status from a valve.

Parameters:

address (str) –

The Bluetooth address of the valve (e.g. 00:11:22:33:44:55)
password (str) –

The valve's connection password

Source code in pycsmeter/cli.py

@main.command()
@click.argument("address")
@click.argument("password")
def status(address: str, password: str) -> None:
    """Get current status from a valve.

    Parameters:
        address: The Bluetooth address of the valve (e.g. 00:11:22:33:44:55)
        password: The valve's connection password
    """

    async def _run() -> None:
        valve = await _connect_valve(address, password)
        if valve:
            try:
                # await valve.get_data2()
                data = await valve.get_data()
                click.echo(_format_valve_data(data))
            except Exception as e:  # noqa: BLE001
                click.echo(f"Error getting valve data: {e}", err=True)
            finally:
                await valve.disconnect()

    asyncio.run(_run())

exceptions ¶

Exceptions for pycsmeter.

Classes:

AuthenticationError –

Raised when authentication to the valve fails.
DataRetrievalError –

Raised when failing to retrieve data from the valve.
PacketError –

Raised for packet parsing or validation errors.
PacketParseError –

Raised when a raw packet cannot be parsed.
PacketValidationError –

Raised when a parsed packet fails validation.
ValveConnectionError –

Raised when BLE connection fails.
ValveError –

Base exception for valve-related errors.

AuthenticationError ¶

Bases: ValveError

Raised when authentication to the valve fails.

DataRetrievalError ¶

Bases: ValveError

Raised when failing to retrieve data from the valve.

PacketError ¶

Bases: ValveError

Raised for packet parsing or validation errors.

PacketParseError ¶

Bases: PacketError

Raised when a raw packet cannot be parsed.

PacketValidationError ¶

Bases: PacketError

Raised when a parsed packet fails validation.

ValveConnectionError ¶

Bases: ValveError

Raised when BLE connection fails.

ValveError ¶

Bases: Exception

Base exception for valve-related errors.

valve ¶

Valve client for interacting with CS valves.

Classes:

AdvancedData –

Advanced valve status information.
DashboardData –

Current valve status.
Valve –

Valve client for interacting with CS valves.
ValveData –

Complete valve status including dashboard, advanced, and history data.

AdvancedData `dataclass` ¶

AdvancedData(regeneration_days: int, days_to_regeneration: int)

Advanced valve status information.

DashboardData `dataclass` ¶

DashboardData(hour: int, minute: int, battery_voltage: float, current_flow: float, soft_water_remaining: int, treated_usage_today: int, peak_flow_today: float, water_hardness: int, regeneration_hour: int)

Current valve status.

Valve ¶

Valve(address: str)

Valve client for interacting with CS valves.

Parameters:

address (str) –

The Bluetooth address of the valve (e.g. 00:11:22:33:44:55)

Methods:

connect –

Connect to the valve via BLE, send authentication using the provided password, and return True if authentication succeeds.
disconnect –

Disconnect from the valve if currently connected.
get_data –

Retrieve Dashboard, Advanced, and History packets from the valve, returning a ValveData object. Retries up to three times on failure.
id –

Return a string identifier for this valve device (its BLE address).

Source code in pycsmeter/valve.py

def __init__(self, address: str):
    """Initialize the Valve client with the given BLE device address.

    Parameters:
        address: The Bluetooth address of the valve (e.g. 00:11:22:33:44:55)
    """
    self.address = address
    self.client = BleakClient(address)
    self.uart = {} # type: ignore  # noqa: PGH003
    self.packet_queue = asyncio.Queue() # type: ignore  # noqa: PGH003
    self.connected = False
    self.authenticated = False
    self.parser = _PacketParser()

connect `async` ¶

connect(password: str) -> bool

Connect to the valve via BLE, send authentication using the provided password, and return True if authentication succeeds.

Parameters:

password (str) –

The valve's connection password

Source code in pycsmeter/valve.py

async def connect(self, password: str) -> bool:
    """Connect to the valve via BLE, send authentication using the provided password, and return True if authentication succeeds.

    Parameters:
        password: The valve's connection password
    """
    if self.connected:
        raise ValveConnectionError(f"Valve {self.address} is already connected")

    # Connect BLE and send hello packet
    try:
        await self.__connect_to_valve()
    except (asyncio.TimeoutError, DataRetrievalError, PacketValidationError):
        return False

    try:
        seed = self.hello_packet.seed
        login_packet = _LoginPacket(seed, password).generate()
        await self.__send_login(login_packet)

        packet = await asyncio.wait_for(self.packet_queue.get(), timeout=15.0)
        hello_packet = await self.parser.parse([packet])
        if not isinstance(hello_packet, _HelloPacket):
            raise PacketValidationError(f"Expected HelloPacket but received {type(hello_packet).__name__}")

        if hello_packet.authenticated:
            self.authenticated = True
            return True
    except asyncio.TimeoutError:
        return False
    return False

disconnect `async` ¶

disconnect() -> None

Disconnect from the valve if currently connected.

Source code in pycsmeter/valve.py

async def disconnect(self) -> None:
    """Disconnect from the valve if currently connected."""
    if self.connected:
        await self.client.disconnect()
        self.connected = False
        self.authenticated = False

get_data `async` ¶

get_data() -> ValveData

Retrieve Dashboard, Advanced, and History packets from the valve, returning a ValveData object. Retries up to three times on failure.

Source code in pycsmeter/valve.py

async def get_data(self) -> ValveData:
    """Retrieve Dashboard, Advanced, and History packets from the valve, returning a ValveData object. Retries up to three times on failure."""
    if not self.connected:
        raise ValveConnectionError(f"Not connected to valve {self.address}. Call connect() first")
    if not self.authenticated:
        raise AuthenticationError(
            f"Not authenticated with valve {self.address}. Authentication failed during connect()",
        )

    for attempt in range(3):
        try:
            return await self.__get_data()
        except (PacketParseError, DataRetrievalError, PacketValidationError):
            if attempt < 2:  # Only sleep if we're going to retry
                await asyncio.sleep(0.1)

    raise DataRetrievalError(f"Failed to retrieve data from valve {self.address} after 3 attempts")

id `async` ¶

id() -> str

Return a string identifier for this valve device (its BLE address).

Source code in pycsmeter/valve.py

async def id(self) -> str:
    """Return a string identifier for this valve device (its BLE address)."""
    return str(self.address)

ValveData `dataclass` ¶

ValveData(dashboard: DashboardData, advanced: AdvancedData, water_usage_history: list[WaterUsageHistoryItem])

Complete valve status including dashboard, advanced, and history data.

Methods:

from_internal –

Create ValveData from internal packet types.
get_history_for_date –

Return the history item for a specific date, if available.
get_history_last_n_days –

Return the most recent n days of history.

from_internal `classmethod` ¶

from_internal(dashboard: DashboardPacket, advanced: AdvancedPacket, water_usage_history: WaterUsageHistoryPacket) -> ValveData

Create ValveData from internal packet types.

Source code in pycsmeter/valve.py

@classmethod
def from_internal(
    cls,
    dashboard: _DashboardPacket,
    advanced: _AdvancedPacket,
    water_usage_history: _WaterUsageHistoryPacket,
) -> "ValveData":
    """Create ValveData from internal packet types."""
    dashboard_data = DashboardData(
        hour=dashboard.hour,
        minute=dashboard.minute,
        battery_voltage=dashboard.battery_volt,
        current_flow=dashboard.current_flow,
        soft_water_remaining=dashboard.soft_remaining,
        treated_usage_today=dashboard.treated_usage_today,
        peak_flow_today=dashboard.peak_flow_today,
        water_hardness=dashboard.water_hardness,
        regeneration_hour=dashboard.regen_hour,
    )

    advanced_data = AdvancedData(
        regeneration_days=advanced.regen_days,
        days_to_regeneration=advanced.days_to_regen,
    )

    # Get all 62 days of water usage history
    history_items = water_usage_history.get_history_last_n_days(62)

    return cls(
        dashboard=dashboard_data,
        advanced=advanced_data,
        water_usage_history=history_items,
    )

get_history_for_date ¶

get_history_for_date(target_date: date) -> Optional[float]

Return the history item for a specific date, if available.

Source code in pycsmeter/valve.py

def get_history_for_date(self, target_date: date) -> Optional[float]:
    """Return the history item for a specific date, if available."""
    for date_val, gallons in self.water_usage_history:
        if date_val == target_date:
            return gallons
    return None

get_history_last_n_days ¶

get_history_last_n_days(n: int) -> list[WaterUsageHistoryItem]

Return the most recent n days of history.

Source code in pycsmeter/valve.py

def get_history_last_n_days(self, n: int) -> list[WaterUsageHistoryItem]:
    """Return the most recent n days of history."""
    return self.water_usage_history[:n]

pycsmeter ¶

cli ¶

connect ¶

main ¶

status ¶

exceptions ¶

AuthenticationError ¶

DataRetrievalError ¶

PacketError ¶

PacketParseError ¶

PacketValidationError ¶

ValveConnectionError ¶

ValveError ¶

valve ¶

AdvancedData dataclass ¶

DashboardData dataclass ¶

Valve ¶

connect async ¶

disconnect async ¶

get_data async ¶

id async ¶

ValveData dataclass ¶

from_internal classmethod ¶

get_history_for_date ¶

get_history_last_n_days ¶

Feedback

AdvancedData `dataclass` ¶

DashboardData `dataclass` ¶

connect `async` ¶

disconnect `async` ¶

get_data `async` ¶

id `async` ¶

ValveData `dataclass` ¶

from_internal `classmethod` ¶