xref: /openbmc/bmcweb/scripts/websocket_test.py (revision a529a6aa) 
1#!/usr/bin/env python3
2
3# Example of streaming sensor values from openbmc using the /subscribe api
4# requires websockets package to be installed
5
6import argparse
7import asyncio
8import base64
9import json
10import ssl
11
12import websockets
13
14parser = argparse.ArgumentParser()
15parser.add_argument("--host", help="Host to connect to", required=True)
16parser.add_argument(
17    "--username", help="Username to connect with", default="root"
18)
19parser.add_argument("--password", help="Password to use", default="0penBmc")
20parser.add_argument(
21    "--ssl", default=True, action=argparse.BooleanOptionalAction
22)
23
24args = parser.parse_args()
25
26sensor_type_map = {
27    "voltage": "Volts",
28    "power": "Watts",
29    "fan": "RPM",
30    "fan_tach": "RPM",
31    "temperature": "Degrees C",
32    "altitude": "Meters",
33    "current": "Amps",
34    "energy": "Joules",
35    "cfm": "CFM",
36}
37
38
39async def hello():
40    protocol = "ws"
41    if args.ssl:
42        protocol += "s"
43    uri = "{}://{}/subscribe".format(protocol, args.host)
44    ssl_context = ssl.SSLContext()
45    authbytes = "{}:{}".format(args.username, args.password).encode("ascii")
46    auth = "Basic {}".format(base64.b64encode(authbytes).decode("ascii"))
47    headers = {"Authorization": auth}
48    async with (
49        websockets.connect(uri, ssl=ssl_context, extra_headers=headers)
50        if args.ssl
51        else websockets.connect(uri, extra_headers=headers)
52    ) as websocket:
53        request = json.dumps(
54            {
55                "paths": ["/xyz/openbmc_project/sensors"],
56                "interfaces": ["xyz.openbmc_project.Sensor.Value"],
57            }
58        )
59        await websocket.send(request)
60
61        while True:
62            payload = await websocket.recv()
63            j = json.loads(payload)
64            path = j.get("path", "unknown/unknown")
65            name = path.split("/")[-1]
66            sensor_type = path.split("/")[-2]
67            units = sensor_type_map.get(sensor_type, "")
68            properties = j.get("properties", [])
69            value = properties.get("Value", None)
70            if value is None:
71                continue
72            print(f"{name:<20} {value:4.02f} {units}")
73
74
75asyncio.get_event_loop().run_until_complete(hello())
76