Source code for machine_common_sense.serializer

import io
from abc import ABCMeta, abstractmethod

import msgpack
import numpy as np
import PIL.Image as Image

from .goal_metadata import GoalMetadata
from .object_metadata import ObjectMetadata
from .step_metadata import StepMetadata


class ISerializer:
    __metaclass__ = ABCMeta

    @classmethod
    def version(cls):
        return "0.1.0"

    @staticmethod
    @abstractmethod
    def serialize(unpacked_object):
        raise NotImplementedError

    @staticmethod
    @abstractmethod
    def deserialize(packed_object):
        raise NotImplementedError

    @staticmethod
    def image_to_bytes(image):
        """Converts PIL image to bytes."""
        byte_io = io.BytesIO()
        image.save(byte_io, format='PNG')
        return byte_io.getvalue()

    @staticmethod
    def bytes_to_image(bytes_image):
        """Converts bytes to PIL image."""
        return Image.open(io.BytesIO(bytes_image))


[docs]class SerializerMsgPack(ISerializer): """Serializer to (de)serialize StepMetadata into/from MsgPack format.""" @staticmethod def _ext_pack(x): """Hook to serialize MCS Step Metadata as MsgPack, e.g. serialized = msgpack.packb(output, default=ext_pack, strict_types=True) """ if isinstance(x, StepMetadata): # TODO In future can investigate reflection for automating field # extraction, but notice that strict types flag is highly # dependent on order. Something like this might be a start: # [field for field in dir(x) if not callable(getattr(x, field)) # and not field.startswith("__")] return msgpack.ExtType( 1, msgpack.packb([ x.action_list, x.camera_aspect_ratio, x.camera_clipping_planes, x.camera_field_of_view, x.camera_height, x.depth_map_list, x.goal, x.head_tilt, x.image_list, x.object_list, x.object_mask_list, x.position, x.return_status, x.reward, x.rotation, x.step_number, x.structural_object_list ], default=SerializerMsgPack._ext_pack, strict_types=True)) elif isinstance(x, tuple): return msgpack.ExtType( 2, msgpack.packb([x[0], x[1]], default=SerializerMsgPack._ext_pack, strict_types=True)) elif isinstance(x, Image.Image): return msgpack.ExtType( 3, msgpack.packb(SerializerMsgPack.image_to_bytes(x), default=SerializerMsgPack._ext_pack, strict_types=True)) elif isinstance(x, GoalMetadata): return msgpack.ExtType( 4, msgpack.packb([ x.action_list, x.category, x.description, x.habituation_total, x.last_preview_phase_step, x.last_step, x.metadata ], default=SerializerMsgPack._ext_pack, strict_types=True)) elif isinstance(x, ObjectMetadata): return msgpack.ExtType( 5, msgpack.packb([ x.uuid, x.dimensions, x.direction, x.distance, x.distance_in_steps, x.distance_in_world, x.held, x.mass, x.material_list, x.position, x.rotation, x.segment_color, x.shape, x.state_list, x.texture_color_list, x.visible ], default=SerializerMsgPack._ext_pack, strict_types=True)) elif isinstance(x, np.ndarray): return msgpack.ExtType( 6, msgpack.packb(x.tolist(), default=SerializerMsgPack._ext_pack, strict_types=True)) return x @staticmethod def _ext_unpack(code, data): """ Hook to deserialize MCS Step Metadata from MsgPack, e.g. deserialized = msgpack.unpackb(packed_bytes, ext_hook=ext_unpack) """ if code == 1: action_list, camera_aspect_ratio, camera_clipping_planes, \ camera_field_of_view, camera_height, depth_map_list, goal, \ head_tilt, image_list, object_list, object_mask_list, \ position, return_status, reward, rotation, step_number, \ structural_object_list = \ msgpack.unpackb(data, ext_hook=SerializerMsgPack._ext_unpack) return StepMetadata(action_list=action_list, camera_aspect_ratio=camera_aspect_ratio, camera_clipping_planes=camera_clipping_planes, camera_field_of_view=camera_field_of_view, camera_height=camera_height, depth_map_list=depth_map_list, goal=goal, head_tilt=head_tilt, image_list=image_list, object_list=object_list, object_mask_list=object_mask_list, position=position, return_status=return_status, reward=reward, rotation=rotation, step_number=step_number, structural_object_list=structural_object_list) elif code == 2: x0, x1 = msgpack.unpackb(data, ext_hook=SerializerMsgPack._ext_unpack) return x0, x1 elif code == 3: x = msgpack.unpackb(data, ext_hook=SerializerMsgPack._ext_unpack) return SerializerMsgPack.bytes_to_image(x) elif code == 4: action_list, category, description, habituation_total, \ last_preview_phase_step, last_step, metadata = \ msgpack.unpackb(data, ext_hook=SerializerMsgPack._ext_unpack) return GoalMetadata(action_list, category, description, habituation_total, last_preview_phase_step, last_step, metadata) elif code == 5: uuid, dimensions, direction, distance, distance_in_steps, \ distance_in_world, held, mass, material_list, position, \ rotation, segment_color, shape, state_list, \ texture_color_list, visible = msgpack.unpackb( data, ext_hook=SerializerMsgPack._ext_unpack) return ObjectMetadata(uuid, dimensions, direction, distance, distance_in_steps, distance_in_world, held, mass, material_list, position, rotation, segment_color, shape, state_list, texture_color_list, visible) elif code == 6: x = msgpack.unpackb(data, ext_hook=SerializerMsgPack._ext_unpack) return np.asarray(x) return msgpack.ExtType(code, data)
[docs] @staticmethod def serialize(step_metadata: StepMetadata): """ Serializes step metadata into MsgPack. You can use .. code-block:: python object_to_persist = { 'payload': step_metadata, 'additional_info': 'info' } to add extra data. Args: step_metadata: MCS step metadata output Returns: Serialized version of step metadata in MsgPack format. """ return msgpack.packb(step_metadata, default=SerializerMsgPack._ext_pack, strict_types=True)
@staticmethod def deserialize(packed_step_metadata): return msgpack.unpackb(packed_step_metadata, ext_hook=SerializerMsgPack._ext_unpack)