def _load_msg1()

in src/rosbridge/zmq_serialize.py [0:0]

• 35 lines of code
• 16 McCabe index (conditional complexity)

def _load_msg1(o, parts):
    if isinstance(o, dict):
        t = o.get('__type', None)
        if t is not None:
            if t == 'ndarray':
                if o['dtype'] == 'object':
                    return np.array(o['flat']).reshape(o['shape'])
                else:
                    return np.frombuffer(parts[o['partno']], dtype=o['dtype']).reshape(o['shape'])
            elif t == 'Box':
                return gym.spaces.Box(low=_load_msg1(o['low'], parts), high=_load_msg1(o['high'], parts))
            elif t == 'Tuple':
                return gym.spaces.Tuple(spaces=_load_msg1(o['spaces'], parts))
            elif t == 'Discrete':
                return gym.spaces.Discrete(n=_load_msg1(o['n'], parts))
            elif t == 'tuple':
                return tuple(_load_msg1(o['elems'], parts))
            elif t == 'number':
                if o['value'] == '+inf':
                    return np.inf
                elif o['value'] == '-inf':
                    return -np.inf
                elif o['value'] == 'nan':
                    return np.nan
                else:
                    raise Exception('Unknown value %s' % o['value'])
            else:
                logger.warn('Unimplemented object to reconstruct %s', t)
                return o
        else:
            return dict([(k, _load_msg1(v, parts)) for k,v in o.items()])
    elif isinstance(o, list):
        return [_load_msg1(v, parts) for v in o]
    else:
        return o