pulse-eight-matrix-client 0.5.0

An async Python client for controlling Pulse8 HDBaseT Matrix devices via their REST API.

Pulse8 HDBaseT Matrix Client

An async Python client for controlling Pulse8 HDBaseT Matrix devices via their REST API.

Features

• ✅ Fully async using aiohttp
• ✅ Type hints throughout
• ✅ Comprehensive data models
• ✅ Easy-to-use API
• ✅ Route by port number or name
• ✅ Context manager support
• ✅ Proper error handling

Installation

Using uv:

uv pip install -e .

Or with pip:

pip install -e .

Quick Start

import asyncio
from pulse_eight_matrix_client import PulseEightMatrixClient

async def main():
    # Connect to your matrix
    async with PulseEightMatrixClient(host="192.168.1.150") as client:
        # Get system info
        details = await client.get_system_details()
        print(f"Model: {details.model}, Version: {details.version}")
        
        # List all ports
        ports = await client.get_ports()
        for port in ports:
            print(f"{port.mode} {port.bay}: {port.name}")
        
        # Route input to output
        await client.set_port(input_bay=7, output_bay=5)
        
        # Or route by name
        await client.route_by_name("Sky", "Bedroom TV")

asyncio.run(main())

Auto-Polling for Instant State Access

Enable background polling to maintain cached state. This eliminates HTTP calls when querying routing information:

import asyncio
from pulse_eight_matrix_client import CachingPulseEightMatrixClient

async def main():
    # Enable auto-polling (polls every 5 seconds by default)
    async with CachingPulseEightMatrixClient(
        host="192.168.1.150",
        poll_interval=5
    ) as client:
        # All get_ports() calls now return from cache instantly!
        ports = await client.get_ports()
        
        # Check routing from cache (no HTTP call)
        routing = client.get_cached_routing_map()
        
        # See which input feeds an output (instant)
        source = client.get_cached_output_source(output_bay=5)
        
        # Find where an input is going (instant)
        destinations = client.get_cached_input_destinations(input_bay=7)
        
        # Changes automatically update the cache
        await client.set_port(7, 5)

Benefits of auto-polling:

• Zero latency for state queries - no HTTP overhead
• Always up-to-date - background polling keeps state fresh
• Automatic updates - cache refreshes after routing changes
• Real-time monitoring - detect changes as they happen

API Reference

Client Initialization

from pulse_eight_matrix_client import PulseEightMatrixClient

client = PulseEightMatrixClient(
    host="192.168.1.150",
    port=80,
    timeout=10,
)

Parameters:

• host: IP address or hostname of the matrix
• port: HTTP port (default: 80)
• timeout: Request timeout in seconds (default: 10)

System Methods

get_system_details() -> SystemDetails

Get system information including model, version, serial number, and status.

details = await client.get_system_details()
print(f"Model: {details.model}")
print(f"Version: {details.version}")
print(f"Serial: {details.serial}")
print(f"Status: {details.status_message}")

get_system_features() -> SystemFeatures

Get system features and capabilities.

features = await client.get_system_features()
print(f"HDBaseT: {features.hdbaset}")
print(f"Input Bays: {features.video.input['Bays']}")

Port Methods

get_ports() -> List[Port]

Get list of all ports (inputs and outputs).

When using CachingPulseEightMatrixClient, use get_cached_ports() to return cached data instantly. get_ports() will always perform a live HTTP call.

ports = await client.get_ports()
for port in ports:
    print(f"{port.mode} {port.bay}: {port.name}")

get_ports() on CachingPulseEightMatrixClient

Get list of all ports with a fresh HTTP call, bypassing cache.

# Always fetches from the matrix, even with auto_poll enabled
ports = await client.get_ports()

#### `get_input_ports() -> List[Port]`
Get list of input ports only.

#### `get_output_ports() -> List[Port]`
Get list of output ports only.

#### `get_input_details(bay: int) -> PortDetails`
Get detailed information about an input port.

```python
details = await client.get_input_details(7)
print(f"Has Signal: {details.has_signal}")
print(f"Signal: {details.signal}")

get_output_details(bay: int) -> PortDetails

Get detailed information about an output port.

Routing Methods

set_port(input_bay: int, output_bay: int) -> SetPortResponse

Route an input to an output by bay numbers.

response = await client.set_port(input_bay=7, output_bay=5)
print(response.message)  # "Changed"

route_by_name(input_name: str, output_name: str) -> SetPortResponse

Route an input to an output by port names (case-insensitive).

await client.route_by_name("Sky", "Bedroom TV")

get_port_by_name(name: str) -> Optional[Port]

Find a port by its name.

port = await client.get_port_by_name("Kitchen TV")
if port:
    print(f"Found: {port.mode} bay {port.bay}")

Cached Query Methods

These methods only work when auto_poll=True and return data instantly from the cache without making HTTP calls.

get_cached_output_source(output_bay: int) -> Optional[int]

Get which input is currently routed to an output.

# Requires auto_poll=True
source = client.get_cached_output_source(5)
if source is not None:
    print(f"Output 5 is receiving from Input {source}")

get_cached_input_destinations(input_bay: int) -> List[int]

Get which outputs are receiving from an input.

# Requires auto_poll=True
destinations = client.get_cached_input_destinations(7)
print(f"Input 7 is routed to outputs: {destinations}")

get_cached_routing_map() -> Dict[int, int]

Get a complete mapping of all outputs to their source inputs.

# Requires auto_poll=True
routing = client.get_cached_routing_map()
# Returns: {output_bay: input_bay, ...}
# Example: {0: 1, 2: 0, 5: 7}

get_last_poll_time() -> Optional[float]

Get the Unix timestamp of the last successful poll.

timestamp = client.get_last_poll_time()

get_cache_age() -> Optional[float]

Get the age of the cached data in seconds.

age = client.get_cache_age()
print(f"Cache is {age:.1f} seconds old")

get_cached_ports() -> List[Port]

Get all ports from the cached state.

# Only available on CachingPulseEightMatrixClient
ports = client.get_cached_ports()
for port in ports:
    print(f"{port.mode} {port.bay}: {port.name}")

SystemDetails

• model: Device model (e.g., "FF88A")
• version: Firmware version
• serial: Serial number
• mac: MAC address
• status_message: Current status
• status: Status code

Port

• bay: Port bay number
• mode: "Input" or "Output"
• type: Port type (e.g., "HDMI2", "HDBT-TXLITE")
• status: Connection status
• name: User-assigned name
• receive_from: (Output only) Which input it's receiving from

PortDetails

Extends Port with additional fields:

• has_signal: Whether signal is detected
• signal: Signal information
• hdcp: HDCP status
• allowed_sinks: List of allowed output bays

Error Handling

The client raises specific exceptions for different error types:

from pulse_eight_matrix_client import (
    PulseEightConnectionError,
    PulseEightAPIError
)

try:
    async with PulseEightMatrixClient(host="192.168.1.150") as client:
        await client.set_port(7, 5)
except PulseEightConnectionError as e:
    print(f"Connection failed: {e}")
except PulseEightAPIError as e:
    print(f"API error: {e}")

Context Manager

The client supports async context managers for automatic session management:

# Session is automatically created and closed
async with PulseEightMatrixClient(host="192.168.1.150") as client:
    await client.get_ports()

Or manage the session manually:

client = PulseEightMatrixClient(host="192.168.1.150")
await client.connect()
try:
    await client.get_ports()
finally:
    await client.close()

Monitoring Example

Below is a minimal example using CachingPulseEightMatrixClient. It polls in the background and allows instant reads from the cache.

async def monitor_matrix():
    async with CachingPulseEightMatrixClient(
        host="192.168.1.150",
        poll_interval=5,
    ) as client:
        while True:
            routing = client.get_cached_routing_map()
            print(f"Routing map: {routing}")
            await asyncio.sleep(1)

Development

Install development dependencies:

uv pip install -e ".[dev]"

Run tests:

pytest

License

MIT