app.py

Go to the documentation of this file.

# Software License Agreement (BSD License)
#
# Copyright (c) 2011, Willow Garage, Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
#  * Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
#  * Redistributions in binary form must reproduce the above
#    copyright notice, this list of conditions and the following
#    disclaimer in the documentation and/or other materials provided
#    with the distribution.
#  * Neither the name of Willow Garage, Inc. nor the names of its
#    contributors may be used to endorse or promote products derived
#    from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# Revision $Id: app.py 14667 2011-08-12 23:55:04Z pratkanis $

# author: leibs, kwc

import os
import errno
import yaml

import roslaunch
import roslib.names
import rospkg
from rospkg import ResourceNotFound
from .exceptions import AppException, InvalidAppException, NotFoundException, InternalAppException

class Interface(object):
    def __init__(self, subscribed_topics, published_topics):
        self.subscribed_topics = subscribed_topics
        self.published_topics = published_topics

    def __eq__(self, other):
        if not isinstance(other, Interface):
            return False
        return self.subscribed_topics == other.subscribed_topics and \
               self.published_topics == other.published_topics
               
class Client(object):
    __slots__ = ['client_type', 'manager_data', 'app_data']
    def __init__(self, client_type, manager_data, app_data):
        self.client_type = client_type
        self.manager_data = manager_data
        self.app_data = app_data
        
    def as_dict(self):
        return {'client_type': self.client_type, 'manager_data': self.manager_data, 'app_data': self.app_data}
    
    def __eq__(self, other):
        if not isinstance(other, Client):
            return False
        return self.client_type == other.client_type and \
               self.manager_data == other.manager_data and \
               self.app_data == other.app_data

    def __repr__(self):
        return yaml.dump(self.as_dict())
               
class AppDefinition(object):
    __slots__ = ['name', 'display_name', 'description', 'platform',
                 'launch', 'run', 'interface', 'clients', 'icon', 'plugins', 'plugin_order',
                 'timeout', 'allow_parallel']
    def __init__(self, name, display_name, description, platform,
                 interface, clients, launch=None, run=None, icon=None, plugins=None, plugin_order=None,
                 timeout=None, allow_parallel=True):
        self.name = name
        self.display_name = display_name
        self.description = description
        self.platform=platform
        self.interface = interface
        self.clients = clients
        self.launch = launch
        self.run = run
        self.icon = icon
        self.plugins = plugins
        self.plugin_order = plugin_order
        self.timeout = timeout
        self.allow_parallel = allow_parallel

    def __repr__(self):
        d = {}
        for s in self.__slots__:
            if s == 'clients':
                d[s] = [c.as_dict() for c in self.clients]
            else:
                d[s] = getattr(self, s)
        return yaml.dump(d)
    #    return "name: %s\ndisplay: %s\ndescription: %s\nplatform: %s\nlaunch: %s\ninterface: %s\nclients: %s"%(self.name, self.display_name, self.description, self.platform, self.launch, self.interface, self.clients)
    
    def __eq__(self, other):
        if not isinstance(other, AppDefinition):
            return False
        return self.name == other.name and \
               self.display_name == other.display_name and \
               self.description == other.description and \
               self.platform == other.platform and \
               self.interface == other.interface and \
               self.clients == other.clients and \
               self.launch == other.launch and \
               self.run == other.run and \
               self.icon == other.icon
               
def find_resource(resource, rospack=None):
    """
    @return: filepath of resource.  Does not validate if filepath actually exists.
    
    @raise ValueError: if resource is not a valid resource name.
    @raise rospkg.ResourceNotFound: if package referred
        to in resource name cannot be found.
    @raise NotFoundException: if resource does not exist.
    """
    p, a = roslib.names.package_resource_name(resource)
    if not p:
        raise ValueError("Resource is missing package name: %s"%(resource))

    if rospack is None:
        rospack = rospkg.RosPack()
    matches = roslib.packages.find_resource(p, a, rospack=rospack)

    # TODO: convert ValueError to better type for better error messages
    if len(matches) == 1:
        return matches[0]
    elif not matches:
        raise NotFoundException("No resource [%s]"%(resource))        
    else:
        raise ValueError("Multiple resources named [%s]"%(resource))        

def load_Interface_from_file(filename):
    """
    @raise IOError: I/O error reading file (e.g. does not exist)
    @raise InvalidAppException: if app file is invalid
    """
    with open(filename,'r') as f:
        y = yaml.load(f.read())
    y = y or {} #coerce to dict
    try:
        subscribed_topics = y.get('subscribed_topics', {})
        published_topics = y.get('published_topics', {})
    except KeyError:
        raise InvalidAppException("Malformed interface, missing keys")
    return Interface(published_topics=published_topics, subscribed_topics=subscribed_topics)

def _AppDefinition_load_icon_entry(app_data, appfile="UNKNOWN", rospack=None):
    """
    @raise InvalidAppExcetion: if app definition is invalid.
    """

    if rospack is None:
        rospack = rospkg.RosPack()
    # load/validate launch entry
    try:
        icon_resource = app_data.get('icon', '')
        if icon_resource == '':
            return None
        icon_filename = find_resource(icon_resource, rospack=rospack)
        if not icon_filename or not os.path.exists(icon_filename):
            return None
        return icon_filename
    except ValueError as e:
        raise InvalidAppException("Malformed appfile [%s]: bad icon entry: %s"%(appfile, e))
    except NotFoundException:
        # TODO: make this a soft fail?
        raise InvalidAppException("App file [%s] refers to icon that cannot be found"%(appfile))
    except ResourceNotFound as e:
        raise InvalidAppException("App file [%s] refers to package that is not installed: %s"%(appfile, str(e)))

def _AppDefinition_load_launch_entry(app_data, appfile="UNKNOWN", rospack=None):
    """
    @raise InvalidAppExcetion: if app definition is invalid.
    """
    # load/validate launch entry
    if rospack is None:
        rospack = rospkg.RosPack()
    try:
        launch_resource = app_data.get('launch', '')
        if launch_resource == '':
            return None
        launch = find_resource(launch_resource, rospack=rospack)
        if not os.path.exists(launch):
            raise InvalidAppException("Malformed appfile [%s]: refers to launch that does not exist."%(appfile))
        return launch
    except ValueError as e:
        raise InvalidAppException("Malformed appfile [%s]: bad launch entry: %s"%(appfile, e))
    except NotFoundException:
        raise InvalidAppException("App file [%s] refers to launch that is not installed"%(appfile))
    except ResourceNotFound as e:
        raise InvalidAppException("App file [%s] refers to package that is not installed: %s"%(appfile, str(e)))

def _AppDefinition_load_run_args_entry(app_data, appfile="UNKNOWN"):
    """
    @raise InvalidAppException: if app definition is invalid.
    """
    # load/validate launch entry
    try:
        run_args = app_data.get('run_args', '')
        if run_args == '':
            return None
        return run_args
    except ValueError as e:
        raise InvalidAppException("Malformed appfile [%s]: bad run_args entry: %s"%(appfile, e))

def _AppDefinition_load_run_entry(app_data, appfile="UNKNOWN", rospack=None):
    """
    @raise InvalidAppExcetion: if app definition is invalid.
    """
    # load/validate run entry
    if rospack is None:
        rospack = rospkg.RosPack()
    try:
        run_resource = app_data.get('run', '')
        if run_resource == '':
            return None

        # check if file exists
        run = find_resource(run_resource, rospack=rospack)
        if not os.path.exists(run):
            raise InvalidAppException("Malformed appfile [%s]: refers to run that does not exist."%(appfile))
        # create node
        p, a = roslib.names.package_resource_name(run_resource)
        args = _AppDefinition_load_run_args_entry(app_data, appfile)
        node = roslaunch.core.Node(p, a, args=args, output='screen')
        return node
    except ValueError as e:
        raise InvalidAppException("Malformed appfile [%s]: bad run entry: %s"%(appfile, e))
    except NotFoundException:
        raise InvalidAppException("App file [%s] refers to run that is not installed"%(appfile))
    except ResourceNotFound as e:
        raise InvalidAppException("App file [%s] refers to package that is not installed: %s"%(appfile, str(e)))

def _AppDefinition_load_interface_entry(app_data, appfile="UNKNOWN", rospack=None):
    """
    @raise InvalidAppExcetion: if app definition is invalid.
    """
    # load/validate interface entry
    if rospack is None:
        rospack = rospkg.RosPack()
    try:
        return load_Interface_from_file(
            find_resource(app_data['interface'], rospack=rospack))
    except IOError as e:
        if e.errno == errno.ENOENT:
            raise InvalidAppException("Malformed appfile [%s]: refers to interface file that does not exist"%(appfile))
        else:
            raise InvalidAppException("Error with appfile [%s]: cannot read interface file"%(appfile))
    except ValueError:
        raise InvalidAppException("Malformed appfile [%s]: bad interface entry"%(appfile))
    except ResourceNotFound as e:
        raise InvalidAppException("App file [%s] refers to package that is not installed: %s"%(appfile, str(e)))
    
def _AppDefinition_load_clients_entry(app_data, appfile="UNKNOWN"):
    """
    @raise InvalidAppExcetion: if app definition is invalid.
    """
    clients_data = app_data.get('clients', [])
    clients = []
    for c in clients_data:
        for reqd in ['type', 'manager']:
            if not reqd in c:
                raise InvalidAppException("Malformed appfile [%s], missing required key [%s]"%(appfile, reqd))
        client_type = c['type']
        manager_data = c['manager']
        if not type(manager_data) == dict:
            raise InvalidAppException("Malformed appfile [%s]: manager data must be a map"%(appfile))

        app_data = c.get('app', {})
        if not type(app_data) == dict:
            raise InvalidAppException("Malformed appfile [%s]: app data must be a map"%(appfile))

        clients.append(Client(client_type, manager_data, app_data))
    return clients


def _AppDefinition_load_plugins_entry(app_data, appfile="UNKNOWN"):
    """
    @raise InvalidAppException: if app definition is invalid.
    """
    # load/validate launch entry
    try:
        plugins = app_data.get('plugins', '')
        if plugins == '':
            return None
        for plugin in plugins:
            for key in ['launch_args', 'plugin_args', 'start_plugin_args', 'stop_plugin_args']:
                if key in plugin and not type(plugin[key]) == dict:
                    raise InvalidAppException("Malformed appfile [%s]: plugin data(%s) must be a map"%(appfile, key))
        return plugins
    except ValueError as e:
        raise InvalidAppException("Malformed appfile [%s]: bad plugins entry: %s"%(appfile, e))


def _AppDefinition_load_plugin_order_entry(app_data, appfile="UNKNOWN"):
    """
    @raise InvalidAppException: if app definition is invalid.
    """
    # load/validate launch entry
    try:
        plugin_order = app_data.get('plugin_order', '')
        if plugin_order == '':
            return []
        return plugin_order
    except ValueError as e:
        raise InvalidAppException("Malformed appfile [%s]: bad plugin_order entry: %s"%(appfile, e))


def _AppDefinition_load_timeout_entry(app_data, appfile="UNKNOWN"):
    """
    @raise InvalidAppException: if app definition is invalid.
    """
    # load/validate launch entry
    try:
        timeout = app_data.get('timeout', '')
        if timeout == '':
            return None
        return timeout
    except ValueError as e:
        raise InvalidAppException("Malformed appfile [%s]: bad timeout entry: %s"%(appfile, e))


def _AppDefinition_load_allow_parallel_entry(app_data, appfile="UNKNOWN"):
    """
    @raise InvalidAppException: if app definition is invalid.
    """
    # load/validate launch entry
    try:
        allow_parallel = app_data.get('allow_parallel', '')
        if allow_parallel == '':
            return True
        return allow_parallel
    except ValueError as e:
        raise InvalidAppException("Malformed appfile [%s]: bad allow_parallel entry: %s"%(appfile, e))


def load_AppDefinition_from_file(appfile, appname, rospack=None):
    """
    @raise InvalidAppExcetion: if app definition is invalid.
    @raise IOError: I/O error reading appfile (e.g. file does not exist).
    """
    with open(appfile,'r') as f:
        app_data = yaml.load(f.read())
    for reqd in ['interface', 'platform']:
        if not reqd in app_data:
            raise InvalidAppException("Malformed appfile [%s], missing required key [%s]"%(appfile, reqd))
    if not 'launch' in app_data and not 'run' in app_data:
        raise InvalidAppException("Malformed appfile [%s], must have a [launch] or a [run] key"%(appfile))
    if 'launch' in app_data and 'run' in app_data:
        raise InvalidAppException("Malformed appfile [%s], cannot have both [launch] and [run] keys"%(appfile))

    display_name = app_data.get('display', appname)
    description = app_data.get('description', '')
    platform = app_data['platform']

    if rospack is None:
        rospack = rospkg.RosPack()
    launch = _AppDefinition_load_launch_entry(
        app_data, appfile, rospack=rospack)
    run = _AppDefinition_load_run_entry(
        app_data, appfile, rospack=rospack)
    interface = _AppDefinition_load_interface_entry(
        app_data, appfile, rospack=rospack)
    clients = _AppDefinition_load_clients_entry(app_data, appfile)
    icon = _AppDefinition_load_icon_entry(
        app_data, appfile, rospack=rospack)
    plugins = _AppDefinition_load_plugins_entry(app_data, appfile)
    plugin_order = _AppDefinition_load_plugin_order_entry(app_data, appfile)
    timeout = _AppDefinition_load_timeout_entry(app_data, appfile)
    allow_parallel = _AppDefinition_load_allow_parallel_entry(
        app_data, appfile)

    return AppDefinition(appname, display_name, description, platform,
                         interface, clients, launch, run, icon,
                         plugins, plugin_order, timeout, allow_parallel)
    
def load_AppDefinition_by_name(appname, rospack=None):
    """
    @raise InvalidAppExcetion: if app definition is invalid.
    @raise NotFoundExcetion: if app definition is not installed.
    @raise ValueError: if appname is invalid.
    """

    if not appname:
        raise ValueError("app name is empty")

    if rospack is None:
        rospack = rospkg.RosPack()
    try:
        appfile = find_resource(appname + '.app', rospack=rospack)
    except ResourceNotFound as e:
        raise NotFoundException("Cannot locate app file for %s: package is not installed."%(appname))

    try:
        return load_AppDefinition_from_file(appfile, appname, rospack=rospack)
    except IOError as e:
        if e.errno == errno.ENOENT:
            raise NotFoundException("Cannot locate app file for %s."%(appname))
        else:
            raise InternalAppException("I/O error loading AppDefinition file: %s."%(e.errno))