<s> import time <EOL> from datetime import datetime <EOL> from functools import partial <EOL> try : <EOL> from cStringIO import StringIO , InputType , OutputType <EOL> from StringIO import StringIO as pyStringIO <EOL> def _checkIsStringIO ( obj ) : <EOL> return isinstance ( obj , ( InputType , OutputType , pyStringIO ) ) <EOL> except ImportError : <EOL> from StringIO import StringIO <EOL> def _checkIsStringIO ( obj ) : <EOL> return isinstance ( obj , StringIO ) <EOL> try : <EOL> from genmsg . names import package_resource_name <EOL> from genpy . message import Message <EOL> from rospy . rostime import Duration , Time <EOL> except ImportError : <EOL> print ( '<STR_LIT>' ) <EOL> print ( '<STR_LIT>' ) <EOL> exit ( <NUM_LIT:1> ) <EOL> from zope . interface import implements <EOL> from rce . util . error import InternalError <EOL> from rce . util . interface import verifyClass <EOL> from rce . util . converters . interfaces import ICustomROSConverter <EOL> def _stringify ( obj ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( obj , unicode ) : <EOL> return obj . encode ( '<STR_LIT:utf-8>' ) <EOL> elif isinstance ( obj , str ) : <EOL> return obj <EOL> else : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> class _DurationConverter ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ICustomROSConverter ) <EOL> def decode ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return Duration . from_sec ( float ( data ) ) <EOL> def encode ( self , rosMsg ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return ( rosMsg . to_sec ( ) , { } ) <EOL> except AttributeError : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> class _TimeConverter ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ICustomROSConverter ) <EOL> def decode ( self , data ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:+>' in data : <EOL> data = data [ : data . index ( '<STR_LIT:+>' ) ] <EOL> try : <EOL> dt = datetime ( year = int ( data [ <NUM_LIT:0> : <NUM_LIT:4> ] ) , month = int ( data [ <NUM_LIT:5> : <NUM_LIT:7> ] ) , <EOL> day = int ( data [ <NUM_LIT:8> : <NUM_LIT:10> ] ) , hour = int ( data [ <NUM_LIT:11> : <NUM_LIT> ] ) , <EOL> minute = int ( data [ <NUM_LIT> : <NUM_LIT:16> ] ) , second = int ( data [ <NUM_LIT> : <NUM_LIT> ] ) , <EOL> microsecond = int ( data [ <NUM_LIT:20> : ] ) ) <EOL> except ValueError : <EOL> return Time ( ) <EOL> return Time . from_sec ( time . mktime ( dt . timetuple ( ) ) ) <EOL> def encode ( self , rosMsg ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> dt = datetime . fromtimestamp ( rosMsg . to_sec ( ) ) <EOL> except AttributeError : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> return ( dt . isoformat ( ) , { } ) <EOL> verifyClass ( ICustomROSConverter , _DurationConverter ) <EOL> verifyClass ( ICustomROSConverter , _TimeConverter ) <EOL> class Converter ( object ) : <EOL> """<STR_LIT>""" <EOL> _BASE_TYPES = { '<STR_LIT:bool>' : bool , <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT>' : int , <EOL> '<STR_LIT>' : long , <EOL> '<STR_LIT>' : long , <EOL> '<STR_LIT>' : float , <EOL> '<STR_LIT>' : float , <EOL> '<STR_LIT:string>' : str } <EOL> _SPECIAL_TYPES = { '<STR_LIT:time>' : _TimeConverter , <EOL> '<STR_LIT>' : _DurationConverter } <EOL> def __init__ ( self , loader ) : <EOL> """<STR_LIT>""" <EOL> self . _loader = loader <EOL> self . _customTypes = { } <EOL> def addCustomConverter ( self , converter ) : <EOL> """<STR_LIT>""" <EOL> verifyClass ( ICustomROSConverter , converter ) <EOL> if converter . MESSAGE_TYPE in self . _customTypes : <EOL> raise InternalError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( <EOL> converter . MESSAGE_TYPE ) ) <EOL> try : <EOL> pkg , name = package_resource_name ( converter . MESSAGE_TYPE ) <EOL> except ValueError : <EOL> raise InternalError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . _customTypes [ converter . MESSAGE_TYPE ] = ( converter , <EOL> self . _loader . loadMsg ( pkg , name ) ) <EOL> def removeCustomConverter ( self , msgType ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> del self . _customTypes [ msgType ] <EOL> except KeyError : <EOL> InternalError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _encode ( self , rosMsg ) : <EOL> """<STR_LIT>""" <EOL> data = { } <EOL> for ( slotName , slotType ) in zip ( rosMsg . __slots__ , rosMsg . _slot_types ) : <EOL> if '<STR_LIT>' == slotType [ - <NUM_LIT:2> : ] : <EOL> listBool = True <EOL> slotType = slotType [ : - <NUM_LIT:2> ] <EOL> else : <EOL> listBool = False <EOL> if slotType in self . _BASE_TYPES : <EOL> convFunc = self . _BASE_TYPES [ slotType ] <EOL> elif slotType in self . _SPECIAL_TYPES : <EOL> convFunc = self . _SPECIAL_TYPES [ slotType ] ( ) . encode <EOL> elif slotType in self . _customTypes : <EOL> convFunc = self . _customTypes [ slotType ] [ <NUM_LIT:0> ] ( ) . encode <EOL> else : <EOL> convFunc = self . _encode <EOL> if listBool : <EOL> convFunc = partial ( map , convFunc ) <EOL> try : <EOL> data [ slotName ] = convFunc ( getattr ( rosMsg , slotName ) ) <EOL> except ValueError as e : <EOL> raise ValueError ( '<STR_LIT>' . format ( <EOL> rosMsg . __class__ . __name__ , slotName , e ) ) <EOL> return data <EOL> def encode ( self , rosMsg ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( rosMsg , Message ) : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> for converter , cls in self . _customTypes . itervalues ( ) : <EOL> if isinstance ( rosMsg , cls ) : <EOL> return converter ( ) . encode ( rosMsg ) <EOL> return self . _encode ( rosMsg ) <EOL> def _decode ( self , msgCls , data ) : <EOL> """<STR_LIT>""" <EOL> rosMsg = msgCls ( ) <EOL> for ( slotName , slotType ) in zip ( rosMsg . __slots__ , rosMsg . _slot_types ) : <EOL> if slotName not in data : <EOL> continue <EOL> if '<STR_LIT>' == slotType [ - <NUM_LIT:2> : ] : <EOL> listBool = True <EOL> slotType = slotType [ : - <NUM_LIT:2> ] <EOL> else : <EOL> listBool = False <EOL> field = data [ slotName ] <EOL> if listBool and not isinstance ( field , ( list , tuple ) ) : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if slotType == '<STR_LIT:string>' : <EOL> convFunc = _stringify <EOL> elif slotType in self . _BASE_TYPES : <EOL> convFunc = self . _BASE_TYPES [ slotType ] <EOL> elif slotType in self . _SPECIAL_TYPES : <EOL> convFunc = self . _SPECIAL_TYPES [ slotType ] ( ) . decode <EOL> elif slotType in self . _customTypes and _checkIsStringIO ( field ) : <EOL> convFunc = self . _customTypes [ slotType ] [ <NUM_LIT:0> ] ( ) . decode <EOL> else : <EOL> convFunc = partial ( self . _decode , <EOL> self . _loader . loadMsg ( * slotType . split ( '<STR_LIT:/>' ) ) ) <EOL> if listBool : <EOL> convFunc = partial ( map , convFunc ) <EOL> setattr ( rosMsg , slotName , convFunc ( field ) ) <EOL> return rosMsg <EOL> def decode ( self , msgCls , data ) : <EOL> """<STR_LIT>""" <EOL> if _checkIsStringIO ( data ) : <EOL> for converter , cls in self . _customTypes . itervalues ( ) : <EOL> if msgCls == cls : <EOL> return converter ( ) . decode ( msgCls , data ) <EOL> return self . _decode ( msgCls , data ) </s>
