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 mcdata from mosfet.info import mobs class SleepWithBedStates: def idle(self): return None def init(self): self.my_bed = False self.area = None self.opening = None self.bed_head = None self.beds = [] self.sleep_use_attempts = 0 self.waiting_for_night = False self.waiting_for_time_update = False current_time = self.g.get('time') if current_time is None or current_time < 12000: print('Aborting sleep, not night') self.state = self.cleanup return if self.g.get('dimension') != 'overworld': print('Aborting sleep, not in overworld') self.state = self.cleanup return self.state = self.find_beds @staticmethod def _bed_facing(state_id): props = blocks.PROPS.get(state_id, {}) return { 'north': (0, 0, -1), 'south': (0, 0, 1), 'east': (1, 0, 0), 'west': (-1, 0, 0), }.get(props.get('facing')) def _bed_foot(self, position): w = self.g.world state_id = w.block_at(*position) props = blocks.PROPS.get(state_id, {}) if props.get('part') != 'head': return position facing = self._bed_facing(state_id) if facing is None: return position foot = utils.psub(position, facing) foot_state = w.block_at(*foot) if (foot_state in blocks.BED_IDS and blocks.PROPS.get(foot_state, {}).get('part') == 'foot'): return foot return position def _bed_head(self, foot): state_id = self.g.world.block_at(*foot) facing = self._bed_facing(state_id) return utils.padd(foot, facing) if facing is not None else foot def find_beds(self): print('Finding beds...') w = self.g.world p = utils.pint(self.g.pos) if self.force_new_bed: result = [] print('Skipping existing beds; placing a new bed as requested') else: indexed = w.find_blocks_indexed(p, blocks.BED_IDS, 80) # Keep a tiny direct scan as a guard against stale/missing chunk # index entries, especially around respawn and chunk-cache changes. nearby = list(w.find_blocks_3d( p, blocks.BED_IDS, distance=8, y_limit=8)) result = list(dict.fromkeys(nearby + indexed)) result.sort(key=lambda bed: utils.phyp(p, bed)) if nearby and not indexed: print('Found beds by local scan; chunk index had none') result = list(dict.fromkeys(self._bed_foot(bed) for bed in result)) self.beds = [] for bed in result: if bed in self.bad_beds: continue if w.check_bed_occupied(bed): continue self.beds.append(bed) print('Found:', self.beds) self.state = self.choose_bed def choose_bed(self): print('Choosing a bed...') w = self.g.world p = utils.pint(self.g.pos) while self.beds: bed = self.beds.pop(0) print('Chose:', bed) navpath = w.path_to_place_faked(p, bed) print('navpath:', navpath) if navpath is None: self.bad_beds.append(bed) print('Cant get to bed, blacklisting') continue self.area = bed self.bed_head = self._bed_head(bed) self.opening = navpath[-2] if len(navpath) > 1 else p self.g.path = navpath[:-1] self.opening_deadline = time.monotonic() + max( 8.0, len(navpath) * 0.75) self.g.look_at = self.area self.state = self.going_to_bed return print('No reachable beds; trying to place one') self.state = self.select_bed def _at_opening(self): position = utils.pint(self.g.pos) return utils.phyp(position, self.opening) <= 1.25 def going_to_bed(self): if self._at_opening(): print('At existing bed') self.g.path = [] self.my_bed = False self.g.look_at = self.area self.wait_time = 0.5 self.state = self.turn_to_bed elif time.monotonic() >= self.opening_deadline: print('Could not reach bed opening', self.opening, '; trying another bed') self.bad_beds.append(self.area) self.state = self.choose_bed def turn_to_bed(self): self.g.look_at = self.area self.wait_time -= utils.TICK if self.wait_time <= 0: self.state = self.use_bed def select_bed(self): if self.g.game.select_item(items.BED_IDS): self.state = self.find_bed_spot else: print('No bed, aborting.') self.state = self.cleanup def find_bed_spot(self): print('Finding a bed spot...') w = self.g.world p = utils.pint(self.g.pos) for area in w.find_bed_areas(p, 100): 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_bed_openings(self.area) for o in openings: toward_player = (o[0] - self.area[0], 0, o[2] - self.area[2]) # The bed head points away from the player when placing from this # opening, so require that footprint and its support to be clear. head = (self.area[0] - toward_player[0], self.area[1], self.area[2] - toward_player[2]) if (w.block_at(*head) != blocks.AIR or w.block_at(*utils.padd(head, path.BLOCK_ABOVE)) != blocks.AIR): continue head_support = w.block_at(*utils.padd(head, path.BLOCK_BELOW)) if (head_support is None or head_support in blocks.NON_SOLID_IDS or head_support in blocks.AVOID_IDS): continue navpath = w.path_to_place(p, o) self.opening = o if navpath is not None: break else: # for print('Unable to get to bed area', self.area) self.bad_areas.append(self.area) self.state = self.cleanup return self.g.path = navpath self.opening_deadline = time.monotonic() + max( 8.0, len(navpath) * 0.75) self.g.look_at = self.area self.state = self.going_to_area def going_to_area(self): if self._at_opening(): print('At bed area') self.g.path = [] self.g.look_at = self.area self.wait_time = 0.5 self.state = self.turn_to_bed_area elif time.monotonic() >= self.opening_deadline: print('Could not reach bed placement opening', self.opening) self.bad_areas.append(self.area) self.state = self.cleanup def turn_to_bed_area(self): self.g.look_at = self.area self.wait_time -= utils.TICK if self.wait_time <= 0: self.state = self.place_bed def place_bed(self): support = utils.padd(self.area, path.BLOCK_BELOW) print('Placing bed at', self.area, 'on', support) self.g.game.place_block( support, BlockFace.TOP, hand=self.g.get('holding_hand', 0)) self.my_bed = True self.bed_place_deadline = time.monotonic() + 5.0 self.state = self.wait_use def wait_use(self): # Block updates can arrive after the use packet; poll authoritative # chunk state for several seconds instead of giving up after 0.5s. if self.g.world.block_at(*self.area) in blocks.BED_IDS: self.bed_head = self._bed_head(self.area) self.state = self.use_bed elif time.monotonic() >= self.bed_place_deadline: print('Bed placement was not confirmed at', self.area) self.bad_areas.append(self.area) self.my_bed = False self.state = self.cleanup def use_bed(self): w = self.g.world current_time = self.g.get('time') if current_time is None: if not getattr(self, 'waiting_for_time_update', False): print('Waiting for world time update before using bed') self.waiting_for_time_update = True return if current_time < 12550: if not getattr(self, 'waiting_for_night', False): print('Bed reached; waiting until sleep time (time={})'.format( current_time)) self.waiting_for_night = True return self.waiting_for_night = False self.waiting_for_time_update = False if w.block_at(*self.area) not in blocks.BED_IDS: print('Bed disappeared before use:', self.area) self.my_bed = False self.state = self.cleanup return if w.check_bed_occupied(self.area): print('Bed occupied, aborting.') self.state = self.cleanup return print('Using bed') # Beds are only 9/16 block tall. Aim inside their collision shape; # the near-top cursor used for placing a new block misses them. self.g.game.place_block( self.area, BlockFace.TOP, cursor_y=0.5) self.sleep_deadline = time.monotonic() + 5.0 self.sleep_retry_at = time.monotonic() + 1.5 self.sleep_use_attempts = 1 self.sleep_confirmed = False if not self.silent: self.g.chat.send('zzz') self.state = self.sleep_bed def sleep_bed(self): w = self.g.world p = utils.pint(self.g.pos) current_time = self.g.get('time') threats = w.find_threats(p, 10) if w.check_bed_occupied(self.area): self.sleep_confirmed = True if threats: print('Waking up due to threats:') print(threats) if self.sleep_confirmed: self.g.game.leave_bed() self.state = self.cleanup elif current_time is not None and current_time < 100: print('Woke up time') self.state = self.break_bed elif self.g.get('correction_count', 0) and self.sleep_confirmed: print('Woke up by movement') self.state = self.break_bed elif (not self.sleep_confirmed and self.sleep_use_attempts < 2 and time.monotonic() >= self.sleep_retry_at): print('Bed use not confirmed on foot; trying head block', self.bed_head) self.g.game.place_block( self.bed_head, BlockFace.TOP, cursor_y=0.5) self.sleep_use_attempts += 1 self.sleep_retry_at = time.monotonic() + 1.5 elif (not self.sleep_confirmed and time.monotonic() >= self.sleep_deadline): print('Bed use was not confirmed; blacklisting', self.area) self.bad_beds.append(self.area) self.state = self.cleanup def break_bed(self): if self.my_bed: if self.g.game.break_block(self.area): self.state = self.collect_bed else: print('Could not break the placed bed at', self.area) self.my_bed = False self.state = self.cleanup else: self.state = self.cleanup def collect_bed(self): if not self.g.breaking: self.g.path = [utils.padd(self.area, utils.spiral(n)) for n in range(9)] self.wait_time = 2 self.collect_deadline = time.monotonic() + 8 self.state = self.wait def wait(self): # Finish the short pickup sweep, but don't let it block forever. if (self.g.path and time.monotonic() < self.collect_deadline): return if self.wait_time > 0: self.wait_time -= utils.TICK else: self.state = self.cleanup def cleanup(self): if (self.my_bed and self.area and self.g.world.block_at(*self.area) in blocks.BED_IDS): self.state = self.break_bed return self.g.look_at = None self.g.path = [] self.my_bed = False self.force_new_bed = False self.state = self.done if (self.g.get('job') and getattr(self.g.job.state, '__name__', None) == 'sleep'): 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.my_bed = False self.beds = [] self.bad_beds = [] self.area = None self.opening = None self.bed_head = None self.bad_areas = [] self.wait_time = 0 self.sleep_use_attempts = 0 self.sleep_retry_at = 0 self.force_new_bed = False self.sleep_confirmed = False self.sleep_deadline = 0 def run(self): self.state()