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(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)