import collections import json import re import time import random from math import floor from itertools import count from copy import copy from mosfet import utils from mosfet import path from mosfet.info import blocks from mosfet.info import items from mosfet.info import mobs from minecraft.networking.types import BlockFace CHEST_OPENING_RADIUS = 3 class World: def __init__(self, global_state): self.g = global_state self._bed_occupancy_states = {} def block_at(self, x, y, z): return self.g.chunks.get_block_at(x, y, z) @staticmethod def _is_sign_state(state_id): name = blocks.BLOCKS.get(state_id) return name is not None and name.endswith('_sign') @staticmethod def _sign_component_text(value): if value is None: return '' if isinstance(value, str): stripped = value.strip() if stripped.startswith(('{', '[')): try: parsed = json.loads(stripped) except ValueError: return value if parsed != value: return World._sign_component_text(parsed) return value if isinstance(value, (list, tuple)): return ''.join(World._sign_component_text(part) for part in value) if isinstance(value, dict): text = value.get('text', '') if not text and value.get('translate'): text = value['translate'] if isinstance(text, (dict, list)): text = World._sign_component_text(text) return (str(text) + World._sign_component_text(value.get('with', [])) + World._sign_component_text(value.get('extra', []))) return str(value) def find_signs(self, center, distance=16): """Return nearby loaded signs as position/side/four-lines records.""" result = [] for position, entity in self.g.chunks.block_entities.items(): if utils.phyp(center, position) > distance: continue get_block_at = getattr(self.g.chunks, 'get_block_at', None) if callable(get_block_at): state_id = get_block_at(*position) if (state_id is not None and not self._is_sign_state(state_id)): continue data = entity.get('data') or {} for side in ('front', 'back'): text_data = data.get(side + '_text') if not isinstance(text_data, dict): continue messages = text_data.get('messages', []) lines = tuple(self._sign_component_text(line) for line in messages[:4]) lines += ('',) * (4 - len(lines)) if any(line.strip() for line in lines): result.append({ 'position': position, 'side': side, 'lines': lines, }) if not any(key in data for key in ('front_text', 'back_text')): old_lines = tuple(self._sign_component_text(data.get( 'Text{}'.format(index), '')) for index in range(1, 5)) if any(line.strip() for line in old_lines): result.append({ 'position': position, 'side': 'front', 'lines': old_lines, }) result.sort(key=lambda sign: utils.phyp(center, sign['position'])) return result def _double_chest_positions(self, chest): state_id = self.block_at(*chest) properties = blocks.PROPS.get(state_id, {}) chest_type = properties.get('type') if chest_type not in ('left', 'right'): return (tuple(chest),) other_type = 'right' if chest_type == 'left' else 'left' for offset in ((1, 0, 0), (-1, 0, 0), (0, 0, 1), (0, 0, -1)): partner = utils.padd(chest, offset) other = blocks.PROPS.get(self.block_at(*partner), {}) if (other.get('type') == other_type and other.get('facing') == properties.get('facing')): return tuple(sorted((tuple(chest), partner))) return (tuple(chest),) def find_labeled_chests(self, center, distance=80): """Find single/double chests assigned item categories by nearby signs.""" grouped = {} sign_offsets = ( (0, 0, 0), (0, -1, 0), (1, 0, 0), (-1, 0, 0), (0, 0, 1), (0, 0, -1), ) for sign in self.find_signs(center, distance): categories = set() labels = set() for line in sign['lines']: item_ids = items.storage_item_ids_for_label(line) if item_ids: categories.update(item_ids) labels.add(line.strip()) if not categories: continue sign_pos = sign['position'] adjacent = { utils.padd(sign_pos, offset) for offset in sign_offsets if self.block_at(*utils.padd(sign_pos, offset)) in blocks.CHEST_IDS } if not adjacent: nearby = self.find_blocks_indexed( sign_pos, blocks.CHEST_IDS, 2) if nearby: nearest = min(utils.phyp(sign_pos, chest) for chest in nearby) adjacent = {chest for chest in nearby if utils.phyp(sign_pos, chest) == nearest} for chest in adjacent: positions = self._double_chest_positions(chest) record = grouped.setdefault(positions, { 'position': tuple(chest), 'positions': positions, 'item_ids': set(), 'labels': set(), 'sign_positions': set(), }) record['item_ids'].update(categories) record['labels'].update(labels) record['sign_positions'].add(sign_pos) result = list(grouped.values()) result.sort(key=lambda chest: utils.phyp(center, chest['position'])) return result def find_supply_chests(self, center, bot_name, distance=80): """Find chests whose sign is addressed to this bot's supplies.""" bot_name = str(bot_name).casefold() group_name = (bot_name[:-1] if len(bot_name) > 1 and bot_name[-1] in '123' else None) addresses = {bot_name} if group_name: addresses.add(group_name) grouped = {} sign_offsets = ( (0, 0, 0), (0, -1, 0), (1, 0, 0), (-1, 0, 0), (0, 0, 1), (0, 0, -1), ) for sign in self.find_signs(center, distance): text = ' '.join(line for line in sign['lines'] if line.strip()) tokens = re.findall(r'[a-z0-9_]+', text.casefold()) if (len(tokens) < 2 or tokens[0] not in addresses or 'supplies' not in tokens[1:]): continue sign_pos = sign['position'] candidates = { utils.padd(sign_pos, offset) for offset in sign_offsets if self.block_at(*utils.padd(sign_pos, offset)) in blocks.CHEST_IDS } if not candidates: nearby = self.find_blocks_indexed( sign_pos, blocks.CHEST_IDS, 2) if nearby: nearest = min(utils.phyp(sign_pos, chest) for chest in nearby) candidates = {chest for chest in nearby if utils.phyp(sign_pos, chest) == nearest} for chest in candidates: positions = self._double_chest_positions(chest) record = grouped.setdefault(positions, { 'position': tuple(chest), 'positions': positions, 'labels': set(), 'sign_positions': set(), }) record['labels'].update(tokens[1:]) record['sign_positions'].add(sign_pos) result = list(grouped.values()) result.sort(key=lambda chest: utils.phyp(center, chest['position'])) return result def sign_policies(self, bot_name, distance=64): """Parse sign policies addressed to this bot or its numbered group.""" bot_name = str(bot_name).casefold() group_name = (bot_name[:-1] if len(bot_name) > 1 and bot_name[-1] in '123' else None) addresses = {bot_name} if group_name: addresses.add(group_name) center = utils.pint(self.g.pos) policies = [] active_sign_position = None for sign in self.find_signs(center, distance): lines = [line.strip() for line in sign['lines'] if line.strip()] if not lines: continue first = lines[0].split(maxsplit=1) address = first[0].rstrip(':,').casefold() if address not in addresses: continue directives = ([first[1]] if len(first) > 1 else []) + lines[1:] text = ' '.join(directives).casefold() sign_policies = [] for match in re.finditer( r'\bstay\s+within\s*(\d+)(?:\s+(squares?|circle))?\b', text): shape = match.group(2) or 'circle' if shape == 'squares': shape = 'square' sign_policies.append({ 'position': sign['position'], 'distance': int(match.group(1)), 'shape': shape, 'side': sign['side'], }) if not sign_policies: continue if active_sign_position is None: # The nearest addressed policy sign overrides farther signs. active_sign_position = sign['position'] if sign['position'] == active_sign_position: policies.extend(sign_policies) return policies def _active_work_policies(self): job = self.g.get('job') machine = getattr(job, 'active_machine', None) if job else None if machine is None: return [] support_machines = ( 'grab_supplies_states', 'sleep_with_bed_states', 'eat_food_states', 'cache_items_states', 'check_threats_states', ) if any(machine is getattr(job, name, None) for name in support_machines): return [] return self.sign_policies(self.g.get('name', '')) @staticmethod def _sign_policy_distance(position, policy): # Work zones are horizontal; crop height or terrain elevation does not # consume the sign's radius. dx = abs(position[0] - policy['position'][0]) dz = abs(position[2] - policy['position'][2]) if policy['shape'] == 'square': return max(dx, dz) return utils.hypot(dx, dz) @classmethod def _inside_sign_policy(cls, position, policy): return cls._sign_policy_distance(position, policy) <= policy['distance'] def work_position_allowed(self, position): """Whether a work target satisfies all currently active sign rules.""" return all(self._inside_sign_policy(position, policy) for policy in self._active_work_policies()) def filter_work_positions(self, positions): """Filter a candidate set once against the active sign policies.""" policies = self._active_work_policies() positions = list(positions) if not policies: return positions return [position for position in positions if all(self._inside_sign_policy(position, policy) for policy in policies)] @classmethod def _path_allowed_by_sign(cls, points, policy): distances = [cls._sign_policy_distance(point, policy) for point in points] radius = policy['distance'] if distances[0] <= radius: return all(distance <= radius for distance in distances) # If the bot starts just outside a newly read boundary, let it return # without allowing it to travel farther away from that sign. start_distance = distances[0] entered = False for distance in distances[1:]: if distance > start_distance or (entered and distance > radius): return False if distance <= radius: entered = True return entered def check_air_column(self, pos, distance): for i in range(distance): check = utils.padd(pos, (0, i, 0)) if self.block_at(*check) not in blocks.NON_SOLID_IDS: return False return True def find_blocks_3d(self, center, block_ids, distance=0, y_limit=0, thru_air=False): to_visit = collections.deque([(0, 0, 0)]) visited = set() while to_visit: cur = to_visit.pop() if cur in visited: continue if y_limit and abs(cur[1]) > y_limit: continue if distance and utils.hypot(*cur) > distance: continue check = utils.padd(center, cur) if not thru_air or self.block_at(*check) in blocks.NON_SOLID_IDS: for neighbor in utils.get_neighbors_3d(*cur): to_visit.appendleft(neighbor) visited.add(cur) if self.block_at(*check) in block_ids: yield check def find_blocks_indexed(self, center, block_ids, distance=0): index = [] for bid in block_ids: index.extend(self.g.chunks.index.get(bid, [])) result = [] for block in index: if self.block_at(*block) not in block_ids: continue if distance and utils.phyp(center, block) > distance: continue if block not in result: result.append(block) result.sort(key=lambda x: utils.phyp(center, x)) return result def find_blocks(self, center, distance, block_ids, limit=0): # search in a spiral from center to all blocks with ID result = [] for n in count(): offset = utils.spiral(n) check = utils.padd(center, offset) if self.block_at(*check) in block_ids: if utils.hypot(*offset) < distance: result.append(check) if limit and len(result) == limit: return result if offset[0] > distance: return result def find_trees(self, center, distance): found_trees = set() # 26.3 block-state caches keep an index of tree logs. Searching a # three-dimensional radius voxel-by-voxel is prohibitively expensive # and can walk through unloaded space, so use loaded indexed states. for candidate in self.find_blocks_indexed( center, blocks.LOG_IDS, distance): # Crawl to the bottom log of this trunk. log = candidate while self.block_at(*utils.padd(log, path.BLOCK_BELOW)) in blocks.LOG_IDS: log = utils.padd(log, path.BLOCK_BELOW) base = log if base in found_trees: continue # Only classify trees when their supporting block is known. support = self.block_at(*utils.padd(base, path.BLOCK_BELOW)) if support is None or support in blocks.NON_SOLID_IDS: continue # Crawl to the top log to count trunk blocks and check for leaves. log_count = 1 good_leaves = False while self.block_at(*utils.padd(log, path.BLOCK_ABOVE)) in blocks.LOG_IDS: log = utils.padd(log, path.BLOCK_ABOVE) log_count += 1 for offset in path.CHECK_DIRECTIONS: if self.block_at(*utils.padd(log, offset)) in blocks.LEAF_IDS: good_leaves = True if not good_leaves or log_count < 3: continue found_trees.add(base) yield base def find_tree_openings(self, tree): # returns coords in a cardinal direction where we can stand by tree maze_solver = path.Pathfinder(self.g) result = [] # TODO: make sure only non-solid and leaves between # make sure traversable too and non-avoid for distance in range(5): for direction in path.CHECK_DIRECTIONS: offset = utils.pmul(direction, distance+1) if maze_solver.check_traverse(tree, offset): result.append(utils.padd(tree, offset)) return result def path_to_place(self, start, place): maze_solver = path.Pathfinder(self.g) try: solution = maze_solver.astar(start, place) if not solution: return None navpath = list(solution) except path.AStarTimeout: return None policies = self._active_work_policies() points = [tuple(start)] + [tuple(point) for point in navpath] for policy in policies: if not self._path_allowed_by_sign(points, policy): print('[sign policy] rejected work path outside {}-block {} ' 'around {}'.format(policy['distance'], policy['shape'], policy['position']), flush=True) return None return navpath def path_to_place_faked(self, start, place): # same as above, but adds a fake block below and air before pathfinding # so that the pathfinder can actually make it to the block c = self.g.chunks above = utils.padd(place, path.BLOCK_ABOVE) below = utils.padd(place, path.BLOCK_BELOW) tmp = c.get_block_at(*place) tmp2 = c.get_block_at(*above) tmp3 = c.get_block_at(*below) c.set_block_at(*place, blocks.AIR) c.set_block_at(*above, blocks.AIR) c.set_block_at(*below, blocks.STONE) navpath = self.path_to_place(start, place) c.set_block_at(*place, tmp) c.set_block_at(*above, tmp2) c.set_block_at(*below, tmp3) return navpath def find_bed_areas(self, center, distance): """Yield nearby two-block bed footprints on solid, safe ground.""" # Only search loaded-neighborhood distance. The old 3-D flood fill # explored millions of air blocks and used overly broad 5x5 clearance. distance = min(distance, 32) headroom = path.BLOCK_ABOVE support = path.BLOCK_BELOW for offset in utils.search_2d(distance): if utils.hypot(offset[0], offset[2]) > distance: continue for dy in (0, 1, -1, 2, -2): area = (center[0] + offset[0], center[1] + dy, center[2] + offset[2]) if area == center: continue if self.block_at(*area) != blocks.AIR: continue if self.block_at(*utils.padd(area, headroom)) != blocks.AIR: continue support_block = self.block_at(*utils.padd(area, support)) if (support_block is None or support_block in blocks.NON_SOLID_IDS or support_block in blocks.AVOID_IDS): continue footprint_found = False for direction in path.CHECK_DIRECTIONS: head = utils.padd(area, direction) if (self.block_at(*head) != blocks.AIR or self.block_at(*utils.padd(head, headroom)) != blocks.AIR): continue head_support = self.block_at(*utils.padd(head, support)) if (head_support is None or head_support in blocks.NON_SOLID_IDS or head_support in blocks.AVOID_IDS): continue footprint_found = True break if not footprint_found: continue # Avoid placing a bed within range of nearby monsters. if self.find_monsters(area, 20): continue yield area def find_cache_areas(self, center, distance): """Yield nearby, supported air blocks suitable for placing a chest.""" center = tuple(center) for offset in utils.search_2d(distance): if utils.hypot(offset[0], offset[2]) > distance: continue for dy in (0, 1, -1, 2, -2): area = (center[0] + offset[0], center[1] + dy, center[2] + offset[2]) if area == center: continue if self.block_at(*area) != blocks.AIR: continue if self.block_at(*utils.padd(area, path.BLOCK_ABOVE)) != blocks.AIR: continue support = self.block_at(*utils.padd(area, path.BLOCK_BELOW)) if (support is None or support in blocks.NON_SOLID_IDS or support in blocks.AVOID_IDS): continue yield area def sand_adjacent_safe(self, sand): for direction in path.CHECK_DIRECTIONS: if self.block_at(*utils.padd(sand, direction)) in blocks.AVOID_IDS: return False return True def find_sand(self, center, distance, player): sand = [] sand.extend(self.find_blocks(center, distance, [blocks.SAND], 25)) safe_sand = [] for s in sand: # make sure it has solid below if self.block_at(*utils.padd(s, path.BLOCK_BELOW)) in blocks.NON_SOLID_IDS: continue # make sure it has solid two below - prevent hanging sand if self.block_at(*utils.padd(s, path.BLOCK_BELOW2)) in blocks.NON_SOLID_IDS: continue # and walkable air above if self.block_at(*utils.padd(s, path.BLOCK_ABOVE)) not in blocks.NON_SOLID_IDS: continue if not self.sand_adjacent_safe(s): continue safe_sand.append(s) safe_sand.sort(key=lambda x: utils.phyp(player, x)) return safe_sand def check_sand_slice(self, center): # checks if a 5x5x1 slice has sand in it for i in range(9): s = utils.padd(center, utils.spiral(i)) if self.block_at(*s) != blocks.SAND: continue # make sure it has solid below if self.block_at(*utils.padd(s, path.BLOCK_BELOW)) in blocks.NON_SOLID_IDS: continue # make sure it has solid two below - prevent hanging sand if self.block_at(*utils.padd(s, path.BLOCK_BELOW2)) in blocks.NON_SOLID_IDS: continue # and walkable air above if self.block_at(*utils.padd(s, path.BLOCK_ABOVE)) not in blocks.NON_SOLID_IDS: continue if not self.sand_adjacent_safe(s): continue return True return False def find_sand_slice(self, center, distance, y_limit=0, bad_slices=[], prev_layer=0): # returns the centre coord of the next 5x5x1 slice that still has # diggable sand in it. lower slices are only valid if there's an # adjacent slice farther at the same level. this should ensure an # upside down pyramid gets excavated so the edges are still climbable for v in count(prev_layer): peak = utils.padd(center, (0, 10-v, 0)) slices = [] layer = 0 for step in count(): offset = utils.spiral(step) layer = max(layer, *offset) offset = utils.pmul(offset, 3) check = utils.padd(peak, offset) check = utils.padd(check, (0, layer, 0)) if y_limit and check[1] - center[1] > y_limit: break if utils.phyp_king(center, check) > distance: break if self.check_sand_slice(check) and check not in bad_slices: slices.append(check) if len(slices): return v, slices[-1] elif v > 40: return None, None def find_bed_openings(self, area): # returns coords in a cardinal direction where we can stand by bed result = [] for direction in path.CHECK_DIRECTIONS: result.append(utils.padd(area, direction)) return result def check_bed_occupied(self, bed): """Return occupancy, logging only the first observation or a change.""" bed = tuple(bed) bid = self.g.chunks.get_block_at(*bed) occupied = blocks.PROPS.get(bid, {}).get('occupied') == 'true' previous = self._bed_occupancy_states.get(bed) if previous is None or previous != occupied: print('Checking bed occupancy:', bed, '->', 'occupied' if occupied else 'free') self._bed_occupancy_states[bed] = occupied return occupied @staticmethod def _is_chest_state(state_id): name = blocks.BLOCKS.get(state_id) return (name is not None and name != 'ender_chest' and (name == 'chest' or name.endswith('_chest'))) def _cache_opening_standable(self, position): at_feet = self.block_at(*position) above = self.block_at(*utils.padd(position, path.BLOCK_ABOVE)) below = self.block_at(*utils.padd(position, path.BLOCK_BELOW)) return (at_feet in blocks.NON_SOLID_IDS and at_feet not in blocks.AVOID_IDS and above in blocks.NON_SOLID_IDS and above not in blocks.AVOID_IDS and below is not None and below not in blocks.NON_SOLID_IDS and below not in blocks.AVOID_IDS) @staticmethod def _segment_intersects_aabb(start, end, minimum, maximum): lower, upper = 0.0, 1.0 for axis in range(3): origin = start[axis] delta = end[axis] - origin if abs(delta) < 1e-9: if origin < minimum[axis] or origin > maximum[axis]: return False continue first = (minimum[axis] - origin) / delta second = (maximum[axis] - origin) / delta if first > second: first, second = second, first lower = max(lower, first) upper = min(upper, second) if lower > upper: return False return upper >= 0.0 and lower <= 1.0 def _has_cache_line_of_sight(self, start, end, target): first = tuple(floor(min(start[i], end[i])) for i in range(3)) last = tuple(floor(max(start[i], end[i])) for i in range(3)) for x in range(first[0], last[0] + 1): for y in range(first[1], last[1] + 1): for z in range(first[2], last[2] + 1): position = (x, y, z) if position == target: continue if not self._segment_intersects_aabb( start, end, position, utils.padd(position, (1, 1, 1))): continue state_id = self.block_at(*position) if state_id is None: return False for box in blocks.COLLISION_BOXES.get(state_id, ()): minimum = (x + box[0], y + box[1], z + box[2]) maximum = (x + box[3], y + box[4], z + box[5]) if self._segment_intersects_aabb( start, end, minimum, maximum): return False return True @staticmethod def _chest_interaction_points(target, boxes): points = [] for box in boxes: x0, y0, z0, x1, y1, z1 = box xm, zm = (x0 + x1) / 2, (z0 + z1) / 2 for fraction in (0.2, 0.5, 0.8): y = y0 + (y1 - y0) * fraction points.extend(( (target[0] + xm, target[1] + y, target[2] + z0), (target[0] + xm, target[1] + y, target[2] + z1), (target[0] + x0, target[1] + y, target[2] + zm), (target[0] + x1, target[1] + y, target[2] + zm), )) for x_fraction, z_fraction in ( (0.25, 0.25), (0.25, 0.75), (0.75, 0.25), (0.75, 0.75), (0.5, 0.5)): points.append(( target[0] + x0 + (x1 - x0) * x_fraction, target[1] + y1, target[2] + z0 + (z1 - z0) * z_fraction, )) return points @staticmethod def _ray_chest_surface_hit(start, end, target, boxes): negative_faces = (BlockFace.WEST, BlockFace.BOTTOM, BlockFace.NORTH) positive_faces = (BlockFace.EAST, BlockFace.TOP, BlockFace.SOUTH) delta = tuple(end[i] - start[i] for i in range(3)) hits = [] for box in boxes: minimum = tuple(target[i] + box[i] for i in range(3)) maximum = tuple(target[i] + box[i + 3] for i in range(3)) enter, leave = float('-inf'), float('inf') entering_face = None for axis in range(3): if abs(delta[axis]) < 1e-9: if (start[axis] < minimum[axis] or start[axis] > maximum[axis]): break continue near = (minimum[axis] - start[axis]) / delta[axis] far = (maximum[axis] - start[axis]) / delta[axis] face = negative_faces[axis] if near > far: near, far = far, near face = positive_faces[axis] if near > enter: enter = near entering_face = face leave = min(leave, far) if enter > leave: break else: if entering_face is None or enter < 0 or enter > 1: continue hit_point = tuple(start[i] + delta[i] * enter for i in range(3)) cursor = tuple(hit_point[i] - target[i] for i in range(3)) hits.append((enter, entering_face, cursor, hit_point)) if not hits: return None _distance, face, cursor, hit_point = min(hits, key=lambda hit: hit[0]) return face, cursor, hit_point def find_cache_openings(self, area, center=None, radius=None): """Find standable nearby spots with a clear ray to a chest surface.""" area = tuple(area) center = tuple(center if center is not None else area) radius = CHEST_OPENING_RADIUS if radius is None else radius state_id = self.block_at(*area) targets = (self._double_chest_positions(area) if self._is_chest_state(state_id) else (area,)) default_chest = min(blocks.CHEST_IDS) interactions = {} for target in targets: target_state = self.block_at(*target) boxes = blocks.COLLISION_BOXES.get(target_state) or \ blocks.COLLISION_BOXES[default_chest] hit_points = self._chest_interaction_points(target, boxes) for horizontal in utils.search_2d(radius + 1): for dy in range(-radius - 1, radius + 2): offset = (horizontal[0], dy, horizontal[2]) opening = utils.padd(target, offset) if (opening in targets or not self._cache_opening_standable(opening)): continue eye = (opening[0] + 0.5, opening[1] + 1.62, opening[2] + 0.5) for aim_point in hit_points: if not self._has_cache_line_of_sight( eye, aim_point, target): continue hit = self._ray_chest_surface_hit( eye, aim_point, target, boxes) if hit is None: continue face, cursor, hit_point = hit hit_distance = utils.phyp(eye, hit_point) if hit_distance > radius: continue interaction = { 'position': opening, 'target': target, 'face': face, 'cursor': cursor, 'hit_point': hit_point, 'look_at': tuple(value - 0.5 for value in hit_point), } distance = utils.phyp(center, opening) interaction['distance'] = distance interaction['hit_distance'] = hit_distance previous = interactions.get(opening) if (previous is None or interaction['hit_distance'] < previous['hit_distance']): interactions[opening] = interaction break return sorted(interactions.values(), key=lambda result: (result['distance'], result['position'])) def find_objects(self, object_ids): result = [] for eid, obj in copy(self.g.objects).items(): if obj.get('item_id', None) in object_ids: result.append(obj) return result def find_leaves(self, center, distance): for a in self.find_blocks_3d(center, blocks.LEAF_IDS, distance, 10): yield a def find_monsters(self, center, distance): # finds monsters within distance result = [] for eid, mob in copy(self.g.mobs).items(): if mob.type not in mobs.EVIL_IDS: continue pos = utils.pint((mob.x, mob.y, mob.z)) if utils.phyp(center, pos) > distance: continue result.append(mob) result.sort(key=lambda mob: utils.phyp(center, (mob.x, mob.y, mob.z))) return result def find_threats(self, center, distance): # finds monsters on the surface within distance monsters = self.find_monsters(center, distance) result = [] for mob in monsters: pos = utils.pint((mob.x, mob.y, mob.z)) # check distance number of blocks above, close enough? if not self.check_air_column(pos, distance): continue result.append(mob) result.sort(key=lambda mob: utils.phyp(center, (mob.x, mob.y, mob.z))) return result def find_villagers(self, center, distance): # finds villagers within distance result = [] for eid, mob in copy(self.g.mobs).items(): type_name = mobs.MOB_NAMES[mob.type] if type_name != 'villager' : continue pos = utils.pint((mob.x, mob.y, mob.z)) if utils.phyp(center, pos) > distance: continue result.append(mob) result.sort(key=lambda mob: utils.phyp(center, (mob.x, mob.y, mob.z))) return result def find_villager_openings(self, villager): # returns coords in a cardinal direction where we can stand by a villager maze_solver = path.Pathfinder(self.g) result = [] for distance in range(3): for direction in path.CHECK_DIRECTIONS: offset = utils.pmul(direction, distance+1) if not maze_solver.check_traverse(villager, offset): continue # check for line of sight for check in range(distance+1): offset2 = utils.pmul(direction, check+1) offset2 = utils.padd(offset2, path.BLOCK_ABOVE) check = utils.padd(villager, offset2) if self.block_at(*check) not in blocks.NON_SOLID_IDS: break else: # for result.append(utils.padd(villager, offset)) return result