mkdir rcms_api
cd rcms_api

python -m pip install requests

import hashlib
import hmac
import json
import time
import requests
from urllib.parse import urlencode

class RCMSClient:
    def __init__(self, endpoint, client_id, client_secret):
        self.endpoint = endpoint.rstrip('/')
        self.client_id = client_id
        self.client_secret = client_secret

    def _build_canonical_query_string(self, timestamp, query_params=None):
        """Build canonical query string according to API specification"""
        params = {
            'apiVersion': '1.0',
            'clientId': self.client_id,
            'signatureVersion': '1.0',
            'timestamp': timestamp,
            'uniqueCode': self.client_id
        }
        
        if query_params:
            params.update(query_params)
            
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        return '&'.join([f"{k}={v}" for k, v in sorted_params])

    def _generate_signature(self, method, path, timestamp, query_params=None):
        canonical_query_string = self._build_canonical_query_string(timestamp, query_params)
        string_to_sign = method + path + canonical_query_string + self.client_secret
        
        key = self.client_id + self.client_id
        hmac_obj = hmac.new(
            key.encode('utf-8'),
            string_to_sign.encode('utf-8'),
            hashlib.sha256
        )
        
        signature = hmac_obj.hexdigest().lower()
        return signature, canonical_query_string

    def get(self, path, query_params=None):
        if not path.startswith('/'):
            path = '/' + path

        timestamp = str(int(time.time() * 1000))
        signature, canonical_query_string = self._generate_signature('GET', path, timestamp, query_params)

        headers = {
            'apiVersion': '1.0',
            'signatureVersion': '1.0',
            'uniqueCode': self.client_id,
            'clientId': self.client_id,
            'timestamp': timestamp,
            'signature': signature
        }

        url = f"{self.endpoint}{path}?{canonical_query_string}"
        response = requests.get(url, headers=headers)
        return response.json()

    def put(self, path, body, query_params=None):
        if not path.startswith('/'):
            path = '/' + path

        timestamp = str(int(time.time() * 1000))
        signature, canonical_query_string = self._generate_signature('PUT', path, timestamp, query_params)

        headers = {
            'apiVersion': '1.0',
            'signatureVersion': '1.0',
            'uniqueCode': self.client_id,
            'clientId': self.client_id,
            'timestamp': timestamp,
            'signature': signature,
            'Content-Type': 'application/json'
        }

        url = f"{self.endpoint}{path}?{canonical_query_string}"
        response = requests.put(url, headers=headers, json=body)
        return response.json()

Example Scripts

1. Get Device Total Statistics

from rcms_client import RCMSClient
import json

client = RCMSClient(
    endpoint="https://rcms-cloud.robustel.net",
    client_id="YOUR_CLIENT_ID",  # Replace with your Client ID
    client_secret="YOUR_SECRET_KEY"  # Replace with your Secret Key
)

response = client.get('/api/link/dashboard/deviceTotal')
print(json.dumps(response, indent=2))

2. Get Device Status Data

from rcms_client import RCMSClient
import json
import time

client = RCMSClient(
    endpoint="https://rcms-cloud.robustel.net",
    client_id="YOUR_CLIENT_ID",  # Replace with your Client ID
    client_secret="YOUR_SECRET_KEY"  # Replace with your Secret Key
)

device_sn = "YOUR_DEVICE_SN"  # Replace with your device SN
current_time = int(time.time() * 1000)
one_hour_ago = current_time - (1 * 60 * 60 * 1000)

query_params = {
    "pageNum": 1,
    "pageSize": 10,
    "beginTime": one_hour_ago,
    "endTime": current_time
}

response = client.get(f'/api/link/device/devices/{device_sn}/statusData', query_params)
print(json.dumps(response, indent=2))

3. Update Device Group

from rcms_client import RCMSClient
import json

client = RCMSClient(
    endpoint="https://rcms-cloud.robustel.net",
    client_id="YOUR_CLIENT_ID",  # Replace with your Client ID
    client_secret="YOUR_SECRET_KEY"  # Replace with your Secret Key
)

response = client.put(
    '/api/link/devices/YOUR_DEVICE_SN/group',
    {"groupName": "Demo"}
)
print(json.dumps(response, indent=2))

File Structure

rcms_api/
├── rcms_client.py
├── test_api_deviceTotal.py
├── device_status.py
└── test_api_updateGroup.py

Expected Responses

Device Total Response

{
  "msg": "Success",
  "code": 0,
  "data": {
    "deviceTotal": 14,
    "onlineTotal": 5,
    "offlineTotal": 9,
    "registeredTotal": 14,
    "unRegisteredTotal": 0,
    "userTotal": 9
  }
}

Device Status Response

{
  "msg": "Success",
  "code": 0,
  "data": [
    {
      "data": {
        "system": {
          "system_time": "...",
          "device_model": "...",
          "firmware_version": "...",
          "uptime": "..."
        },
        "active_link": {
          "ip": "...",
          "gateway": "...",
          "link_uptime": "..."
        }
      }
    }
  ]
}

Troubleshooting