ros3d-controllerctl

#!/usr/bin/env python2
#
# Copyright (c) 2015 Open-RnD Sp. z o.o.
#
# Permission is hereby granted, free of charge, to any person
# obtaining a copy of this software and associated documentation
# files (the "Software"), to deal in the Software without
# restriction, including without limitation the rights to use, copy,
# modify, merge, publish, distribute, sublicense, and/or sell copies
# of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from __future__ import absolute_import, print_function
import argparse
import logging
import requests
import json
import sys
from urlparse import urljoin
from pprint import pprint


def parse_arguments():
    default_host = 'localhost'
    default_port = 8090

    parser = argparse.ArgumentParser(description='Ros3D Controller Tool')
    parser.add_argument('--host',
                        help='Host address, default: {}'.format(default_host),
                        default=default_host)
    parser.add_argument('--port',
                        help='Port, default: {}'.format(default_port),
                        default=default_port, type=int)
    parser.add_argument('-l', '--list-parameters', action='store_true',
                        help='List parameters',
                        default=False)
    parser.add_argument('-s', '--set-parameter',
                        help='Set parameters, format: <param>:<type>:<value>, ex: focus_distance_m:float:20; parameter types: float, int, str',
                        nargs='+',
                        default=[])
    parser.add_argument('--save-snapshot', help='Save a snapshot of paramters',
                        default=False, action='store_true')
    parser.add_argument('--list-snapshots', help='List available snapshots',
                        default=False, action='store_true')
    parser.add_argument('--download-snapshot', help='Download snapshot',
                        default=None, type=int)
    parser.add_argument('--delete-snapshot', help='Delete snapshot',
                        default=None, type=int)
    parser.add_argument('--delete-all-snapshots', help='Delete all snapshots',
                        default=False, action='store_true')
    return parser.parse_args()


def print_response(resp):
    """Helper for pretty printing of a response"""
    if resp.status_code == 200:
        pprint(resp.json())
    else:
        logging.error('request failed with code: %d', resp.status_code)
        logging.error('response text:\n\t%s', resp.text)


def list_params(url):
    durl = urljoin(url, '/api/parameters/list')
    logging.debug('request to URL: %s', durl)

    print_response(requests.get(durl))


def as_ptype(ptype, val):
    if ptype == 'float':
        return float(val)
    elif ptype == 'str':
        return str(val)
    elif ptype == 'int':
        return int(val)
    raise ValueError('unknown parameter type %s' % (ptype))


def decode_parameters_data(params):
    """Helper for converting a list of parameters in format
    <param>:<type>:<value> to a dict suitable for using as data for
    /parameters/update API call

    """
    data = {}
    for par in params:
        try:
            name, ptype, val = par.split(':')
            data[name] = {'value': as_ptype(ptype, val)}
        except ValueError:
            logging.exception('failed to parse parameter %s', par)
            raise

    logging.debug('decoded params: %s', data)
    return data


def set_params(url, params):
    logging.debug('set parameters: %s', params)

    data = decode_parameters_data(params)

    durl = urljoin(url, '/api/parameters/update')
    logging.debug('request to URL: %s', durl)

    print_response(requests.put(durl, json=data))


def save_snapshot(url):
    logging.debug('save snapshot')

    durl = urljoin(url, '/api/snapshots/capture')
    logging.debug('request to URL: %s', durl)

    print_response(requests.post(durl))


def list_snapshots(url):
    logging.debug('list snapshots')

    durl = urljoin(url, '/api/snapshots/list')
    logging.debug('request to URL: %s', durl)

    print_response(requests.get(durl))


def delete_all_snapshots(url):
    logging.debug('delete all snapshots')

    durl = urljoin(url, '/api/snapshots/list')
    logging.debug('request to URL: %s', durl)

    print_response(requests.delete(durl))


def download_snapshot(url, snapshot_id):
    logging.debug('download snapshot %s', snapshot_id)

    durl = urljoin(url, '/api/snapshots/{:d}'.format(snapshot_id))
    logging.debug('request to URL: %s', durl)

    print_response(requests.get(durl))


def delete_snapshot(url, snapshot_id):
    logging.debug('delete snapshot %s', snapshot_id)

    durl = urljoin(url, '/api/snapshots/{:d}'.format(snapshot_id))
    logging.debug('request to URL: %s', durl)

    print_response(requests.delete(durl))


def main(opts):
    logging.info('connecting to: %s:%d', opts.host, opts.port)

    opts.url = 'http://{:s}:{:d}'.format(opts.host, opts.port)

    if opts.list_parameters:
        # list parameters
        list_params(opts.url)
    elif opts.set_parameter:
        set_params(opts.url, opts.set_parameter)
    elif opts.save_snapshot:
        save_snapshot(opts.url)
    elif opts.list_snapshots:
        list_snapshots(opts.url)
    elif opts.download_snapshot:
        download_snapshot(opts.url, opts.download_snapshot)
    elif opts.delete_snapshot:
        delete_snapshot(opts.url, opts.delete_snapshot)
    elif opts.delete_all_snapshots:
        delete_all_snapshots(opts.url)


if __name__ == '__main__':

    opts = parse_arguments()

    logging.basicConfig(level=logging.DEBUG)

    if len(sys.argv) == 1:
        # default to list
        opts.list_parameters = True

    main(opts)