tanner/minecraft-bot
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Detect monsters and flee to safety

Browse Source
This commit is contained in:
Tanner Collin
2020-12-04 02:49:22 +00:00
parent 5a7a2f0113
commit b8952db742
7 changed files with 475 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
game.py
+272 -176
View File
@@ -2,6 +2,7 @@ import re
import time
import importlib
import random
import functools
from math import hypot
from itertools import count
from munch import Munch
@@ -19,6 +20,7 @@ from protocol.packets import (
ClickWindowPacket, CloseWindowPacket, ServerWindowConfirmationPacket,
ClientWindowConfirmationPacket, EntityMetadataPacket,
SpawnLivingEntityPacket, EntityPositionRotationPacket, DestroyEntitiesPacket,
EntityActionPacket,
)
from protocol.types import Slot
@@ -33,6 +35,8 @@ import items
importlib.reload(items)
import data
importlib.reload(data)
import mobs
importlib.reload(mobs)
class MCWorld:
def __init__(self, global_state):
@@ -41,6 +45,13 @@ class MCWorld:
def block_at(self, x, y, z):
return self.g.chunks.get_block_at(x, y, z)
def check_air_column(self, pos, distance):
for i in range(distance):
check = utils.padd(pos, (0, i, 0))
if self.block_at(*check) not in blocks.NON_SOLID_IDS:
return False
return True
def find_blocks_3d(self, center, block_ids, distance=0, y_limit=0):
for offset in utils.search_3d(distance, y_limit):
check = utils.padd(center, offset)
@@ -189,10 +200,9 @@ class MCWorld:
return safe_sand
def check_sand_slice(self, center):
# checks if a 5x5x1 slice has diggable sand in it
# checks if a 5x5x1 slice has sand in it
for i in range(9):
s = utils.padd(center, utils.spiral(i))
if self.block_at(*s) != blocks.SAND:
continue
# make sure it has solid below
@@ -206,18 +216,17 @@ class MCWorld:
continue
if not self.sand_adjacent_safe(s):
continue
return True
return False
def find_sand_slice(self, center, distance, bad_slices=[]):
def find_sand_slice(self, center, distance, y_limit=0, bad_slices=[], prev_layer=0):
# returns the centre coord of the next 5x5x1 slice that still has
# diggable sand in it. lower slices are only valid if there's an
# adjacent slice farther at the same level. this should ensure an
# upside down pyramid gets excavated so the edges are still climbable
for v in count():
peak = utils.padd(center, (0, 20-v, 0))
for v in count(prev_layer):
peak = utils.padd(center, (0, 10-v, 0))
slices = []
layer = 0
@@ -228,14 +237,16 @@ class MCWorld:
check = utils.padd(peak, offset)
check = utils.padd(check, (0, layer, 0))
if utils.phyp(center, check) >= distance:
if y_limit and check[1] - center[1] > y_limit:
break
if utils.phyp_king(center, check) > distance:
break
if self.check_sand_slice(check) and check not in bad_slices:
slices.append(check)
if len(slices):
return slices[-1]
return v, slices[-1]
elif v > 40:
return None, None
@@ -262,6 +273,31 @@ class MCWorld:
for a in self.find_blocks_3d(center, blocks.LEAF_IDS, distance, 10):
yield a
def find_monsters(self, center, distance):
# finds monsters within distance
result = []
for eid, mob in self.g.mobs.items():
if mob.type not in mobs.EVIL_IDS:
continue
pos = utils.pint((mob.x, mob.y, mob.z))
if utils.phyp(center, pos) > distance:
continue
result.append(mob)
return result
def find_threats(self, center, distance):
# finds monsters on the surface within distance
monsters = self.find_monsters(center, distance)
result = []
for mob in monsters:
pos = utils.pint((mob.x, mob.y, mob.z))
# check distance number of blocks above, close enough?
if not self.check_air_column(pos, distance):
continue
result.append(mob)
return result
class Game:
def __init__(self, global_state):
@@ -365,7 +401,7 @@ class Game:
else:
return
if text == 'zzz':
if text.startswith('zzz'):
text = '!zzz'
if text.startswith('! '):
@@ -382,166 +418,205 @@ class Game:
command = text
data = None
if command == 'ping':
reply = 'pong'
try:
if command == 'ping':
reply = 'pong'
if command == 'echo' and data:
reply = data
if command == 'echo' and data:
reply = data
if command == 'respawn':
packet = serverbound.play.ClientStatusPacket()
packet.action_id = serverbound.play.ClientStatusPacket.RESPAWN
self.g.connection.write_packet(packet)
reply = 'ok'
if command == 'pos':
reply = str(utils.pint(self.g.pos))[1:-1]
if command == 'afk':
reply = '/afk'
if command == 'error':
reply = 'ok'
raise
if command == 'break':
self.break_block(blocks.TEST_BLOCK)
reply = 'ok'
if command == 'gather' and data:
if data == 'wood':
self.g.job.state = self.g.job.gather_wood
reply = 'ok'
elif data == 'sand':
self.g.job.state = self.g.job.gather_sand
if command == 'respawn':
packet = serverbound.play.ClientStatusPacket()
packet.action_id = serverbound.play.ClientStatusPacket.RESPAWN
self.g.connection.write_packet(packet)
reply = 'ok'
if reply:
for i in self.g.inv.values():
print(i.item_id)
if i.item_id in items.BED_IDS:
break
else:
reply += ', I need a bed'
if command == 'pos':
reply = str(utils.pint(self.g.pos))[1:-1]
if command == 'farm' and data:
if data == 'wood':
self.g.job.state = self.g.job.farm_wood
reply = 'ok'
if reply:
for i in self.g.inv.values():
print(i.item_id)
if i.item_id in items.BED_IDS:
break
else:
reply += ', I need a bed'
if command == 'stop':
self.g.job.stop()
self.g.path = []
self.g.look_at = None
reply = 'ok'
if command == 'inv':
inv_list = []
for i in self.g.inv.values():
if i.present:
inv_list.append('{}:{} x {}'.format(items.ITEM_NAMES[i.item_id], str(i.item_id), i.item_count))
inv_list.sort()
result = '\n'.join(inv_list)
print(result or 'Empty')
if command == 'drop':
self.drop_stack()
if command == 'time':
reply = str(self.g.time)
if command == 'select' and data:
item = int(data)
if self.select_item([item]):
reply = 'ok'
else:
reply = 'not found'
if command == 'dump' and data:
item = int(data)
if self.count_items([item]):
self.g.dumping = item
reply = 'ok'
else:
reply = 'not found'
if command == 'count' and data:
item = int(data)
reply = str(self.count_items([item]))
if command == 'open':
self.open_container(blocks.TEST_BLOCK)
if command == 'close':
if self.g.window:
self.close_window()
else:
reply = 'nothing open'
if command == 'click' and data:
if self.g.window:
slot, button, mode = [int(x) for x in data.split(' ')]
try:
item = self.g.window.contents[slot]
except KeyError:
item = Slot(present=False, item_id=None, item_count=None, nbt=None)
print(item)
self.click_window(slot, button, mode, item)
else:
reply = 'nothing open'
if command == 'loaded':
reply = str(self.g.chunks.get_loaded_area())
if command == 'gapple':
self.g.job.state = self.g.job.find_gapple
if data:
self.g.job.find_gapple_states.count = int(data)
reply = 'ok'
if command == 'objects':
for k, v in self.g.objects.items():
if data and v.item_id != int(data): continue
print(str(k) + ':', v)
if command == 'cache':
self.g.job.state = self.g.job.cache_items
self.g.job.cache_items_states.minimum = 0
self.g.job.cache_items_states.silent = True
reply = 'ok'
if command == 'spiral' and data:
for i in range(int(data)):
print(utils.spiral(i))
if command == 'sand_slice':
try:
result = self.g.world.find_sand_slice(utils.pint(self.g.pos), 50)
reply = str(result)
except:
import traceback
print(traceback.format_exc())
reply = 'error'
if command == 'zzz':
if not self.g.afk:
if command == 'afk':
reply = '/afk'
if command == 'print':
try:
if command == 'error':
reply = 'ok'
raise
if command == 'break':
self.break_block(blocks.TEST_BLOCK)
reply = 'ok'
if command == 'gather' and data:
if data == 'wood':
self.g.job.state = self.g.job.gather_wood
reply = 'ok'
elif data == 'sand':
self.g.job.state = self.g.job.gather_sand
reply = 'ok'
if reply:
for i in self.g.inv.values():
print(i.item_id)
if i.item_id in items.BED_IDS:
break
else:
reply += ', I need a bed'
if command == 'farm' and data:
if data == 'wood':
self.g.job.state = self.g.job.farm_wood
reply = 'ok'
elif data == 'sand':
self.g.job.state = self.g.job.farm_sand
reply = 'ok'
if reply:
for i in self.g.inv.values():
if i.item_id in items.BED_IDS:
break
else:
reply += ', I need a bed'
if command == 'stop':
self.g.job.stop()
self.g.path = []
self.g.look_at = None
reply = 'ok'
if command == 'inv':
inv_list = []
for i in self.g.inv.values():
if i.present:
inv_list.append('{}:{} x {}'.format(items.ITEM_NAMES[i.item_id], str(i.item_id), i.item_count))
inv_list.sort()
result = '\n'.join(inv_list)
print(result or 'Empty')
if command == 'drop':
self.drop_stack()
if command == 'time':
reply = str(self.g.time)
if command == 'select' and data:
item = int(data)
if self.select_item([item]):
reply = 'ok'
else:
reply = 'not found'
if command == 'dump' and data:
item = int(data)
if self.count_items([item]):
self.g.dumping = item
reply = 'ok'
else:
reply = 'not found'
if command == 'count' and data:
item = int(data)
reply = str(self.count_items([item]))
if command == 'open':
self.open_container(blocks.TEST_BLOCK)
if command == 'close':
if self.g.window:
self.close_window()
else:
reply = 'nothing open'
if command == 'click' and data:
if self.g.window:
slot, button, mode = [int(x) for x in data.split(' ')]
try:
item = self.g.window.contents[slot]
except KeyError:
item = Slot(present=False, item_id=None, item_count=None, nbt=None)
print(item)
self.click_window(slot, button, mode, item)
else:
reply = 'nothing open'
if command == 'loaded':
reply = str(self.g.chunks.get_loaded_area())
if command == 'gapple':
self.g.job.state = self.g.job.find_gapple
if data:
self.g.job.find_gapple_states.count = int(data)
reply = 'ok'
if command == 'objects':
if data == 'clear':
self.g.objects = {}
reply = 'ok'
else:
for k, v in self.g.objects.items():
if data and v.item_id != int(data): continue
print(str(k) + ':', v, items.ITEM_NAMES[v.item_id])
if command == 'mobs':
if data == 'clear':
self.g.mobs = {}
reply = 'ok'
else:
all_mobs = sorted(list(self.g.mobs.items()), key=lambda x: x[1].type)
for k, v in all_mobs:
if data and v.type != int(data): continue
print(str(k) + ':', v, mobs.MOB_NAMES[v.type])
reply = str(len(all_mobs)) + ' mobs'
if command == 'monsters':
monsters = sorted(list(self.g.mobs.items()), key=lambda x: x[1].type)
count = 0
for k, v in monsters:
if v.type not in mobs.EVIL_IDS: continue
if data and v.type != int(data): continue
count += 1
print(str(k) + ':', v, mobs.MOB_NAMES[v.type])
reply = str(count) + ' monsters'
if command == 'threats':
distance = int(data) if data else 20
p = utils.pint(self.g.pos)
threats = self.g.world.find_threats(p, distance)
for t in threats:
print(str(t.entity_id) + ':', t, mobs.MOB_NAMES[t.type])
reply = str(len(threats)) + ' threats'
if command == 'cache':
self.g.job.state = self.g.job.cache_items
self.g.job.cache_items_states.minimum = 0
self.g.job.cache_items_states.silent = True
reply = 'ok'
if command == 'spiral' and data:
for i in range(int(data)):
print(utils.spiral(i))
if command == 'sand_slice':
result = self.g.world.find_sand_slice(utils.pint(self.g.pos), 50)
reply = str(result)
if command == 'zzz':
if not self.g.afk:
if self.g.path:
travel_time = int(len(self.g.path) * 0.4) + 2
reply = 'give me ' + str(travel_time) + ' secs, moving'
self.g.queue_afk = True
else:
reply = '/afk'
if command == 'print' and sender == 'tanner6':
data = data.replace('^', '.')
reply = str(eval(data))
except BaseException as e:
import traceback
print(traceback.format_exc())
reply = 'Error: {} - {}\n'.format(e.__class__.__name__, e)
except BaseException as e:
import traceback
print(traceback.format_exc())
reply = 'Error: {} - {}\n'.format(e.__class__.__name__, e)
pass
if reply:
print(reply)
@@ -567,8 +642,8 @@ class Game:
def break_block(self, location):
p = utils.pint(self.g.pos)
if utils.phyp(p, location) > blocks.BREAK_DISTANCE:
return False
#if utils.phyp(p, location) > blocks.BREAK_DISTANCE + 1:
# return False
bid = self.g.chunks.get_block_at(*location)
if bid == 0:
@@ -596,6 +671,7 @@ class Game:
self.g.breaking = None
def handle_break_animation(self, packet):
return
print(packet)
def handle_break_ack(self, packet):
@@ -761,22 +837,28 @@ class Game:
obj.item_count = entry.value.item_count
def handle_spawn_living(self, packet):
print(packet)
return
self.g.mobs[packet.entity_id] = Munch(
entity_id=packet.entity_id,
entity_uuid=packet.entity_uuid,
type=packet.type,
x=packet.x,
y=packet.y,
z=packet.z,
)
def handle_entity_position(self, packet):
obj = self.g.objects.get(packet.entity_id, None)
if obj:
pass
#obj.x += packet.delta_x
#obj.y += packet.delta_y
#obj.z += packet.delta_z
mob = self.g.mobs.get(packet.entity_id, None)
if mob:
mob.x += packet.delta_x / 4096.0
mob.y += packet.delta_y / 4096.0
mob.z += packet.delta_z / 4096.0
def handle_entity_position_rotation(self, packet):
obj = self.g.objects.get(packet.entity_id, None)
if obj:
print('object rotation found:', packet)
raise
mob = self.g.mobs.get(packet.entity_id, None)
if mob:
mob.x += packet.delta_x / 4096.0
mob.y += packet.delta_y / 4096.0
mob.z += packet.delta_z / 4096.0
def handle_entity_velocity(self, packet):
obj = self.g.objects.get(packet.entity_id, None)
@@ -790,6 +872,16 @@ class Game:
for eid in packet.entity_ids:
if eid in self.g.objects:
del self.g.objects[eid]
if eid in self.g.mobs:
del self.g.mobs[eid]
def leave_bed(self):
packet = EntityActionPacket()
packet.entity_id = self.g.eid
packet.action_id = 2
packet.jump_boost = 0
self.g.connection.write_packet(packet)
def tick(self):
if self.g.breaking:
@@ -805,6 +897,10 @@ class Game:
else:
self.g.dumping = None
if not len(self.g.path):
if not self.g.path:
self.g.correction_count = 0
if self.g.queue_afk:
self.g.chat.send('/afk')
self.g.queue_afk = False