minecraft-bot/mosfet/commands.py

586 lines
23 KiB
Python

import re
import time
from datetime import datetime, timedelta
import random
from itertools import count
from munch import Munch
from mosfet.protocol.types import Slot
from mosfet import print_help
from mosfet import utils
from mosfet import path
from mosfet import bot
from mosfet.info import blocks
from mosfet.info import items
from mosfet.info import mobs
class Commands:
def __init__(self, global_state):
self.g = global_state
self.g.chat.set_handler(self.handle_chat)
def handle_chat(self, message):
source, sender, text = message
reply = ''
private = False
for_me = False
authed = sender == '0c123cfa-1697-4427-9413-4b645dee7ec0'
bot_num = self.g.name[-1]
if source == 'SYSTEM':
self.g.command_lock = False
if text == 'You are now AFK.':
self.g.afk = True
elif text == 'You are no longer AFK.':
self.g.afk = False
text = text.replace('zzz', '!zzz')
match = re.match(r'(.*\W+)\s+(['+bot_num+'|!])(\w+) ?(.*)', text)
if match:
meta, prefix, command, data = match.groups()
else:
return
if '-> me' in meta:
private = True
if prefix == bot_num:
for_me = True
if data.startswith('[') and data.endswith(']'):
command = 'nosquarebrackets'
try:
## ### Public Commands
## These can be ran by anyone, all bots will reply.
## !help - prints this whole help message to console
## !help [command] - replies in-game explaining command
if command == 'help':
if data:
for line in print_help.HELP_LINES:
if line[1:].startswith(data) or line[1:].startswith(data[1:]):
reply = 'command ' + line
break
else: # for
reply = 'command not found'
else:
print()
print()
for line in print_help.HELP_LINES:
print(line)
reply = 'check console'
## !ping - replies with "pong"
if command == 'ping':
reply = 'pong'
## !echo [data] - replies with "data"
if command == 'echo' and data:
reply = data
if command == 'nosquarebrackets':
reply = 'don\'t literally put the [ ]'
## !pos - replies with position and dimension
if command == 'pos':
reply = str(utils.pint(self.g.pos))[1:-1] + ', ' + self.g.dimension
## !afk - goes AFK with /afk
if command == 'afk':
if not self.g.afk:
reply = '/afk'
## !unafk - goes not AFK with /afk
if command == 'unafk':
if self.g.afk:
reply = '/afk'
## !error - raises an error
if command == 'error':
reply = 'ok'
raise
## !inv - replies and prints current inventory
if command == 'inv' or command == 'inventory':
inv_list = []
uniq_item_counts = {}
for i in self.g.inv.values():
if i.present:
inv_list.append((items.ITEM_NAMES[i.item_id], str(i.item_id), i.item_count))
if i.item_id not in uniq_item_counts:
uniq_item_counts[i.item_id] = 0
uniq_item_counts[i.item_id] += i.item_count
inv_list.sort()
console_result = '\n'.join(['{}:{} x {}'.format(*x) for x in inv_list])
if not console_result:
print('Empty')
reply = 'empty'
else:
print(console_result)
reply_result_1 = ', '.join(['{}:{} x {}'.format(*x) for x in inv_list])
reply_result_2 = ', '.join(['{}:{} x {}'.format(items.ITEM_NAMES[k], str(k), v) for k,v in uniq_item_counts.items()])
reply_result_3 = ', '.join(['{}:{}x{}'.format(items.ITEM_NAMES[k], str(k), v) for k,v in uniq_item_counts.items()])
reply_result_4 = ', '.join(['{}:{}x{}'.format(re.sub(r'[aeiou]', '', items.ITEM_NAMES[k]), str(k), v) for k,v in uniq_item_counts.items()])
reply_result_5 = ' '.join(['{}{}x{}'.format(re.sub(r'[aeiou]', '', items.ITEM_NAMES[k]), str(k), v) for k,v in uniq_item_counts.items()])
for r in [reply_result_1, reply_result_2, reply_result_3, reply_result_4, reply_result_5]:
reply = r
if len(reply) < 256:
break
## !time - replies with Minecraft world time
if command == 'time':
seconds = self.g.time * 3.6
start = datetime(2000, 1, 1, hour=6)
mctime = start + timedelta(seconds=seconds)
reply = str(self.g.time) + ' - ' + mctime.strftime('%I:%M %p')
## !count [id] - counts the number of items with that id
if command == 'count' and data:
item = int(data)
reply = str(self.g.game.count_items([item]))
## !loaded - replies with the current loaded area
if command == 'loaded':
l = self.g.chunks.get_loaded_area()
reply = '{}, {}, {} to {}, {}, {}'.format(*l[0], *l[1])
reply += ' - ' + str(len(self.g.chunks.chunks)//16) + ' chunks'
## !players - prints the current players
## !players clear - clears the current player list
if command == 'players':
if data == 'clear':
self.g.players = {}
reply = 'ok'
else:
for k, v in self.g.players.items():
print(str(k) + ':', v, self.g.player_names[v.player_uuid])
## !objects - prints the current items on ground
## !objects clear - clears the current object list
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])
## !mobs - prints the current mobs
## !mobs clear - clears the current mob list
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'
## !monsters - prints the current monsters
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'
## !villagers - prints the current villagers
if command == 'villagers':
all_mobs = sorted(list(self.g.mobs.items()), key=lambda x: x[1].type)
count = 0
for k, v in all_mobs:
type_name = mobs.MOB_NAMES[v.type]
if type_name != 'villager' : continue
count += 1
print(str(k) + ':', v, type_name)
reply = str(count) + ' villagers'
## !threats - prints the dangerous monsters within 20 blocks
## !threats [num] - prints the dangerous monsters within num blocks
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 == '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)
## "zzz" or !zzz - bot does /afk to let others sleep
if command == 'zzz':
if not self.g.afk and self.g.dimension == 'overworld':
reply = '/afk'
self.g.afk_timeout = 5.0
## !tree - replies with the closest tree
if command == 'tree':
pos = utils.pint(self.g.pos)
tree = next(self.g.world.find_trees(pos, 50))
reply = str(tree)[1:-1]
## !info [query] - replies with info on a coordinate, block, item, or player
if command == 'info':
if not reply:
try:
check = data.replace('(', ' ').replace(')', ' ').replace(',', ' ')
x1, y1, z1 = [int(x) for x in check.split()]
coord = (x1, y1, z1)
block = self.g.world.block_at(*coord)
if not reply and block is None:
reply = 'coord out of range'
if not reply:
reply = 'Block: ' + blocks.BLOCKS[block] + ':' + str(block)
if blocks.PROPS[block]:
reply += ' - ' + ', '.join(['{}:{}'.format(k, v) for k, v in blocks.PROPS[block].items()])
except (AttributeError, ValueError):
pass
if not reply:
try:
check = int(data)
if check in blocks.BLOCKS:
block = check
reply += 'Block: ' + blocks.BLOCKS[block] + ':' + str(block)
if blocks.PROPS[block]:
reply += ' - ' + ', '.join(['{}:{}'.format(k, v) for k, v in blocks.PROPS[block].items()])
if check in blocks.BLOCKS and check in items.ITEM_NAMES:
reply += ' / '
if check in items.ITEM_NAMES:
item = check
reply += 'Item: ' + items.ITEM_NAMES[item] + ':' + str(item)
except ValueError:
pass
check = data.lower()
if not reply and check in self.g.player_names:
uuid = self.g.player_names[check]
for p in self.g.players.values():
if p.player_uuid == uuid:
player = p
break
else: # for
reply = 'player out of range'
if not reply:
reply += 'Player: '
results = []
for k, v in player.items():
try:
results.append('{}:{}'.format(k, int(v)))
except ValueError:
results.append('{}:{}'.format(k, str(v)))
reply += ', '.join(results)
################# Specific commands ##########################
## ### Bot-specific Commands
## These will only run for the bot they are addressed to.
if for_me:
## 1respawn - respawns the bot if it's dead
if command == 'respawn':
self.g.game.respawn()
reply = 'ok'
## 1gather wood - gathers wood from the world
## 1gather sand - gathers sand from the world
if command == 'gather' and data:
if data == 'wood':
self.g.job.state = self.g.job.gather_wood
reply = 'ok'
elif data == 'sand':
if not self.g.sand_origin or not self.g.chunks.check_loaded(self.g.sand_origin):
self.g.sand_origin = utils.pint(self.g.pos)
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'
## 1farm wood - farms wood from a certain area
## 1farm sand - farms sand from a certain area
## 1farm wart - farms netherwart from a certain area
## 1farm crop - farms mature crops from a certain area
if command == 'farm' and data:
if data == 'wood':
self.g.job.state = self.g.job.farm_wood
reply = 'ok'
elif data == 'sand':
if not self.g.sand_origin or not self.g.chunks.check_loaded(self.g.sand_origin):
self.g.sand_origin = utils.pint(self.g.pos)
self.g.job.state = self.g.job.farm_sand
reply = 'ok'
elif data == 'wart':
self.g.job.state = self.g.job.farm_wart
reply = 'ok'
elif data.startswith('crop'):
self.g.job.state = self.g.job.farm_crop
reply = 'ok'
if reply and self.g.dimension == 'overworld':
for i in self.g.inv.values():
if i.item_id in items.BED_IDS:
break
else:
reply += ', I need a bed'
## 1loiter - stands still but eats, sleeps, and flees
if command == 'loiter':
self.g.job.state = self.g.job.loiter
reply = 'ok'
## 1trade - sells items to villagers to get emeralds
if command == 'trade':
self.g.job.state = self.g.job.trade
reply = 'ok'
## 1stop - stops the current job and resets bot
if command == 'stop':
self.g.game.close_window()
bot.init(self.g)
reply = 'ok'
## 1drop - drops the current stack its holding
if command == 'drop':
self.g.game.drop_stack()
## 1select [id] - moves item with id into main hand
if command == 'select' and data:
item = int(data)
if self.g.game.select_item([item]):
reply = 'ok'
else:
reply = 'not found'
## 1dump [id] - drops all items matching id
if command == 'dump' and data:
item = int(data)
if self.g.game.count_items([item]):
self.g.dumping = item
reply = 'ok'
else:
reply = 'not found'
## 1drain - drops all items in inventory
if command == 'drain':
self.g.draining = True
reply = 'ok'
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 == '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'
## 1fill [x] [y] [z] [x] [y] [z] - fills the cuboid with the block at the first coordinate
if command == 'fill':
try:
data = data.replace('(', ' ').replace(')', ' ').replace(',', ' ')
x1, y1, z1, x2, y2, z2 = [int(x) for x in data.split()]
except (AttributeError, ValueError):
reply = 'usage: !fill x1 y1 z1 x2 y2 z2'
if not reply:
coord1 = (x1, y1, z1)
coord2 = (x2, y2, z2)
block = self.g.world.block_at(*coord1)
if not reply and y1 > y2:
reply = 'can only fill upwards'
if not reply and block is None:
reply = 'first coord out of range'
if not reply and block == 0:
reply = 'can\'t fill with air'
if not reply:
self.g.filling = Munch(coord1=coord1, coord2=coord2, block=block)
self.g.job.state = self.g.job.fill_blocks
reply = 'filling ' + str(utils.pvolume(coord1, coord2)) + ' with ' + blocks.BLOCKS[block]
## 1here - bot comes to your location
if command == 'here':
if not reply:
for p in self.g.players.values():
if p.player_uuid == sender:
player = p
break
else: # for
reply = 'can\'t find you'
if not reply:
pos = utils.pint(self.g.pos)
goal = utils.pint((p.x, p.y, p.z))
start = time.time()
navpath = self.g.world.path_to_place(pos, goal)
if navpath:
self.g.path = navpath
if self.g.job:
self.g.job.stop()
print(len(navpath))
print(navpath)
print(round(time.time() - start, 3), 'seconds')
if self.g.job:
self.g.job.stop()
self.g.look_at = None
reply = 'ok'
else:
reply = 'no path'
## 1goto [x] [y] [z] - sends the bot to coordinate (x, y, z)
if command == 'goto':
try:
data = data.replace('(', ' ').replace(')', ' ').replace(',', ' ')
x2, y2, z2 = [int(x) for x in data.split()]
except (AttributeError, ValueError):
reply = 'usage: !goto x y z'
if not reply:
pos = utils.pint(self.g.pos)
goal = utils.pint((x2, y2, z2))
start = time.time()
navpath = self.g.world.path_to_place(pos, goal)
if navpath:
self.g.path = navpath
if self.g.job:
self.g.job.stop()
print(len(navpath))
print(navpath)
print(round(time.time() - start, 3), 'seconds')
if self.g.job:
self.g.job.stop()
self.g.look_at = None
reply = 'ok'
else:
reply = 'no path'
if command == 'break':
self.g.game.break_block(blocks.TEST_BLOCK)
reply = 'ok'
if command == 'open':
self.g.game.open_container(blocks.TEST_BLOCK)
## 1close - closes the current Minecraft window
if command == 'close':
if self.g.window:
self.g.game.close_window()
reply = 'ok'
else:
reply = 'nothing open'
## 1click [slot] [button] [mode] - clicks the current window
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)
print(item)
self.g.game.click_window(slot, button, mode, item)
else:
reply = 'nothing open'
## 1use - use the item it's currently holding
if command == 'use':
self.g.game.use_item(0)
## 1interact [entity id] - interacts with that entity
if command == 'interact' and data:
self.g.game.interact(int(data))
if command == 'test':
reply = 'ok'
r = self.g.world.find_villager_openings((615, 78, 493))
print(r)
################# Authorized commands ##########################
## ### Authorized Commands
## These dangerous commands can only be ran by the bot owner.
if authed:
## 1print [expression] - replies with Python eval(expression)
if command == 'print':
data = data.replace('`', '.')
reply = str(eval(data))
## 1exit - exits the program
if command == 'exit':
import os
os._exit(0)
except BaseException as e:
import traceback
print(traceback.format_exc())
reply = 'Error: {} - {}\n'.format(e.__class__.__name__, e)
pass
if reply:
print('Reply:', reply)
if len(reply) >= 256:
reply = 'reply too long, check console'
if private and not reply.startswith('/'):
self.g.chat.send('/r ' + reply)
else:
self.g.chat.send(reply)