"""Entity and player-list updates for the 26.3 play protocol.""" from munch import Munch from minecraft.networking.modern import ( AddEntityPacket, EntityPositionSyncPacket, MoveEntityPositionPacket, MoveEntityPositionRotationPacket, MoveEntityRotationPacket, PlayerInfoRemovePacket, PlayerInfoUpdatePacket, RemoveEntitiesPacket, ) from mosfet.info import mobs class ModernEntityManager: def __init__(self, global_state): self.g = global_state self.g.mobs = self.g.get('mobs') or {} self.g.players = self.g.get('players') or {} self.g.objects = self.g.get('objects') or {} self.g.player_names = self.g.get('player_names') or {} register = self.g.connection.register_packet_listener for packet_type, handler in ( (AddEntityPacket, self.handle_add_entity), (RemoveEntitiesPacket, self.handle_remove_entities), (MoveEntityPositionPacket, self.handle_move), (MoveEntityPositionRotationPacket, self.handle_move), (MoveEntityRotationPacket, self.handle_rotation), (EntityPositionSyncPacket, self.handle_position_sync), (PlayerInfoUpdatePacket, self.handle_player_info), (PlayerInfoRemovePacket, self.handle_player_info_remove), ): register(handler, packet_type) def handle_add_entity(self, packet): name = mobs.MOB_NAMES.get(packet.entity_type_id, 'unknown') common = dict(entity_id=packet.entity_id, x=packet.x, y=packet.y, z=packet.z) if name == 'player': player_uuid = str(packet.entity_uuid) self.g.players[packet.entity_id] = Munch( **common, player_uuid=player_uuid, yaw=packet.yaw, pitch=packet.pitch) elif name == 'item': self.g.objects[packet.entity_id] = Munch( **common, entity_uuid=str(packet.entity_uuid), velocity_x=packet.velocity_x * 8000, velocity_y=packet.velocity_y * 8000, velocity_z=packet.velocity_z * 8000) else: self.g.mobs[packet.entity_id] = Munch( **common, entity_uuid=str(packet.entity_uuid), type=packet.entity_type_id, yaw=packet.yaw, pitch=packet.pitch) def _entity(self, entity_id): for collection in (self.g.players, self.g.mobs, self.g.objects): entity = collection.get(entity_id) if entity is not None: return entity return None def handle_move(self, packet): entity = self._entity(packet.entity_id) if entity is None: return entity.x += packet.delta_x entity.y += packet.delta_y entity.z += packet.delta_z if hasattr(packet, 'yaw'): entity.yaw = packet.yaw entity.pitch = packet.pitch def handle_rotation(self, packet): entity = self._entity(packet.entity_id) if entity is not None: entity.yaw = packet.yaw entity.pitch = packet.pitch def handle_position_sync(self, packet): entity = self._entity(packet.entity_id) if entity is not None: entity.x, entity.y, entity.z = packet.x, packet.y, packet.z entity.yaw = packet.yaw entity.pitch = packet.pitch def handle_remove_entities(self, packet): for entity_id in packet.entity_ids: self.g.players.pop(entity_id, None) self.g.mobs.pop(entity_id, None) self.g.objects.pop(entity_id, None) def handle_player_info(self, packet): for profile_uuid, name in packet.entries: if name: uuid_text = str(profile_uuid) self.g.player_names[uuid_text] = name self.g.player_names[name] = profile_uuid self.g.player_names[name.lower()] = profile_uuid def handle_player_info_remove(self, packet): removed = {str(uuid) for uuid in packet.uuids} for key, value in list(self.g.player_names.items()): if str(value) in removed or str(key) in removed: del self.g.player_names[key]