An asynchronous Python library for controlling LG WebOS TVs. This is an async port of the popular pywebostv library.
- Asynchronous API for controlling LG WebOS TVs
- WebSocket-based communication
- Support for all major TV controls (media, system, input, applications, TV channels, sources, etc.)
- Real-time event subscriptions (volume, channel, foreground app, power state, audio output)
- SSDP-based network device discovery
- Secure SSL/TLS connections with certificate handling
- Modern Python async/await syntax
- Type hints for better IDE support
- Comprehensive test coverage
- Python 3.11+
- aiohttp>=3.8.0
- websockets>=15.0.1
pip install asyncwebostvimport asyncio
from asyncwebostv.connection import WebOSClient
from asyncwebostv.controls import (
MediaControl, SystemControl, ApplicationControl, TvControl, InputControl, SourceControl
)
async def main():
# Create a WebOS client
client = WebOSClient("192.168.1.100") # Replace with your TV's IP
# Connect to TV
await client.connect()
# Register with the TV (if needed)
store = {} # Dictionary to receive the client key
async for status in client.register(store):
if status == WebOSClient.PROMPTED:
print("Please accept the connection on your TV")
elif status == WebOSClient.REGISTERED:
print(f"Registration successful, client key: {store.get('client_key')}")
# Use the client key for future sessions to avoid re-pairing
# client_key = store.get("client_key")
# Create control interfaces
media = MediaControl(client)
system = SystemControl(client)
# Control TV
await media.volume_up()
await system.notify("Hello from AsyncWebOSTV!")
# Close connection
await client.close()
asyncio.run(main())AsyncWebOSTV supports real-time event subscriptions so you don't have to poll
the TV. Six subscriptions are available: get_volume, get_audio_output,
get_sound_output, get_current_channel, get_current (foreground app),
and power_state.
import asyncio
from asyncwebostv.connection import WebOSClient
from asyncwebostv.controls import MediaControl, SystemControl
async def main():
client = WebOSClient("192.168.1.100", client_key="your-client-key")
await client.connect()
media = MediaControl(client)
system = SystemControl(client)
async def on_volume(success, payload):
if success:
print(f"Volume: {payload.get('volume')} muted={payload.get('muted')}")
async def on_power(success, payload):
if success:
print(f"Power state: {payload.get('state')}")
await media.subscribe_get_volume(on_volume)
await system.subscribe_power_state(on_power)
try:
await asyncio.sleep(60) # receive events for a minute
finally:
await media.unsubscribe_get_volume()
await system.unsubscribe_power_state()
await client.close()
asyncio.run(main())Reconnect note: after
await client.close(), discard your control objects (MediaControl,SystemControl, etc.) and create fresh ones against the new client before resubscribing. See the subscription spec for the full reconnect contract and the complete list of event payloads.
AsyncWebOSTV supports secure SSL/TLS connections to WebOS TVs. The library provides two classes for secure connections:
Low-level secure client with enhanced SSL certificate handling:
import asyncio
from asyncwebostv import SecureWebOSClient, extract_certificate
async def main():
# Extract and save certificate from TV
cert_file = "tv_cert.pem"
await extract_certificate("192.168.1.100", 3001, cert_file)
# Create secure client with certificate verification
client = SecureWebOSClient(
host="192.168.1.100",
port=3001,
secure=True,
cert_file=cert_file,
verify_ssl=True,
client_key="your-client-key" # Optional
)
# Connect and use as normal
await client.connect()
# Close connection
await client.close()
asyncio.run(main())High-level secure client with SSL certificate handling:
import asyncio
from asyncwebostv import SecureWebOSTV
async def main():
# Create secure TV client
tv = SecureWebOSTV(
host="192.168.1.100",
port=3001,
cert_file="tv_cert.pem", # Optional
verify_ssl=True # Set to False to skip verification
)
# Extract and save certificate
# await tv.get_certificate("tv_cert.pem")
# Connect and register if needed
await tv.connect()
if not tv.client_key:
client_key = await tv.register()
# Now use the client property to access lower-level API
client = tv.client
# Close connection
await tv.close()
asyncio.run(main())The library includes utility functions for working with TV certificates:
from asyncwebostv import extract_certificate, verify_certificate
# Extract certificate from TV
cert_pem = await extract_certificate("192.168.1.100", 3001, "tv_cert.pem")
# Verify if a saved certificate matches the current one on the TV
matches = await verify_certificate("tv_cert.pem", "192.168.1.100", 3001)Unlike some other libraries, AsyncWebOSTV does not save client keys to disk by default. Instead, it returns the client key to the caller through a provided dictionary.
# During registration
store = {}
async for status in client.register(store):
pass
# After registration, the client key is in the store dictionary
client_key = store.get("client_key")
print(f"Your client key: {client_key}")
# Save it however you want (database, config file, etc.)For subsequent connections, you can provide the client key directly:
client = WebOSClient("192.168.1.100", client_key="your-client-key")This approach gives you more control over how client keys are stored and managed in your application.
Detailed documentation lives in the docs/ directory:
- Subscription API — real-time events (volume, channel, power state, foreground app, audio output) with payloads, callback contracts, and reconnect rules
- SSL / Secure Connections — certificate handling and secure-mode design
- Async Migration Spec — the
pywebostv→asyncwebostvport
- Clone the repository:
git clone https://github.com/droman42/asyncwebostv.git
cd asyncwebostv- Install development dependencies:
pip install -e ".[dev]"- Run tests:
pytest- Format code:
black .
isort .We welcome contributions! Please see our Contributing Guidelines for details.
This project is licensed under the MIT License - see the LICENSE file for details.
- Original
pywebostvlibrary for inspiration - LG for their WebOS platform