pounder_test/pounder_test.py

#!/usr/bin/python3
"""
Description: Test Stabilizer communication and DDS configuration.

Author: Ryan Summers
"""

import socket
import json

HOST = '10.0.16.99'
PORT = 1235

def do_request(s, request):
    """ Perform a request with the Stabilizer.

    Args:
        s: The socket to the stabilizer.
        request: The request to transmit.

    Returns:
        The received response object.
    """
    # Transform the value field.
    request['value'] = json.dumps(request['value'], separators=[',', ':']).replace('"', "'")
    data = (json.dumps(request, separators=[',', ':']) + '\n').encode('ascii')
    s.send(data)

    response = b''
    while not response.endswith(b'\n'):
        response += s.recv(1024)

    # Decode the value array
    response = json.loads(response.decode('ascii'))
    response['value'] = response['value'].replace("'", '"')
    response['value'] = json.loads(response['value'])

    return response


def read_attribute(s, attribute_name):
    """ Read an attribute on the Stabilizer device.

    Args:
        s: The socket to the stabilizer.
        attribute_name: The name of the endpoint to write to (the attribute name).

    Returns:
        The value of the attribute. May be a string or a dictionary.
    """
    request = {
        "req": "Read",
        "attribute": attribute_name,
        "value": "",
    }

    response = do_request(s, request)

    if 'code' not in response or response['code'] != 200:
        raise Exception(f'Failed to read {attribute_name}: {response}')

    return response['value']


def write_attribute(s, attribute_name, attribute_value):
    """ Write an attribute on the Stabilizer device.

    Args:
        s: The socket to the stabilizer.
        attribute_name: The name of the endpoint to write to (the attribute name).
        attribute_value: The value to write to the attribute. May be a string or a dictionary.
    """
    request = {
        "req": "Write",
        "attribute": attribute_name,
        "value": attribute_value,
    }

    response = do_request(s, request)

    if 'code' not in response or response['code'] != 200:
        raise Exception(f'Failed to write {attribute_name}: {response}')


def main():
    """ Main program entry point. """
    with socket.socket() as s:

        # Connect to the stabilizer.
        s.connect((HOST, PORT))

        # A sample configuration for an output channel.
        channel_config = {
            'attenuation': 0.0,
            'parameters': {
                'phase_offset': 0.5,
                'frequency': 100.0e6,
                'amplitude': 1.0,
                'enabled': True,
            }
        }

        # Configure OUT0 and read it back.
        write_attribute(s, "pounder/out0", channel_config)
        print('Pounder OUT0: ', read_attribute(s, "pounder/out0"))

        print('Pounder IN1: ', read_attribute(s, "pounder/in1"))
        print('Pounder OUT1: ', read_attribute(s, "pounder/out1"))

if __name__ == '__main__':
    main()