import re import time import importlib import random from itertools import count from math import 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 GrabSuppliesStates: DEFAULT_SUPPLIES = { tuple(items.BED_IDS): (1, 1), (items.CHEST_ID,): (1, 1), tuple(items.FOOD_IDS): (1, 1), } def default_supplies(self): return dict(self.DEFAULT_SUPPLIES) def idle(self): return None def init(self): print('Started grab supplies states') self.checked_supply_chests = set() used_slots = self.g.game.count_inventory_slots() print('used:', used_slots, 'total:', self.g.maximum_supply_slots) if used_slots >= self.g.maximum_supply_slots: print('Inventory is too full, aborting') self.state = self.cleanup return self.checked_ground_items = set() self.checked_ground_groups = set() self.checked_supply_chests = set() self.ground_stack_counts = {} self.ground_entity_id = None self.ground_position = None if self.supplies: self.state = self.check_supplies else: print('Aborting getting supplies, none specified') self.state = self.cleanup def _has_nearby_ground_items(self, item_ids, maximum_stacks): if (maximum_stacks and self.ground_stack_counts.get(item_ids, 0) >= maximum_stacks): return False p = utils.pint(self.g.pos) return any( obj.entity_id not in self.checked_ground_items and utils.phyp(p, (obj.x, obj.y, obj.z)) <= 16 for obj in self.g.world.find_objects(item_ids)) def has_nearby_ground_supply(self): return any( self._has_nearby_ground_items(item_ids, limits[1]) for item_ids, limits in self.supplies.items()) def _record_ground_pickup(self): self.checked_ground_items.add(self.ground_entity_id) self.ground_stack_counts[self.target_items] = ( self.ground_stack_counts.get(self.target_items, 0) + 1) if (self.target_maximum_stacks and self.ground_stack_counts[self.target_items] >= self.target_maximum_stacks): self.checked_ground_groups.add(self.target_items) self.g.path = [] self.ground_entity_id = None self.ground_position = None def check_supplies(self): # Search every configured item group on the ground before looking in # labeled chests; a missing bed must not hide a nearby chest or food drop. missing = [] for item_ids, limits in self.supplies.items(): minimum, maximum = limits print('Checking items:', item_ids) num_items = self.g.game.count_items(item_ids) print('Have:', num_items) if (item_ids not in self.checked_ground_groups and self._has_nearby_ground_items(item_ids, maximum)): print('Found matching supplies on the ground') self.target_items = item_ids self.target_minimum = minimum self.target_maximum_stacks = maximum self.state = self.find_ground_items return if num_items >= minimum: print('Have enough, skipping') continue missing.append((item_ids, minimum, maximum)) if item_ids not in self.checked_ground_groups: self.target_items = item_ids self.target_minimum = minimum self.target_maximum_stacks = maximum self.state = self.find_ground_items return if missing: self.target_items, self.target_minimum, self.target_maximum_stacks = missing[0] print('No more ground drops; checking supply-labeled chests for', self.target_items) self.state = self.find_supply_chests return print('Aborting, dont need any supplies') self.state = self.cleanup def find_ground_items(self): stack_count = self.ground_stack_counts.get(self.target_items, 0) if self.target_maximum_stacks and stack_count >= self.target_maximum_stacks: self.checked_ground_groups.add(self.target_items) if (self.g.game.count_items(self.target_items) < self.target_minimum): self.state = self.find_supply_chests else: self.state = self.check_supplies return print('Looking for supplies on the ground...') p = utils.pint(self.g.pos) ground_items = self.g.world.find_objects(self.target_items) ground_items.sort(key=lambda obj: utils.phyp( p, (obj.x, obj.y, obj.z))) for obj in ground_items: if obj.entity_id in self.checked_ground_items: continue current_count = self.g.game.count_items(self.target_items) if utils.phyp(p, (obj.x, obj.y, obj.z)) > 16: continue target = utils.pint((obj.x, obj.y, obj.z)) self.ground_entity_id = obj.entity_id self.ground_position = target self.ground_required_count = max( self.target_minimum, current_count + 1) if utils.phyp(p, (obj.x, obj.y, obj.z)) <= 1.25: self.g.path = [] self.wait_time = 1.0 self.state = self.wait_for_ground_item return navpath = self.g.world.path_to_place(p, target) if not navpath: self.checked_ground_items.add(obj.entity_id) continue print('Going to ground item', obj.item_id, target) self.g.path = navpath self.state = self.going_to_ground_item return self.checked_ground_groups.add(self.target_items) self.state = self.check_supplies def going_to_ground_item(self): obj = self.g.objects.get(self.ground_entity_id) have_enough = (self.g.game.count_items(self.target_items) >= self.ground_required_count) if have_enough: print('Picked up required ground supplies') self._record_ground_pickup() self.state = self.check_supplies elif (obj is None or utils.pint(self.g.pos) == self.ground_position or utils.phyp(self.g.pos, (obj.x, obj.y, obj.z)) <= 1.25): self.wait_time = 1.0 self.state = self.wait_for_ground_item def wait_for_ground_item(self): if (self.g.game.count_items(self.target_items) >= self.ground_required_count): print('Picked up required ground supplies') self._record_ground_pickup() self.state = self.check_supplies return if self.wait_time > 0: self.wait_time -= utils.TICK return if self.ground_entity_id not in self.g.objects: # Entity removal and inventory updates can arrive in either order. self.state = self.check_supplies return print('Could not pick up ground item; trying another') self.checked_ground_items.add(self.ground_entity_id) self.g.path = [] self.ground_entity_id = None self.ground_position = None self.state = self.find_ground_items @staticmethod def _chest_event_key(event): return (event.get('timestamp'), tuple(event.get('position', ()))) def _refresh_useless_chests(self): events = self.g.get('chest_open_events', ()) current_events = {self._chest_event_key(event) for event in events} for event in events: event_key = self._chest_event_key(event) if event_key in self._processed_chest_open_events: continue self._processed_chest_open_events.add(event_key) position = tuple(event.get('position', ())) reopened = [key for key in self.useless_chests if position in key] for key in reopened: self.useless_chests.remove(key) self.checked_supply_chests.discard(key) print('Supply chest was opened by another player; checking it again:', position) self._processed_chest_open_events.intersection_update(current_events) def find_supply_chests(self): print('Finding chests labeled for this bot\'s supplies...') p = utils.pint(self.g.pos) self.supply_chests = self.g.world.find_supply_chests( p, self.g.name, 80) print('Found:', len(self.supply_chests), 'supply-labeled chests') self.state = self.choose_supply_chest def choose_supply_chest(self): print('Choosing a supply-labeled chest...') for container in self.supply_chests: key = tuple(container['positions']) if key in self.useless_chests or key in self.checked_supply_chests: continue self.checked_supply_chests.add(key) self.current_supply_chest_key = key self.chest = container['position'] self.area = self.chest print('Chose supply chest:', self.chest) self.state = self.path_to_supply_chest return print('No more supply-labeled chests') self.state = self.cleanup def path_to_supply_chest(self): print('Finding path to supply chest') p = utils.pint(self.g.pos) navpath = self.g.world.path_to_place_faked(p, self.chest) print('navpath:', navpath) if navpath: self.g.path = navpath[:-1] self.opening = self.g.path[-1] self.state = self.going_to_supply_chest self.checked_supplies = [] else: print('Supply chest is unreachable; trying another labeled chest') self.state = self.choose_supply_chest def going_to_supply_chest(self): if utils.pint(self.g.pos) == self.opening: self.g.look_at = self.area self.state = self.open_supply_chest def open_supply_chest(self): print('Opening supply chest') self.g.game.open_container(self.area) self.wait_time = 1 self.state = self.wait def wait(self): # wait for server to send us inventory contents if self.wait_time > 0: self.wait_time -= utils.TICK else: self.state = self.choose_items def choose_items(self): print('Selecting next item') for items, limits in self.supplies.items(): minimum_items, maximum_stacks = limits print('Checking items:', items) num_items = self.g.game.count_items(items) print('Have:', num_items) if num_items >= minimum_items: print('Have enough, skipping') continue if items in self.checked_supplies: print('Already checked, skipping') continue print('Need some') self.checked_supplies.append(items) self.target_items = items self.maximum_stacks = maximum_stacks self.count = 0 self.state = self.grab_items return print('No matching items in this supply chest') key = getattr(self, 'current_supply_chest_key', None) if key is not None and key not in self.useless_chests: self.useless_chests.append(key) self.state = self.close_supply_chest def grab_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 w.get('container_slots', 0) not in (27, 54): print('Got wrong container size, aborting') self.state = self.cleanup return used_slots = self.g.game.count_inventory_slots() print('used:', used_slots, 'total:', self.g.maximum_supply_slots) if used_slots >= self.g.maximum_supply_slots: print('Inventory is too full, aborting') self.g.game.close_window() self.g.look_at = None self.state = self.cleanup return for slot_num in range(w.container_slots): if slot_num not in w.contents: continue slot = w.contents[slot_num] if not slot.present: continue if slot.item_id not in self.target_items: continue if self.maximum_stacks and self.count >= self.maximum_stacks: break self.count += 1 print('Moving', slot) self.g.item_lock = True self.g.game.click_window(slot_num, 0, 1, slot) return print('None left to move') self.wait_time = 0.25 self.state = self.wait def close_supply_chest(self): print('Closing supply chest') self.g.game.close_window() self.g.look_at = None self.state = self.choose_supply_chest def cleanup(self): self.state = self.done def done(self): # never gets ran, placeholder return None def __init__(self, global_state): self.g = global_state self.state = self.idle self.supplies = self.default_supplies() self.supply_chests = [] self.checked_supply_chests = set() self.useless_chests = [] self._processed_chest_open_events = { self._chest_event_key(event) for event in self.g.get('chest_open_events', ()) } self.current_supply_chest_key = None self.supply_chest = None self.checked_supplies = [] self.target_items = None self.maximum_stacks = 0 self.count = 0 self.area = None self.opening = None self.wait_time = 0 def run(self): self._refresh_useless_chests() self.state()