minecraft-bot/protocol/types.py

175 lines
5.0 KiB
Python
Raw Normal View History

from __future__ import division
2020-09-09 02:13:15 +00:00
import struct
from minecraft.networking.types.basic import (
Type, Byte, Short, Integer, Long, Float, Double,
2020-09-23 06:00:28 +00:00
ShortPrefixedByteArray, Boolean, VarInt, TrailingByteArray,
Position, String, UnsignedByte
)
from minecraft.networking.types.utility import Vector
2020-09-09 02:13:15 +00:00
class IntegerPrefixedByteArray(Type):
@staticmethod
def read(file_object):
length = Integer.read(file_object)
return struct.unpack(str(length) + "s", file_object.read(length))[0]
@staticmethod
def send(value, socket):
Integer.send(len(value), socket)
socket.send(value)
TAG_End = 0
TAG_Byte = 1
TAG_Short = 2
TAG_Int = 3
TAG_Long = 4
TAG_Float = 5
TAG_Double = 6
TAG_Byte_Array = 7
TAG_String = 8
TAG_List = 9
TAG_Compound = 10
TAG_Int_Array = 11
TAG_Long_Array = 12
class Nbt(Type):
@staticmethod
def read(file_object):
type_id = Byte.read(file_object)
if type_id != TAG_Compound:
2020-09-23 21:36:18 +00:00
#raise Exception("Invalid NBT header")
return None
name = ShortPrefixedByteArray.read(file_object).decode('utf-8')
a = Nbt.decode_tag(file_object, TAG_Compound)
a['_name'] = name
return a
@staticmethod
def decode_tag(file_object, type_id):
if type_id == TAG_Byte:
return Byte.read(file_object)
elif type_id == TAG_Short:
return Short.read(file_object)
elif type_id == TAG_Int:
return Integer.read(file_object)
elif type_id == TAG_Long:
return Long.read(file_object)
elif type_id == TAG_Float:
return Float.read(file_object)
elif type_id == TAG_Double:
return Double.read(file_object)
elif type_id == TAG_Byte_Array:
return IntegerPrefixedByteArray.read(file_object).decode('utf-8')
elif type_id == TAG_String:
return ShortPrefixedByteArray.read(file_object)
elif type_id == TAG_List:
list_type_id = Byte.read(file_object)
size = Integer.read(file_object)
a = []
for i in range(size):
a.append(Nbt.decode_tag(file_object, list_type_id))
return a
elif type_id == TAG_Compound:
c = { }
child_type_id = Byte.read(file_object)
while child_type_id != TAG_End:
child_name = ShortPrefixedByteArray.read(file_object).decode('utf-8')
c[child_name] = Nbt.decode_tag(file_object, child_type_id)
child_type_id = Byte.read(file_object)
return c
elif type_id == TAG_Int_Array:
size = Integer.read(file_object)
a = []
for i in range(size):
a.append(Integer.read(file_object))
return a
elif type_id == TAG_Long_Array:
size = Integer.read(file_object)
a = []
for i in range(size):
a.append(Long.read(file_object))
return a
else:
raise Exception("Invalid NBT tag type")
@staticmethod
def send(value, socket):
# TODO
pass
class Slot(Type):
def __init__(self, present, item_id, item_count, nbt):
self.present = present
self.item_id = item_id
self.item_count = item_count
self.nbt = nbt
def __str__(self):
return str(self.__dict__)
def __repr__(self):
return 'Slot(present={}, item_id={}, item_count={}, nbt={}'.format(
self.present, self.item_id, self.item_count, self.nbt)
@staticmethod
def read(file_object):
present = Boolean.read(file_object)
2020-09-23 21:36:18 +00:00
if present:
item_id = VarInt.read(file_object)
item_count = Byte.read(file_object)
nbt = Nbt.read(file_object)
else:
item_id = None
item_count = None
nbt = None
return Slot(present, item_id, item_count, nbt)
@staticmethod
def send(value, socket):
Boolean.send(value.present, socket)
VarInt.send(value.item_id, socket)
Byte.send(value.item_count, socket)
Byte.send(0x00, socket)
2020-09-23 06:00:28 +00:00
class Entry(Type):
types = {
0: Byte,
1: VarInt,
2: Float,
3: String,
2020-09-24 22:02:15 +00:00
5: Boolean,
2020-09-23 06:00:28 +00:00
6: Slot,
7: Boolean,
9: Position,
2020-09-24 22:02:15 +00:00
18: VarInt,
2020-09-23 06:00:28 +00:00
}
def __init__(self, index, type, value):
self.index = index
self.type = type
self.value = value
def __str__(self):
return str(self.__dict__)
def __repr__(self):
return 'Entry(index={}, type={}, value={})'.format(
2020-09-23 06:00:28 +00:00
self.index, self.type, self.value)
@staticmethod
def read(file_object, context):
2020-09-23 06:00:28 +00:00
index = UnsignedByte.read(file_object)
if index == 0xff: return None
type = VarInt.read(file_object)
try:
value = Entry.types[type].read(file_object)
except TypeError:
value = Entry.types[type].read_with_context(file_object, context)
2020-09-23 06:00:28 +00:00
except KeyError:
return None
return Entry(index, type, value)