891 lines
36 KiB
Python
891 lines
36 KiB
Python
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 _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
|
|
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
|