"""LCM type definitions This file automatically generated by lcm. DO NOT MODIFY BY HAND!!!! """ from io import BytesIO import struct import oi class msg_t(object): __slots__ = ["pub_id", "pub_timestamp", "pub_sequence", "num_bools", "bool_list", "num_integers", "integer_list", "num_reals", "real_list", "num_strings", "string_list", "num_blobs", "blob_list", "num_arrays", "array_list"] __typenames__ = ["string", "double", "int64_t", "int16_t", "oi.bool_item_t", "int16_t", "oi.integer_item_t", "int16_t", "oi.real_item_t", "int16_t", "oi.string_item_t", "int16_t", "oi.blob_item_t", "int16_t", "oi.array_item_t"] __dimensions__ = [None, None, None, None, ["num_bools"], None, ["num_integers"], None, ["num_reals"], None, ["num_strings"], None, ["num_blobs"], None, ["num_arrays"]] def __init__(self): self.pub_id = "" """ publisher ID LCM Type: string """ self.pub_timestamp = 0.0 """ publish time in seconds since epoch LCM Type: double """ self.pub_sequence = 0 """ publish sequence number LCM Type: int64_t """ self.num_bools = 0 """ bool type LCM Type: int16_t """ self.bool_list = [] """ LCM Type: oi.bool_item_t[num_bools] """ self.num_integers = 0 """ integer type LCM Type: int16_t """ self.integer_list = [] """ LCM Type: oi.integer_item_t[num_integers] """ self.num_reals = 0 """ real type LCM Type: int16_t """ self.real_list = [] """ LCM Type: oi.real_item_t[num_reals] """ self.num_strings = 0 """ string type LCM Type: int16_t """ self.string_list = [] """ LCM Type: oi.string_item_t[num_strings] """ self.num_blobs = 0 """ blob type LCM Type: int16_t """ self.blob_list = [] """ LCM Type: oi.blob_item_t[num_blobs] """ self.num_arrays = 0 """ array type LCM Type: int16_t """ self.array_list = [] """ LCM Type: oi.array_item_t[num_arrays] """ def encode(self): buf = BytesIO() buf.write(msg_t._get_packed_fingerprint()) self._encode_one(buf) return buf.getvalue() def _encode_one(self, buf): __pub_id_encoded = self.pub_id.encode('utf-8') buf.write(struct.pack('>I', len(__pub_id_encoded)+1)) buf.write(__pub_id_encoded) buf.write(b"\0") buf.write(struct.pack(">dqh", self.pub_timestamp, self.pub_sequence, self.num_bools)) for i0 in range(self.num_bools): assert self.bool_list[i0]._get_packed_fingerprint() == oi.bool_item_t._get_packed_fingerprint() self.bool_list[i0]._encode_one(buf) buf.write(struct.pack(">h", self.num_integers)) for i0 in range(self.num_integers): assert self.integer_list[i0]._get_packed_fingerprint() == oi.integer_item_t._get_packed_fingerprint() self.integer_list[i0]._encode_one(buf) buf.write(struct.pack(">h", self.num_reals)) for i0 in range(self.num_reals): assert self.real_list[i0]._get_packed_fingerprint() == oi.real_item_t._get_packed_fingerprint() self.real_list[i0]._encode_one(buf) buf.write(struct.pack(">h", self.num_strings)) for i0 in range(self.num_strings): assert self.string_list[i0]._get_packed_fingerprint() == oi.string_item_t._get_packed_fingerprint() self.string_list[i0]._encode_one(buf) buf.write(struct.pack(">h", self.num_blobs)) for i0 in range(self.num_blobs): assert self.blob_list[i0]._get_packed_fingerprint() == oi.blob_item_t._get_packed_fingerprint() self.blob_list[i0]._encode_one(buf) buf.write(struct.pack(">h", self.num_arrays)) for i0 in range(self.num_arrays): assert self.array_list[i0]._get_packed_fingerprint() == oi.array_item_t._get_packed_fingerprint() self.array_list[i0]._encode_one(buf) @staticmethod def decode(data: bytes): if hasattr(data, 'read'): buf = data else: buf = BytesIO(data) if buf.read(8) != msg_t._get_packed_fingerprint(): raise ValueError("Decode error") return msg_t._decode_one(buf) @staticmethod def _decode_one(buf): self = msg_t() __pub_id_len = struct.unpack('>I', buf.read(4))[0] self.pub_id = buf.read(__pub_id_len)[:-1].decode('utf-8', 'replace') self.pub_timestamp, self.pub_sequence, self.num_bools = struct.unpack(">dqh", buf.read(18)) self.bool_list = [] for i0 in range(self.num_bools): self.bool_list.append(oi.bool_item_t._decode_one(buf)) self.num_integers = struct.unpack(">h", buf.read(2))[0] self.integer_list = [] for i0 in range(self.num_integers): self.integer_list.append(oi.integer_item_t._decode_one(buf)) self.num_reals = struct.unpack(">h", buf.read(2))[0] self.real_list = [] for i0 in range(self.num_reals): self.real_list.append(oi.real_item_t._decode_one(buf)) self.num_strings = struct.unpack(">h", buf.read(2))[0] self.string_list = [] for i0 in range(self.num_strings): self.string_list.append(oi.string_item_t._decode_one(buf)) self.num_blobs = struct.unpack(">h", buf.read(2))[0] self.blob_list = [] for i0 in range(self.num_blobs): self.blob_list.append(oi.blob_item_t._decode_one(buf)) self.num_arrays = struct.unpack(">h", buf.read(2))[0] self.array_list = [] for i0 in range(self.num_arrays): self.array_list.append(oi.array_item_t._decode_one(buf)) return self @staticmethod def _get_hash_recursive(parents): if msg_t in parents: return 0 newparents = parents + [msg_t] tmphash = (0xad15b7e81487bd16+ oi.bool_item_t._get_hash_recursive(newparents)+ oi.integer_item_t._get_hash_recursive(newparents)+ oi.real_item_t._get_hash_recursive(newparents)+ oi.string_item_t._get_hash_recursive(newparents)+ oi.blob_item_t._get_hash_recursive(newparents)+ oi.array_item_t._get_hash_recursive(newparents)) & 0xffffffffffffffff tmphash = (((tmphash<<1)&0xffffffffffffffff) + (tmphash>>63)) & 0xffffffffffffffff return tmphash _packed_fingerprint = None @staticmethod def _get_packed_fingerprint(): if msg_t._packed_fingerprint is None: msg_t._packed_fingerprint = struct.pack(">Q", msg_t._get_hash_recursive([])) return msg_t._packed_fingerprint def get_hash(self): """Get the LCM hash of the struct""" return struct.unpack(">Q", msg_t._get_packed_fingerprint())[0]