import re import time import importlib import random from itertools import count from math import ceil, floor from minecraft.networking.types import BlockFace from mosfet.protocol.managers import ChunkNotLoadedException from mosfet import utils from mosfet import path from mosfet.info import blocks from mosfet.info import items from mosfet.info import mobs class CacheItemsStates: def idle(self): return None def _has_compatible_labeled_chest_near(self, bed, distance=16): world = self.g.get('world') if world is None: return False keep_slots = self._needed_inventory_slots_to_keep() for chest in world.find_labeled_chests(bed, distance): positions = chest.get('positions') or (chest['position'],) if min(utils.phyp(bed, position) for position in positions) > distance: continue item_ids = chest['item_ids'] for slot, item in self.g.inv.items(): if (not 9 <= slot <= 44 or not item.present or item.item_id not in item_ids or item.item_id in items.WANTED_ITEMS): continue if (item.item_id in items.NEEDED_ITEMS and slot in keep_slots): continue return True return False def init(self): self.skip_inventory_slots = set() self.skip_items = [] self.failed_transfer_slots = set() self.pending_transfer = None self.transferred_to_labeled_chest = False self.bad_openings = set() self.is_labeled_container = False self.container_item_ids = None num_stacks = self.g.game.count_inventory_slots() has_override = 'minimum' in self.__dict__ minimum = self.__dict__.pop( 'minimum', self.g.get('minimum_cache_slots', 27)) post_sleep_bed = self.g.pop('post_sleep_cache_bed', None) if (not has_override and post_sleep_bed is not None and self._has_compatible_labeled_chest_near(post_sleep_bed)): minimum = ceil(minimum / 2) print('Lowering cache threshold after sleep near a compatible chest:', minimum) print('Inventory amount:', num_stacks) if num_stacks == 0: self.state = self.cleanup elif num_stacks >= minimum: self.state = self.find_trapped_chests else: print('Aborting caching, not full') self.state = self.cleanup def find_trapped_chests(self): print('Finding labeled and trapped chests...') w = self.g.world p = utils.pint(self.g.pos) self.labeled_chests = w.find_labeled_chests(p, 80) self.checked_labeled_chests = set() self.trapped_chests = w.find_blocks_indexed(p, blocks.TRAPPED_CHEST_IDS) print('Found {} labeled chests; found {} trapped chests: {}'.format( len(self.labeled_chests), len(self.trapped_chests), self.trapped_chests)) self.state = self.choose_trapped_chest def _opening_in_bad_region(self, world, opening): opening = tuple(opening) if opening in self.bad_openings: return True if not self.bad_openings: return False nearest = min(self.bad_openings, key=lambda bad: utils.phyp(opening, bad)) if (world.path_to_place(opening, nearest) and world.path_to_place(nearest, opening)): self.bad_openings.add(opening) return True return False def choose_trapped_chest(self): print('Choosing a labeled or trapped chest...') self.bad_openings = set() w = self.g.world p = utils.pint(self.g.pos) for container in self.labeled_chests: key = container['positions'] if key in self.checked_labeled_chests: continue self.checked_labeled_chests.add(key) if not self._has_transferable_item(container['item_ids']): continue self.area = container['position'] self.container_item_ids = container['item_ids'] self.is_labeled_container = True for opening in w.find_cache_openings(self.area, p): opening_position = tuple(opening['position']) if self._opening_in_bad_region(w, opening_position): continue navpath = w.path_to_place(p, opening_position) if navpath: print('Going to labeled chest', container['labels'], self.area) self.opening = opening_position self.open_interaction = opening self.g.path = navpath self.g.look_at = None self.state = self.going_to_trapped_chest return self.bad_openings.add(opening_position) print('No path to labeled chest', self.area) if not len(self.trapped_chests): self.is_labeled_container = False self.container_item_ids = None if self.transferred_to_labeled_chest: print('Already cached items in labeled chests; skipping new chest') self.state = self.cleanup return print('No labeled or trapped chests') self.state = self.select_chest return self.is_labeled_container = False self.container_item_ids = None chest = self.trapped_chests[0] self.area = chest for opening in w.find_cache_openings(chest, p): opening_position = tuple(opening['position']) if self._opening_in_bad_region(w, opening_position): continue navpath = w.path_to_place(p, opening_position) if navpath: print('navpath to trapped chest:', navpath) self.opening = opening_position self.open_interaction = opening self.g.path = navpath self.g.look_at = None self.state = self.going_to_trapped_chest return self.bad_openings.add(opening_position) self.trapped_chests.pop(0) time.sleep(0.1) def going_to_trapped_chest(self): if utils.pint(self.g.pos) == self.opening: self.g.look_at = self.open_interaction['look_at'] self.wait_time = 0.5 self.state = self.turn_to_trapped_chest else: self.g.look_at = None def turn_to_trapped_chest(self): self.g.look_at = self.open_interaction['look_at'] self.wait_time -= utils.TICK if self.wait_time <= 0: self.state = self.open_chest def select_chest(self): self.is_labeled_container = False self.container_item_ids = None if self.g.game.select_item([items.CHEST_ID]): self.state = self.find_cache_spot else: print('No chest, aborting') self.state = self.cleanup def find_cache_spot(self): print('Finding a chest spot...') w = self.g.world p = utils.pint(self.g.pos) for area in w.find_cache_areas(p, 16): print('Found area:', area) if area not in self.bad_areas: break else: # for print('Unable to find area') self.state = self.cleanup return self.area = area openings = w.find_cache_openings(self.area, p) self.bad_openings = set() for opening in openings: opening_position = tuple(opening['position']) if self._opening_in_bad_region(w, opening_position): continue navpath = w.path_to_place(p, opening_position) if navpath: self.opening = opening_position self.open_interaction = opening break self.bad_openings.add(opening_position) else: # for print('Unable to get to cache area', self.area) self.bad_areas.append(self.area) self.state = self.cleanup return self.g.path = navpath self.g.look_at = None self.state = self.going_to_area def going_to_area(self): if utils.pint(self.g.pos) == self.opening: self.g.look_at = self.open_interaction['look_at'] self.wait_time = 0.5 self.state = self.turn_to_cache_spot else: self.g.look_at = None def turn_to_cache_spot(self): self.g.look_at = self.open_interaction['look_at'] self.wait_time -= utils.TICK if self.wait_time <= 0: self.state = self.place_chest def place_chest(self): support = utils.padd(self.area, path.BLOCK_BELOW) self.g.game.place_block(support, BlockFace.TOP, sneaking=True) self.state = self.open_chest def open_chest(self): print('Opening chest') self.failed_transfer_slots.clear() self.pending_transfer = None self.g.game.open_container( self.open_interaction['target'], face=self.open_interaction['face'], cursor=self.open_interaction['cursor']) self.wait_time = 1 self.state = self.wait def wait(self): # wait for server to send us chest contents if self.wait_time > 0: self.wait_time -= utils.TICK else: self.state = self.move_items def _needed_inventory_slots_to_keep(self): needed = set(items.NEEDED_ITEMS) groups = ( items.BED_IDS, {items.CHEST_ID}, items.FOOD_IDS, items.AXE_IDS, items.SHOVEL_IDS, ) keep = set() covered = set() for group in groups: group = needed.intersection(group) if not group: continue covered.update(group) stacks = [(slot, item) for slot, item in self.g.inv.items() if 5 <= slot <= 45 and item.present and item.item_id in group] if stacks: keep.add(max(stacks, key=lambda pair: pair[1].item_count)[0]) for item_id in needed - covered: stacks = [(slot, item) for slot, item in self.g.inv.items() if 5 <= slot <= 45 and item.present and item.item_id == item_id] if stacks: keep.add(max(stacks, key=lambda pair: pair[1].item_count)[0]) return keep def _has_transferable_item(self, item_ids): item_ids = set(item_ids) needed_slots = self._needed_inventory_slots_to_keep() planned_skip_items = set(self.skip_items) for inventory_slot, item in self.g.inv.items(): if (not item.present or item.item_id not in item_ids or inventory_slot in self.skip_inventory_slots): continue if (item.item_id in items.NEEDED_ITEMS and inventory_slot in needed_slots): continue if (item.item_id in items.WANTED_ITEMS and item.item_id not in planned_skip_items): planned_skip_items.add(item.item_id) continue return True return False def _window_player_slot(self, window_slot): offset = window_slot - self.g.window.container_slots if 0 <= offset < 27: return 9 + offset if 27 <= offset < 36: return 36 + offset - 27 return None def move_items(self): if self.g.item_lock: return w = self.g.window if not w: print('Didnt get a window, aborting') self.state = self.cleanup return if not w.container_slots: print('Chest contents have not arrived, aborting') self.g.game.close_window() self.state = self.cleanup return if w.container_slots not in (27, 54): print('Got non-chest window, aborting') self.g.game.close_window() self.state = self.cleanup return if self.pending_transfer: pending = self.pending_transfer if time.monotonic() < pending['confirm_after']: return source = w.contents.get(pending['window_slot']) moved = (source is None or not source.present or source.item_id != pending['item_id'] or source.item_count < pending['item_count']) if moved: if pending['is_labeled']: self.transferred_to_labeled_chest = True else: self.failed_transfer_slots.add(pending['window_slot']) self.pending_transfer = None w_container_slots = range(w.container_slots) w_inventory_slots = range(w.container_slots, len(w.contents)) used_slots = self.g.game.count_window_slots() print('used:', used_slots, 'total:', len(w_container_slots)) if used_slots >= len(w_container_slots): print('Container is too full, aborting') self.g.game.close_window() self.g.look_at = None self.state = (self.close_chest if self.is_labeled_container else self.cleanup) return slot_list = [] needed_slots = (self._needed_inventory_slots_to_keep() if self.is_labeled_container else set()) for slot_num in w_inventory_slots: if slot_num not in w.contents: continue slot = w.contents[slot_num] if not slot.present: continue if self.is_labeled_container: if slot.item_id not in self.container_item_ids: continue inventory_slot = self._window_player_slot(slot_num) if (slot.item_id in items.NEEDED_ITEMS and inventory_slot in needed_slots): continue elif slot.item_id in items.NEEDED_ITEMS: continue inventory_slot = self._window_player_slot(slot_num) if (inventory_slot in self.skip_inventory_slots or slot_num in self.failed_transfer_slots): continue slot_list.append((slot_num, slot)) slot_list.sort(key=lambda x: x[1].item_count, reverse=True) for slot_num, slot in slot_list: if slot.item_id in items.WANTED_ITEMS and slot.item_id not in self.skip_items: print('skipping wanted item', slot) inventory_slot = self._window_player_slot(slot_num) if inventory_slot is not None: self.skip_inventory_slots.add(inventory_slot) self.skip_items.append(slot.item_id) continue print('moving', slot) self.pending_transfer = { 'window_slot': slot_num, 'item_id': slot.item_id, 'item_count': slot.item_count, 'is_labeled': self.is_labeled_container, 'confirm_after': time.monotonic() + 0.15, } self.g.item_lock = True self.g.game.click_window(slot_num, 0, 1, slot) return print('nothing left to move') self.state = self.close_chest def close_chest(self): print('closing chest') self.g.game.close_window() if not self.silent: self.g.chat.send('cache at ' + str(self.area)[1:-1]) if self.is_labeled_container: self.is_labeled_container = False self.container_item_ids = None self.state = self.choose_trapped_chest else: self.state = self.cleanup def cleanup(self): self.g.look_at = None self.state = self.done if (self.g.get('job') and getattr(self.g.job.state, '__name__', None) == 'cache_items'): self.g.job.state = self.g.job.idle def done(self): # never gets ran, placeholder return None def __init__(self, global_state): self.g = global_state self.state = self.idle self.silent = False self.skip_inventory_slots = set() self.failed_transfer_slots = set() self.pending_transfer = None self.transferred_to_labeled_chest = False self.skip_items = [] self.labeled_chests = [] self.checked_labeled_chests = set() self.is_labeled_container = False self.container_item_ids = None self.area = None self.opening = None self.open_interaction = None self.trapped_chests = [] self.bad_areas = [] self.wait_time = 0 def run(self): self.state()