Skip to content

Python

Note

Since the device can not run on a dedicated domain, the HTTPS encryption is done by a certificate signed by the RACOM certificate authority.

Import libraries and declare constants

import json
import requests
import urllib3
import time
import logging
import sys

logging.basicConfig(level=logging.INFO)
HTTP_OK = 200

password = "admin"
username = "admin"
language = "en"

target = "https://ripex2a.racom.eu/cgi-bin/"
remote = "10.10.10.2"

User defined constants:

  • password and username are kept in default just for purpose of this example
  • target is globaly accesible RipEX2 router located in Racom HQ
  • remote is IP address of device in radio network
    • if RPC should be executed on device behind radio hop, remote must be set
    • if RPC should be executed on localy attached device, remote = "" should be used

Supress SSL warnings

SSL warnings can be suppressed with following line. Otherwise python interpreter will complain about invalid certificate.

# Disable warnings related to self-signed certificates
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

Login to the node

Both username and password must be passed to be verified. Detailed description in authentication.

# Login
login_headers = {'Content-Type': 'application/json', 'apikey': ''}
login_data = {'password': password, 'username': username, 'language_code': language}
logging.info(f"Login to {target}")
login_rsp = requests.post(f"{target}/login.cgi", data=json.dumps(login_data),
    headers=login_headers, verify=False)

if login_rsp.status_code != HTTP_OK:
    logging.error(f"Failed to login to {target}")
    sys.exit(42) #error

User credentials are passed as data in login_data object. Special header must be set for login RPC - object login_headers.

If login RPC is returned successfuly, apikey must be taken from response and used for all other RPCs.

Update node configuration

Execute RPC to get current device configuration settings_get and prepare some configuration change.

req_headers = {'Content-Type': 'application/json', 'apikey': login_rsp.json()['token']}
# Read configuration
logging.info(f"Get configuration from {target}")
get_cnf = {'method': 'settings_get', 'target': remote}
cnf_data = requests.post(f"{target}/rpc.cgi", data=json.dumps(get_cnf),
    headers=req_headers, verify=False)
logging.debug(f"Cnf get response from {target}: {cnf_data.json()}")

# Update configuration
cnf_data = cnf_data.json()['result']
cnf_data['config_data']['main']['RR_StationDesc'] = 'Test 123'

Prepare req_headers object with apikey.

In the example above, RR_StationDesc attribute is modified.

Execute RPC to save updated configuration settings_save_init.

# Save updated configuration
set_cnf = {'method': 'settings_save_init',
    'params': {'config_data': cnf_data['config_data']},
    'target': remote}
logging.info(f"Update configuration on {target}.")
cnf_rsp = requests.post(f"{target}/rpc.cgi", data=json.dumps(set_cnf),
    headers=req_headers, verify=False)
logging.debug(f"Cnf update response on {target}: {cnf_rsp.json()}")

if 'error' in cnf_rsp.json():
    logging.error(f"Failed to update cnf on {target} with {cnf_rsp.json()['error']}")
    sys.exit(42) #error

logging.info(f"Wait until cnf update is applied on {target}")
poll_result = cnf_rsp.json()['result']
# Wait until cnf is applied
poll_req = {'method': 'settings_save_reconnect',
             'params': {'session_id': poll_result['session_id']},
             'target': remote}

Some errors can be returned from node. In such case this example exits. Some robust implementation could try to do some recovery and repeat settings_save_init again.

Successful response of settings_save_init contains info, how to poll for settings_save_init result. Details can be found on settings_save_init page.

Polling intervals are base on type of communication (local / remote) and the difficulty of action to be executed. In order to get the best performace out of network, these intervals must be respected.

Wait until configuration change is applied

Busy-waiting until configuration is ready. Detailed description of poll object in poll init

interval_time = poll_result['interval']
delay_time = poll_result['delay']
# Wait before first poll
time.sleep(delay_time)
# Poll for result
spent = 0
while spent < poll_result['timeout']:
    spent = spent + interval_time
    logging.info(f"Poll for cnf update finished on {target} session {poll_result['session_id']}")
    logging.debug(f"Request {json.dumps(poll_req)}")
    rsp = requests.post(f"{target}/rpc.cgi", data=json.dumps(poll_req),
        headers=req_headers, verify=False)
    if 'error' in rsp.json():
         if rsp.json()['error']['code'] != 'rpc_async_action_in_progress':
              logging.error(f"Cnf update poll failed on {target} with {rsp.json()['error']}")
              sys.exit(42) #error
    else:
         break
    time.sleep(interval_time)

Sequence is always the same

  • delay_time before the first try to poll
  • interval_time between each two polls

Cleanup

Close active session and evaluate timeout condition.

logging.info(f"Logout from {target}.")
logout_rsp = requests.post(f"{target}/logout.cgi",
    headers=req_headers, verify=False)

if spent >= poll_result['timeout']:
     logging.error(f"Cnf update on {target} failed - timeout.")
     sys.exit(42) #error

logging.info(f"Cnf update on {target} done.")
sys.exit(0) #OK

In case all RPC's are executed successfuly, action can be considered completed.