pulse-eight-matrix-client 0.2.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 Pulse8MatrixClient

async def main():
    # Connect to your matrix
    async with Pulse8MatrixClient(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:

# 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

client = Pulse8MatrixClient(
    host="192.168.1.150",
    port=80,
    timeout=10,
    auto_poll=False,
    poll_interval=5
)

Parameters:

• host: IP address or hostname of the matrix
• port: HTTP port (default: 80)
• timeout: Request timeout in seconds (default: 10)
• auto_poll: Enable background polling to maintain cached state (default: False)
• poll_interval: How often to poll in seconds when auto_poll is enabled (default: 5)

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")

Data Models

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 (
    Pulse8ConnectionError,
    Pulse8APIError
)

try:
    async with Pulse8MatrixClient(host="192.168.1.150") as client:
        await client.set_port(7, 5)
except Pulse8ConnectionError as e:
    print(f"Connection failed: {e}")
except Pulse8APIError 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 Pulse8MatrixClient(host="192.168.1.150") as client:
    await client.get_ports()

Or manage the session manually:

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

Monitoring Example

Without Auto-Polling

Poll for changes with explicit HTTP calls:

async def monitor_matrix():
    async with Pulse8MatrixClient(host="192.168.1.150") as client:
        while True:
            ports = await client.get_ports()  # HTTP call each time
            # Display current routing
            for port in ports:
                if port.mode == "Output" and port.receive_from:
                    print(f"{port.name} <- Input {port.receive_from}")
            
            await asyncio.sleep(5)  # Poll every 5 seconds

asyncio.run(monitor_matrix())

With Auto-Polling (Recommended)

Let the client handle polling, query state instantly:

async def monitor_matrix():
    async with Pulse8MatrixClient(
        host="192.168.1.150",
        auto_poll=True,
        poll_interval=5
    ) as client:
        previous_routing = {}
        
        while True:
            # Get current routing from cache - instant, no HTTP call!
            routing = client.get_cached_routing_map()
            
            # Detect changes
            for output_bay, input_bay in routing.items():
                if previous_routing.get(output_bay) != input_bay:
                    print(f"Output {output_bay} changed to Input {input_bay}")
            
            previous_routing = routing
            await asyncio.sleep(1)  # Check for changes every second

asyncio.run(monitor_matrix())

Development

Install development dependencies:

uv pip install -e ".[dev]"

Run tests:

pytest

License

MIT