Files
minecraft-bot/mosfet/commands.py
T

690 lines
28 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.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 _send_reply(self, reply, private=False, sender=None):
if private and not reply.startswith('/'):
if isinstance(sender, str) and re.fullmatch(r'[A-Za-z0-9_]{1,16}', sender):
self.g.chat.send('/msg {} {}'.format(sender, reply))
else:
self.g.chat.send('/r ' + reply)
else:
self.g.chat.send(reply)
def handle_chat(self, message):
source, sender, text = message[:3]
reply_target = message[3] if len(message) > 3 else sender
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+([{}!])(\w+) ?(.*)'.format(re.escape(bot_num)),
text)
if match:
meta, prefix, command, data = match.groups()
else:
return
if source == 'DM' or '-> me' in meta:
private = True
if prefix == bot_num:
for_me = True
if data.startswith('[') and data.endswith(']'):
command = 'nosquarebrackets'
unsupported = {
'trade', 'gapple', 'fill',
'break', 'open', 'click', 'use', 'interact',
}
if ((command == 'gather' and data.lower().strip() != 'wood') or
command in unsupported or
(command == 'farm' and
not data.lower().startswith('crop') and
data.lower().strip() not in ('wood', 'wart'))):
reply = 'not migrated to Minecraft 26.3 yet'
print('Reply:', reply)
self._send_reply(reply, private, reply_target)
return
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
## !health - replies with health, hunger, and saturation
if command == 'health':
if not self.g.get('health_update_received'):
reply = 'health status not received yet'
else:
reply = 'health {:.1f}, hunger {}/20, saturation {:.2f}'.format(
self.g.health, self.g.food, self.g.food_saturation)
## !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 slot, i in self.g.inv.items():
if not i.present:
continue
base_name = items.ITEM_NAMES.get(i.item_id, str(i.item_id))
components = getattr(i, 'components', {}) or {}
custom_name = components.get('custom_name') or components.get('item_name')
if isinstance(custom_name, dict):
custom_name = custom_name.get('text') or custom_name.get('raw')
display_name = str(custom_name or base_name)
details = []
for key in ('enchantments', 'stored_enchantments'):
for name, level in components.get(key, {}).items():
details.append('{} {}'.format(name, level))
contents = components.get('container')
if contents:
nested_count = sum(bool(getattr(item, 'present', False))
for item in contents)
details.append('{} items inside'.format(nested_count))
written = components.get('written_book_content')
if written:
title = written.get('title', {}).get('raw', '')
details.append('book: {} ({} pages)'.format(
title, len(written.get('pages', []))))
writable = components.get('writable_book_content')
if writable:
details.append('writable book ({} pages)'.format(
len(writable)))
if getattr(i, 'components_unknown', False):
details.append('components partial')
detail_text = ' [{}]'.format(', '.join(details)) if details else ''
inv_list.append((slot, display_name, str(i.item_id),
i.item_count, detail_text))
uniq_item_counts[i.item_id] = (
uniq_item_counts.get(i.item_id, 0) + i.item_count)
inv_list.sort(key=lambda entry: (entry[1], entry[2]))
console_result = '\n'.join(
'slot {}: {}:{} x {}{}'.format(*entry)
for entry in inv_list)
if not console_result:
print('Empty')
reply = 'empty'
else:
print(console_result)
if self.g.get('inventory_partial', False):
print('Warning: inventory contains stacks with undecoded components.')
reply_result_1 = ', '.join(
'{}:{} x {}{}'.format(*entry[1:]) for entry in inv_list)
reply_result_components = ', '.join(
'{}{}'.format(entry[1], entry[4])
for entry in inv_list if entry[4])
reply_result_2 = ', '.join(
'{}:{} x {}'.format(items.ITEM_NAMES.get(k, str(k)), str(k), v)
for k, v in uniq_item_counts.items())
reply_result_3 = ', '.join(
'{}:{}x{}'.format(items.ITEM_NAMES.get(k, str(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.get(k, str(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.get(k, str(k))), str(k), v)
for k, v in uniq_item_counts.items())
for r in [reply_result_1, reply_result_components,
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(int(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])
chunk_count = getattr(
self.g.chunks, 'loaded_chunk_count',
len(self.g.chunks.chunks) // 16)
reply += ' - ' + str(chunk_count) + ' 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]
## !signs [distance] - prints nearby sign text
if command == 'signs':
distance = float(data) if data else 32
signs = self.g.world.find_signs(
utils.pint(self.g.pos), distance)
if signs:
results = []
for sign in signs:
text = ' | '.join(line for line in sign['lines']
if line.strip())
entry = '{} {}: {}'.format(
sign['side'], sign['position'], text)
results.append(entry)
print('Sign {} {}:'.format(
sign['side'], sign['position']))
for line in sign['lines']:
print(' ' + line)
reply = '; '.join(results)
if len(reply) > 240:
reply = reply[:237] + '...'
else:
reply = 'no signs found within {} blocks'.format(distance)
## !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'
## 1sleep [new] - use a bed; `new` forces bed placement
if command == 'sleep':
self.g.job.sleep_with_bed_states.force_new_bed = (
data.lower() in ('new', 'place'))
self.g.job.state = self.g.job.sleep
reply = 'ok'
## 1eat - eat one food item when hunger/health requires it
if command == 'eat':
self.g.job.state = self.g.job.eat
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'):
mode = self.g.get('game_mode')
if mode is not None and mode != 0:
reply = 'crop farming needs Survival mode'
else:
self.g.job.state = self.g.job.farm_crop
reply = 'ok'
## 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.job.stop()
self.g.game.close_window()
reply = 'ok'
## 1drop - drops the current stack its holding
if command == 'drop':
reply = 'ok' if self.g.game.drop_stack() else 'empty or busy'
## 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)
count = self.g.game.start_dump(item)
reply = ('busy' if count is None else
'ok' if count else 'not found')
## 1drain - drops all items in inventory
if command == 'drain':
count = self.g.game.start_drain()
reply = ('busy' if count is None else
'ok' if count else 'empty')
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:
if self.g.job:
self.g.job.stop()
self.g.path = navpath
print(len(navpath))
print(navpath)
print(round(time.time() - start, 3), 'seconds')
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:
if self.g.job:
self.g.job.stop()
self.g.path = navpath
print(len(navpath))
print(navpath)
print(round(time.time() - start, 3), 'seconds')
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'
self._send_reply(reply, private, reply_target)