converter.py

Go to the documentation of this file.

import genmsg
import genmsg.msgs
from genmsg import MsgSpec, SrvSpec
from genpy.generator import compute_pkg_type
 
from ._compat import lru_cache
from ._typing import TYPE_CHECKING
from .stub_element import (
    AliasElement,
    ClassElement,
    ClassMethodElement,
    CommentElement,
    EmptyLinesElement,
    FieldElement,
    ImportsElement,
    ParameterElement,
)
 
if TYPE_CHECKING:
    from typing import Dict, List, Sequence, Tuple
 
 
_GENMSG_PRIMITIVES = {
    "byte": "byte",
    "char": "str",
    "int8": "int",
    "int16": "int",
    "int32": "int",
    "int64": "int",
    "uint8": "int",
    "uint16": "int",
    "uint32": "int",
    "uint64": "int",
    "float32": "float",
    "float64": "float",
    "bool": "bool",
    "string": "str",
}
 
_GENPY_DEFINED = {
    genmsg.HEADER: ("std_msgs.msg._Header.Header", "std_msgs.msg._Header"),
    genmsg.TIME: ("genpy.Time", "genpy"),
    genmsg.DURATION: ("genpy.Duration", "genpy"),
}  # type: Dict[str, Tuple[str, str]]
 
 
@lru_cache()
def _get_genmsg_type(first_party_package, field_type, imports):
    # type: (str, str, ImportsElement) -> str
 
    """Get the corresponding python type for the given ROS type.
 
    .. warning::
        As this method is decorated by lru_cache, make sure that
        you are using another instance of ImportsElement when
        calling this method in a different context.
        (e.g. creating a different module)
    """
    primitive = _GENMSG_PRIMITIVES.get(field_type)
    if primitive is not None:
        return primitive
 
    if field_type.endswith("]"):
        base_field_type, is_array, array_len = genmsg.msgs.parse_type(field_type)
        assert is_array
        # TODO: Handle array_len (use typing.Annotated?)
 
        if base_field_type == "uint8":
            # Special case for uint8[]
            return "bytes"
 
        base_type = _get_genmsg_type(first_party_package, base_field_type, imports)
        return "typing.List[{}]".format(base_type)
 
    special = _GENPY_DEFINED.get(field_type)
    if special is not None:
        full_name, module = special
        imports.add_third_party_module(module, None)
        return full_name
 
    package, type_ = compute_pkg_type(first_party_package, field_type)
    module = "{}.msg".format(package)
    imports.add_third_party_module(module, None)
 
    return "{}.{}".format(module, type_)
 
 
def _convert_message_fields(first_party_package, spec, imports):
    # type: (str, MsgSpec, ImportsElement) -> List[FieldElement]
    fields = []  # type: List[FieldElement]
    for field in spec.fields():
        type_, name = field
        fields.append(
            FieldElement(
                name,
                _get_genmsg_type(first_party_package, type_, imports),
            )
        )
 
    return fields
 
 
def _convert_message_constants(first_party_package, spec, imports):
    # type: (str, MsgSpec, ImportsElement) -> List[FieldElement]
    constants = []  # type: List[FieldElement]
    for constant in spec.constants:
        constants.append(
            FieldElement(
                constant.name,
                _get_genmsg_type(first_party_package, constant.type, imports),
            )
        )
 
    return constants
 
 
def convert_message_class(first_party_package, spec, imports):
    # type: (str, MsgSpec, ImportsElement) -> ClassElement
    imports.add_third_party_module("genpy", None)
    imports.add_system_module("typing", None)
    imports.add_system_module("types", None)
 
    msgclass = ClassElement(spec.short_name, "genpy.Message")
 
    # Add private fields
    msgclass.add(FieldElement("_md5sum", "str"))
    msgclass.add(FieldElement("_type", "str"))
    msgclass.add(FieldElement("_has_header", "bool"))
    msgclass.add(FieldElement("_full_text", "str"))
    msgclass.add(FieldElement("__slots__", "typing.List[str]"))
    msgclass.add(FieldElement("_slot_types", "typing.List[str]"))
 
    message_constants = _convert_message_constants(first_party_package, spec, imports)
    if len(message_constants) > 0:
        msgclass.add(EmptyLinesElement())
        msgclass.add(CommentElement("Constants"))
        for c in message_constants:
            msgclass.add(c)
 
    message_fields = _convert_message_fields(first_party_package, spec, imports)
    if len(message_fields) > 0:
        msgclass.add(EmptyLinesElement())
        msgclass.add(CommentElement("Fields"))
        for f in message_fields:
            msgclass.add(f)
 
    msgclass.add(EmptyLinesElement())
 
    # Add __init__ method that accepts message fields as the parameter
    init_method = ClassMethodElement("__init__", "None")
    for field in message_fields:
        init_method.add_parameter(
            ParameterElement(field.name, field.type, has_default=True)
        )
 
    # NOTE: Emit *args and **kwds in order to align with the actual implementation
    init_method.add_parameter(ParameterElement("*args", "typing.Any"))
    init_method.add_parameter(ParameterElement("**kwds", "typing.Any"))
 
    msgclass.add(init_method)
 
    # Add private methods except pattern methods like `_get_struct_I`
    msgclass.add(ClassMethodElement("_get_types", "typing.List[str]"))
 
    # Add public methods
    msgclass.add(
        ClassMethodElement(
            "serialize", "None", [ParameterElement("buff", "typing.BinaryIO")]
        )
    )
    msgclass.add(
        ClassMethodElement(
            "deserialize", spec.short_name, [ParameterElement("str", "bytes")]
        )
    )
    msgclass.add(
        ClassMethodElement(
            "serialize_numpy",
            "None",
            [
                ParameterElement("buff", "typing.BinaryIO"),
                ParameterElement("numpy", "types.ModuleType"),
            ],
        )
    )
    msgclass.add(
        ClassMethodElement(
            "deserialize_numpy",
            spec.short_name,
            [
                ParameterElement("str", "bytes"),
                ParameterElement("numpy", "types.ModuleType"),
            ],
        )
    )
 
    return msgclass
 
 
def convert_service_class(spec, request_class, response_class):
    # type: (SrvSpec, ClassElement, ClassElement) -> ClassElement
    srvclass = ClassElement(spec.short_name, "object")
    srvclass.add(FieldElement("_type", "str"))
    srvclass.add(FieldElement("_md5sum", "str"))
    srvclass.add(AliasElement("_request_class", request_class.name))
    srvclass.add(AliasElement("_response_class", response_class.name))
 
    return srvclass
 
 
def convert_genpy_init(modules):
    # type: (Sequence[str]) -> ImportsElement
    imports = ImportsElement()
    for m in modules:
        imports.add_third_party_module("._{}".format(m), "*")
 
    return imports