src/ctrl-gen.py

#!/usr/bin/env python
from __future__ import print_function
from collections import OrderedDict
import getopt
import sys
import yaml

class quoted(str): pass

def quoted_presenter(dumper, data):
    return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='"')
yaml.add_representer(quoted, quoted_presenter)

class literal(str): pass

def literal_presenter(dumper, data):
    return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
yaml.add_representer(literal, literal_presenter)

def ordered_dict_presenter(dumper, data):
    return dumper.represent_dict(data.items())
yaml.add_representer(OrderedDict, ordered_dict_presenter)

def dict_constructor(loader, node):
    return OrderedDict(loader.construct_pairs(node))
_mapping_tag = yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
yaml.add_constructor(_mapping_tag, dict_constructor)

class IntField(object):
    def __init__(self, name, position, length, signed):
        self.name = name
        self.position = position
        self.length = length
        self.signed = signed

        if not self.length in [1, 2, 4]:
            raise Exception("bad length " + str(self.length))

        self.user_type = ('u' if not signed else '') + 'int' + str(length * 8) + '_t'

    def getter_sig(self):
        return "{0}* {1}".format(self.user_type, self.name)

    def unpack(self):
        if self.length == 1:
            return "*{0} = data[{1}];".format(self.name, self.position)
        elif self.length == 2:
            return "*{0} = SW_TO_SHORT(data + {1});".format(self.name, self.position)
        elif self.length == 4:
            return "*{0} = DW_TO_INT(data + {1});".format(self.name, self.position)

    def setter_sig(self):
        return "{0} {1}".format(self.user_type, self.name)

    def pack(self):
        if self.length == 1:
            return "data[{0}] = {1};".format(self.position, self.name)
        elif self.length == 2:
            return "SHORT_TO_SW({0}, data + {1});".format(self.name, self.position)
        elif self.length == 4:
            return "INT_TO_DW({0}, data + {1});".format(self.name, self.position)

    def spec(self):
        rep = [('position', self.position), ('length', self.length)]
        if self.signed:
            rep.append(('signed', True))
        return rep

    @staticmethod
    def load(spec):
        return IntField(spec['name'], spec['position'], spec['length'], spec['signed'] if signed in spec else False)

def load_field(name, spec):
    if spec['type'] == 'int':
        return IntField(name, spec['position'], spec['length'], spec.get('signed', False))
    else:
        raise Exception("unknown field type '{0}'".format(spec['type']))

GETTER_TEMPLATE = """/** @ingroup ctrl
 * {gen_doc}
 * @param devh UVC device handle
 * {args_doc}
 * @param req_code UVC_GET_* request to execute
 */
uvc_error_t uvc_get_{control_name}(uvc_device_handle_t *devh, {args_signature}, enum uvc_req_code req_code) {{
  uint8_t data[{control_length}];
  uvc_error_t ret;

  ret = libusb_control_transfer(
    devh->usb_devh,
    REQ_TYPE_GET, req_code,
    {control_code} << 8,
    {unit_fn} << 8 | devh->info->ctrl_if.bInterfaceNumber,
    data,
    sizeof(data),
    0);

  if (ret == sizeof(data)) {{
    {unpack}
    return UVC_SUCCESS;
  }} else {{
    return ret;
  }}
}}
"""

SETTER_TEMPLATE = """/** @ingroup ctrl
 * {gen_doc}
 * @param devh UVC device handle
 * {args_doc}
 */
uvc_error_t uvc_set_{control_name}(uvc_device_handle_t *devh, {args_signature}) {{
  uint8_t data[{control_length}];
  uvc_error_t ret;

  {pack}

  ret = libusb_control_transfer(
    devh->usb_devh,
    REQ_TYPE_SET, UVC_SET_CUR,
    {control_code} << 8,
    {unit_fn} << 8 | devh->info->ctrl_if.bInterfaceNumber,
    data,
    sizeof(data),
    0);

  if (ret == sizeof(data))
    return UVC_SUCCESS;
  else
    return ret;
}}
"""

def gen_decl(unit_name, unit, control_name, control):
    fields = [(load_field(field_name, field_details), field_details['doc']) for field_name, field_details in control['fields'].items()] if 'fields' in control else []

    get_args_signature = ', '.join([field.getter_sig() for (field, desc) in fields])
    set_args_signature = ', '.join([field.setter_sig() for (field, desc) in fields])

    return "uvc_error_t uvc_get_{function_name}(uvc_device_handle_t *devh, {args_signature}, enum uvc_req_code req_code);\n".format(**{
        "function_name": control_name,
        "args_signature": get_args_signature
    }) + "uvc_error_t uvc_set_{function_name}(uvc_device_handle_t *devh, {args_signature});\n".format(**{
        "function_name": control_name,
        "args_signature": set_args_signature
    })

def gen_ctrl(unit_name, unit, control_name, control):
    fields = [(load_field(field_name, field_details), field_details['doc']) for field_name, field_details in control['fields'].items()] if 'fields' in control else []

    get_args_signature = ', '.join([field.getter_sig() for (field, desc) in fields])
    set_args_signature = ', '.join([field.setter_sig() for (field, desc) in fields])
    unpack = "\n    ".join([field.unpack() for (field, desc) in fields])
    pack = "\n  ".join([field.pack() for (field, desc) in fields])

    get_gen_doc_raw = None
    set_gen_doc_raw = None

    if 'doc' in control:
        doc = control['doc']

        if isinstance(doc, str):
            get_gen_doc_raw = "\n * ".join(doc.splitlines())
            set_gen_doc_raw = get_gen_doc_raw
        else:
            if 'get' in doc:
                get_gen_doc_raw = "\n * ".join(doc['get'].splitlines())
            if 'set' in doc:
                set_gen_doc_raw = "\n * ".join(doc['set'].splitlines())

    if get_gen_doc_raw is not None:
        get_gen_doc = get_gen_doc_raw.format(gets_sets='Reads')
    else:
        get_gen_doc = '@brief Reads the ' + control['control'] + ' control.'

    if set_gen_doc_raw is not None:
        set_gen_doc = set_gen_doc_raw.format(gets_sets='Sets')
    else:
        set_gen_doc = '@brief Sets the ' + control['control'] + ' control.'

    get_args_doc = "\n * ".join(["@param[out] {0} {1}".format(field.name, desc) for (field, desc) in fields])
    set_args_doc = "\n * ".join(["@param {0} {1}".format(field.name, desc) for (field, desc) in fields])

    control_code = 'UVC_' + unit['control_prefix'] + '_' + control['control'] + '_CONTROL'

    unit_fn = "uvc_get_camera_terminal(devh)->bTerminalID" if (unit_name == "camera_terminal") else ("uvc_get_" + unit_name + "s(devh)->bUnitID")

    return GETTER_TEMPLATE.format(
        unit=unit,
        unit_fn=unit_fn,
        control_name=control_name,
        control_code=control_code,
        control_length=control['length'],
        args_signature=get_args_signature,
        args_doc=get_args_doc,
        gen_doc=get_gen_doc,
        unpack=unpack) + "\n\n" + SETTER_TEMPLATE.format(
            unit=unit,
            unit_fn=unit_fn,
            control_name=control_name,
            control_code=control_code,
            control_length=control['length'],
            args_signature=set_args_signature,
            args_doc=set_args_doc,
            gen_doc=set_gen_doc,
            pack=pack
        )

def export_unit(unit):
    def fmt_doc(doc):
        def wrap_doc_entry(entry):
            if "\n" in entry:
                return literal(entry)
            else:
                return entry

        if isinstance(doc, str):
            return wrap_doc_entry(doc)
        else:
            return OrderedDict([(mode, wrap_doc_entry(text)) for mode, text in doc.items()])

    def fmt_ctrl(control_name, control_details):
        contents = OrderedDict()
        contents['control'] = control_details['control']
        contents['length'] = control_details['length']
        contents['fields'] = control_details['fields']

        if 'doc' in control_details:
            contents['doc'] = fmt_doc(control_details['doc'])

        return (control_name, contents)

    unit_out = OrderedDict()
    unit_out['type'] = unit['type']
    if 'guid' in unit:
        unit_out['guid'] = unit['guid']
    if 'description' in unit:
        unit_out['description'] = unit['description']
    if 'control_prefix' in unit:
        unit_out['control_prefix'] = unit['control_prefix']
    unit_out['controls'] = OrderedDict([fmt_ctrl(ctrl_name, ctrl_details) for ctrl_name, ctrl_details in unit['controls'].items()])
    return unit_out

if __name__ == '__main__':
    try:
        opts, args = getopt.getopt(sys.argv[1:], "hi:", ["help", "input="])
    except getopt.GetoptError as err:
        print(str(err))
        usage()
        sys.exit(-1)

    inputs = []

    for opt, val in opts:
        if opt in ('-h', '--help'):
            usage()
            sys.exit(0)
        elif opt in ('-i', '--input'):
            inputs.append(val)

    mode = None
    for arg in args:
        if arg in ('def', 'decl', 'yaml'):
            if mode is None:
                mode = arg
            else:
                print("Can't specify more than one mode")
                sys.exit(-1)
        else:
            print("Invalid mode '{0}'".format(arg))
            sys.exit(-1)

    def iterunits():
        for input_file in inputs:
            with open(input_file, "r") as fp:
                units = yaml.load(fp)['units']
                for unit_name, unit_details in units.iteritems():
                    yield unit_name, unit_details

    if mode == 'def':
        print("""/* This is an AUTO-GENERATED file! Update it with the output of `ctrl-gen.py def`. */
#include "libuvc/libuvc.h"
#include "libuvc/libuvc_internal.h"

static const int REQ_TYPE_SET = 0x21;
static const int REQ_TYPE_GET = 0xa1;
""")
        fun = gen_ctrl
    elif mode == 'decl':
        fun = gen_decl
    elif mode == 'yaml':
        exported_units = OrderedDict()
        for unit_name, unit_details in iterunits():
            exported_units[unit_name] = export_unit(unit_details)

        yaml.dump({'units': exported_units}, sys.stdout, default_flow_style=False)
        sys.exit(0)

    for unit_name, unit_details in iterunits():
        for control_name, control_details in unit_details['controls'].iteritems():
            code = fun(unit_name, unit_details, control_name, control_details)
            print(code)