scripts/websocket_test.py

#!/usr/bin/env python3

# Example of streaming sensor values from openbmc using the /subscribe api
# requires websockets package to be installed

import asyncio
import ssl
import websockets
import base64
import json
import argparse

parser = argparse.ArgumentParser()
parser.add_argument("--host", help="Host to connect to", required=True)
parser.add_argument(
    "--username", help="Username to connect with", default="root")
parser.add_argument("--password", help="Password to use", default="0penBmc")

args = parser.parse_args()

sensor_type_map = {
    "voltage": "Volts",
    "power": "Watts",
    "fan": "RPM",
    "fan_tach": "RPM",
    "temperature": "Degrees C",
    "altitude": "Meters",
    "current": "Amps",
    "energy": "Joules",
    "cfm": "CFM"
}


async def hello():
    uri = 'wss://{}/subscribe'.format(args.host)
    ssl_context = ssl.SSLContext()
    authbytes = "{}:{}".format(args.username, args.password).encode('ascii')
    auth = "Basic {}".format(base64.b64encode(authbytes).decode('ascii'))
    headers = {"Authorization": auth}
    async with websockets.connect(uri, ssl=ssl_context, extra_headers=headers) as websocket:
        request = json.dumps({
            "paths": ["/xyz/openbmc_project/sensors"],
            "interfaces": ["xyz.openbmc_project.Sensor.Value"]
        })
        await websocket.send(request)

        while True:
            payload = await websocket.recv()
            j = json.loads(payload)
            path = j.get("path", "unknown/unknown")
            name = path.split("/")[-1]
            sensor_type = path.split("/")[-2]
            units = sensor_type_map.get(sensor_type, "")
            properties = j.get("properties", [])
            value = properties.get("Value", None)
            if value is None:
                continue
            print(f"{name:<20} {value:4.02f} {units}")

asyncio.get_event_loop().run_until_complete(hello())