Skip to main content
Version: 10.1

Get RIOS Device Connection Status

This script connects to rWeb and periodically polls the status of the selected RIOS device.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
# "requests",
# ]
# ///
"""Poll device connection status periodically, prints output if it changes"""

import requests
import json
import urllib.parse
import time
from datetime import datetime

# Enter your rWeb address here
BASE_URL = 'https://rweb.local.com'

HEADERS = {'Content-Type':'application/json',
'Accept': 'application/json'}

# Enter your user credentials here
CREDENTIALS = {'login': 'admin', 'password': 'admin123'}

def url(resource, params=None):
"""Turn a resource into a url.

Uses base_url and optionally encodes a hash of parameters"""
querystring = ""
if params != None:
querystring = "?"+urllib.parse.urlencode(params)
return BASE_URL + resource + querystring

def wrap(data):
"""Convert an object to an rWeb suitable json string

Wraps the data in the standard rweb {'data': } object """
return json.dumps({'data': data})

def login(session):
"""Log in to a session.

The authentication cookie is saved to the session"""
result = session.post(url('/authentication'),
wrap({'login':'admin', 'password':'admin123'}))
assert(result.status_code == requests.codes.created)

def logout(session):
"""Close a session"""
session.delete(url('/authentication'))

def find_device(session, name):
"""Find a device based on its name

Query must result in only one device"""
result = session.get(url('/devices', {'q': name}))

device_list = json.loads(result.content)
assert(device_list['rows'] == 1)

return device_list['data'][0]

def main():
with requests.Session() as session:
session.headers = HEADERS
# This is insecure but may be needed if you haven't set up an SSL certificate for rWeb yet
session.verify = False
login(session)

# Put in your device's display name here
device = find_device(session, 'RVM-10G')

resource = device['device_resource']
last_status = None
try:
while True:
device = json.loads(session.get(url(resource)).content)
status = device['connection_status']
if status != last_status:
print(f"{datetime.now()}: {status}")

last_status = status
time.sleep(5)
except KeyboardInterrupt:
logout(session)
print(f"Stopped monitoring and logged out from {BASE_URL}.")

if __name__ == '__main__':
main()