$search

sxserver.py

Go to the documentation of this file.

#!/usr/bin/env python
# Software License Agreement (BSD License)
#
# Based on actionlib axserver.py
# Modified by Christian Dornhege to sxserver.py
#
# Copyright (c) 2009, 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.

"""
usage: %prog /service_name service_type
"""

PKG='fr_tools'

import roslib; roslib.load_manifest(PKG)

from optparse import OptionParser
import wx
import sys
import rospy
import time
import threading
from cStringIO import StringIO
from library import *
from dynamic_service import DynamicService

SUCCEED = 0
FAIL = 1

class SXServerApp(wx.App):
    def __init__(self, service_type, service_name):
        self.service_type = service_type
        wx.App.__init__(self)

        self.server = rospy.Service(service_name, self.service_type.service, self.execute)
        self.condition = threading.Condition()
        self.response_msg = None
        self.execute_type = None

    def set_request(self, request):
        if request is None:
            self.status_bg.SetBackgroundColour(wx.Colour(200,0,0))
            self.status.SetLabel("Waiting For Request...")
            self.succeed.Disable()
            self.fail.Disable()            

            self.request.SetValue("")

        else:
            self.status_bg.SetBackgroundColour(wx.Colour(0,200,0))
            self.status.SetLabel("Received Request.  Send succeed, or fail.")
            self.succeed.Enable()
            self.fail.Enable()

            try:
                self.request.SetValue(to_yaml(request))
            except UnicodeDecodeError:
                self.request.SetValue("Cannot display request due to unprintable characters")

    def execute(self, request):

        wx.CallAfter(self.set_request, request)

        self.condition.acquire()

        self.response_msg = None
        self.execute_type = None
        
        while self.execute_type is None:

            self.response_msg = None
            self.execute_type = None
            
            while self.execute_type is None:
                self.condition.wait()

        retVal = False
        if self.execute_type == SUCCEED:
            retVal = True

        if self.execute_type == FAIL:
            retVal = False

        wx.CallAfter(self.set_request, None)

        self.condition.release()
        if retVal:
            return self.response_msg
        else:
            return None

    def on_succeed(self, event):
        self.condition.acquire()
        
        try:
            self.response_msg = yaml_msg_str(self.service_type.response, self.response.GetValue())
            buff = StringIO()
            self.response_msg.serialize(buff)
            
            self.execute_type = SUCCEED
            self.condition.notify()
        except roslib.message.SerializationError, e:
            self.response_msg = None
            wx.MessageBox(str(e), "Error serializing response", wx.OK)

        self.condition.release()

    def on_fail(self, event):
        self.condition.acquire()

        self.execute_type = FAIL
        self.condition.notify()

        self.condition.release()

    def OnInit(self):
        self.frame = wx.Frame(None, -1, self.service_type.name + ' Standin')

        self.sz = wx.BoxSizer(wx.VERTICAL)

        tmp_response = self.service_type.response()

        self.request = wx.TextCtrl(self.frame, -1, style=(wx.TE_MULTILINE | wx.TE_READONLY))
        self.request_st_bx = wx.StaticBox(self.frame, -1, "Request")
        self.request_st = wx.StaticBoxSizer(self.request_st_bx, wx.VERTICAL)
        self.request_st.Add(self.request, 1, wx.EXPAND)
        
        self.response = wx.TextCtrl(self.frame, -1, style=wx.TE_MULTILINE)
        self.response.SetValue(to_yaml(tmp_response))
        self.response_st_bx = wx.StaticBox(self.frame, -1, "Response")
        self.response_st = wx.StaticBoxSizer(self.response_st_bx, wx.VERTICAL)
        self.response_st.Add(self.response, 1, wx.EXPAND)

        self.succeed = wx.Button(self.frame, -1, label="SUCCEED")
        self.succeed.Bind(wx.EVT_BUTTON, self.on_succeed)

        self.fail    = wx.Button(self.frame, -1, label="FAIL")
        self.fail.Bind(wx.EVT_BUTTON, self.on_fail)

        self.status_bg = wx.Panel(self.frame, -1)
        self.status_bg.SetBackgroundColour(wx.Colour(200,0,0))
        self.status = wx.StaticText(self.status_bg, -1, label="Waiting For Request...")

        self.sz.Add(self.request_st, 1, wx.EXPAND)
        self.sz.Add(self.response_st, 1, wx.EXPAND)
        self.sz.Add(self.succeed, 0, wx.EXPAND)
        self.sz.Add(self.fail, 0, wx.EXPAND)
        self.sz.Add(self.status_bg, 0, wx.EXPAND)

        self.frame.SetSizer(self.sz)

        self.set_request(None)

        self.sz.Layout()
        self.frame.Show()

        return True

if __name__ == '__main__':
    rospy.init_node('sxserver', anonymous=True)

    parser = OptionParser(__doc__.strip())

    (options, args) = parser.parse_args(rospy.myargv())

    if (len(args) != 3):
        parser.error("You must specify the service name and type. Eg: ./sxserver.py my_service my_msgs/Test")

    service = DynamicService(args[2])

    app = SXServerApp(service, args[1])
    app.MainLoop()
    rospy.signal_shutdown('GUI shutdown')

---

fr_tools
Author(s): Armin Hornung, Daniel Maier
autogenerated on Tue Mar 5 12:00:49 2013