tanner/minecraft-bot
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
Files
be5fe260e3e9bf267f0b3e4598bc5fba280eb564
minecraft-bot/mosfet/bot.py
T

599 lines
19 KiB
Python
Raw Blame History

if __name__ == '__main__':
print('Run mosfet.py instead.')
exit(1)
import os
import sys
import time
import importlib
import threading
import traceback
from math import floor, ceil
from copy import copy
from . import monkey_patch # must be before any possible pyCraft imports
from minecraft import authentication
from minecraft.exceptions import YggdrasilError
from minecraft.networking.connection import Connection
from minecraft.networking.packets import Packet, clientbound, serverbound
from mosfet.protocol.managers import DataManager, ChunksManager, ChatManager, ChunkNotLoadedException
from munch import Munch
from mosfet import commands
from mosfet import game
from mosfet import job
from mosfet import path
from mosfet import print_help
from mosfet import utils
from mosfet import vector
from mosfet import world
from mosfet.info import blocks
from mosfet.info import items
from mosfet.info import mcdata
from mosfet.info import mobs
for module in [
blocks,
commands,
game,
items,
job,
mcdata,
mobs,
path,
print_help,
utils,
vector,
world,
]:
importlib.reload(module)
last_tick = time.time()
PITCH_ANGLE_DIR = vector.Vector3D((0, 1, 0))
YAW_ANGLE_DIR = vector.Vector3D((0, 0, -1))
YAW_ANGLE_REF = vector.Vector3D((0, 1, 0))
YAW_LOOK_AHEAD = 4
PLAYER_EYE_HEIGHT = 1.62
MAX_DOWNWARD_PITCH = 45.0
def look_direction_from_eye(position, target):
eye = vector.Point3D((position.x, position.y + PLAYER_EYE_HEIGHT,
position.z))
return eye - target
def update_pitch(g, look_at_d):
if look_at_d is None:
target_pitch = 0.0
elif look_at_d.length() > 0.6:
target_pitch = look_at_d.normalized().angleDeg(PITCH_ANGLE_DIR)
target_pitch = (target_pitch - 90) * -1
else:
return
target_pitch = min(target_pitch, MAX_DOWNWARD_PITCH)
g.pitch += utils.cap(target_pitch - g.pitch, 10)
def missing_movement_chunks(g):
"""Return chunks needed for local physics and the next path steps only."""
position = g.get('pos')
if position is None or position is False:
return [(None, None)]
points = [position]
points.extend(g.path[:2])
required = set()
for point in points:
if hasattr(point, 'x'):
x, z = point.x, point.z
else:
x, z = point[0], point[2]
required.add((floor(x / 16), floor(z / 16)))
return sorted((x, z) for x, z in required
if g.chunks.get_chunk(x, z) is None)
def tick(global_state):
g = global_state
p = g.pos
target = None
# Require only chunks touched by local physics and the next path steps.
# A few unrelated holes in the wider view cache must not stall jobs.
missing_chunks = missing_movement_chunks(g)
if missing_chunks:
if g.get('movement_status') != 'waiting_for_chunks':
print('[movement] status=waiting_for_chunks near {}'.format(
utils.pint(p)), flush=True)
g.movement_status = 'waiting_for_chunks'
if not g.chunks.loading:
print('Loading chunks', end='', flush=True)
g.chunks.loading = time.time()
now = time.time()
if now - g.get('chunk_wait_report_at', 0) >= 5:
center = g.chunks.center
print('\n[movement] chunk wait: pos={}, center={}, '
'loaded={}, missing_required={}'.format(
utils.pint(p), g.chunks.center,
len(g.chunks.chunks), missing_chunks), flush=True)
g.chunk_wait_report_at = now
from minecraft.networking.modern import (
ClientTickEnd, PlayerPositionRotation,
)
packet = PlayerPositionRotation(
x=p.x, y=p.y, z=p.z, pitch=0, yaw=0, on_ground=True)
g.connection.write_packet(packet, force=True)
g.connection.write_packet(ClientTickEnd(), force=True)
return
else:
if g.chunks.loading:
print()
print('Chunks loaded in', round(time.time() - g.chunks.loading, 2), 's')
g.chunks.loading = False
g.chunk_wait_report_at = 0
status = ('draining_inventory' if g.get('draining') else
'following_path' if g.path else 'running_job')
if g.get('movement_status') != status:
print('[movement] status={}'.format(status), flush=True)
g.movement_status = status
########## object physics ##########
# note: it's possible the chunk data is out of date when this runs
for eid, obj in copy(g.objects).items():
if obj.velocity_x:
obj.x += obj.velocity_x / 8000
if obj.velocity_y:
obj.y += obj.velocity_y / 8000
if obj.velocity_z:
obj.z += obj.velocity_z / 8000
block_below = g.chunks.get_block_at(floor(obj.x), floor(obj.y-0.20), floor(obj.z))
in_air = block_below in blocks.NON_SOLID_IDS
if in_air:
obj.velocity_x *= 0.988
obj.velocity_y -= 390
obj.velocity_z *= 0.988
else:
obj.y = int(obj.y-0.20)+1
obj.velocity_x *= 0.5
obj.velocity_y = 0
obj.velocity_z *= 0.5
# float object back up in case it clipped through multiple blocks
if g.chunks.get_block_at(floor(obj.x), floor(obj.y), floor(obj.z)) not in blocks.NON_SOLID_IDS:
obj.y += 0.05
if abs(obj.velocity_x) < 1: obj.velocity_x = 0
if abs(obj.velocity_z) < 1: obj.velocity_z = 0
########## player physics ##########
if g.path and len(g.path):
target = vector.Point3D(g.path[0])
target.x += 0.5
target.z += 0.5
if g.afk_timeout > 0:
target = None
g.afk_timeout -= utils.TICK
if target:
d = p - target
# jump up block
if d.y < -0.9 and not g.y_v:
g.y_v = 8.5
g.y_a = -36.0
# jump gap
if d.xz.length() > 1.6 and not g.y_v:
g.y_v = 8.5
g.y_a = -36.0
if d.length() > 0:
if g.y_v < 5:
p.x -= utils.cap(d.x, 0.2)
p.z -= utils.cap(d.z, 0.2)
if len(g.path) > 1 and d.length() < 0.2:
# removes some jitter in walking
g.path.pop(0)
elif d.length() == 0:
g.path.pop(0)
if g.y_v or g.y_a:
p.y += g.y_v * utils.TICK
g.y_v += g.y_a * utils.TICK
block_below = g.chunks.get_block_at(floor(p.x), ceil(p.y-1), floor(p.z))
block_above = g.chunks.get_block_at(floor(p.x), ceil(p.y+1), floor(p.z))
in_void = p.y < 0
in_air = block_below in blocks.NON_SOLID_IDS or in_void
in_water = block_below in blocks.WATER_IDS
g.crawling = block_above not in blocks.NON_SOLID_IDS
if in_air:
g.y_a = -36.0
elif in_water:
g.y_a = -16.0
else:
p.y = ceil(p.y)
g.y_v = 0
g.y_a = 0
if g.look_at:
look_at = vector.Point3D(g.look_at)
elif g.path and len(g.path) > YAW_LOOK_AHEAD:
look_at = vector.Point3D(g.path[YAW_LOOK_AHEAD])
elif g.path and len(g.path):
look_at = vector.Point3D(g.path[-1])
else:
look_at = None
if look_at:
look_at.x += 0.5
look_at.y += 0.5
look_at.z += 0.5
look_at_d = p - look_at
update_pitch(g, look_direction_from_eye(p, look_at))
# remove vertical component for yaw calculation
look_at_d.y = 0
if look_at_d.length() > 0.6:
target_yaw = look_at_d.normalized().signedAngleDeg(other=YAW_ANGLE_DIR, ref=YAW_ANGLE_REF)
target_yaw_d = target_yaw - g.yaw
target_yaw_d = (target_yaw_d + 180) % 360 - 180
g.yaw += utils.cap(target_yaw_d, 30)
else:
update_pitch(g, None)
from minecraft.networking.modern import (
ClientTickEnd, PlayerPositionRotation,
)
packet = PlayerPositionRotation(
x=p.x, y=p.y, z=p.z, pitch=g.pitch, yaw=g.yaw,
on_ground=(not in_air))
# Keep the client-tick boundary after its movement packet on the wire.
g.connection.write_packet(packet, force=True)
g.connection.write_packet(ClientTickEnd(), force=True)
if g.job:
g.job.tick()
if g.game:
g.game.tick() # order important for correction_count
def init(global_state):
g = global_state
g.time = 0
g.path = []
g.look_at = None
g.y_v = 0
g.y_a = 0
g.yaw = 360
g.pitch = 0
g.crawling = False
g.breaking = None
g.break_time = 0
g.dumping = None
g.dump_waiting = None
g.dump_dropped = 0
g.dump_total = 0
g.dump_error = None
g.drop_waiting = None
g.draining = False
g.item_lock = False
g.command_lock = False
g.trades = []
g.job = job.JobStates(g)
g.chopped_tree = False
g.afk_timeout = 0
g.filling = False
g.minimum_cache_slots = 27
g.maximum_supply_slots = 33
def bot(global_state):
EMAIL = os.getenv('EMAIL')
PASSWORD = os.getenv('PASSWORD')
SERVER = os.getenv('SERVER')
PORT = int(os.environ.get('PORT', 25565))
g = global_state
if not g.mcdata:
g.mcdata = DataManager('./minecraft_data')
if not SERVER:
print()
print('You must specify a server to connect to. For example:')
print('SERVER=minecraft.example.com python mosfet.py')
print('SERVER=localhost PORT=12345 python mosfet.py')
os._exit(0)
elif not g.connection:
if EMAIL and PASSWORD:
auth_token = authentication.AuthenticationToken()
try:
auth_token.authenticate(EMAIL, PASSWORD)
except YggdrasilError as e:
print(e)
os._exit(0)
print("Logged in as %s..." % auth_token.username)
g.connection = Connection(SERVER, PORT, auth_token=auth_token)
elif EMAIL:
print('No password provided, attempting to connect in offline mode...')
g.connection = Connection(SERVER, PORT, username=EMAIL)
elif PASSWORD:
print('')
print('Did you forget to specify an email?')
print('If you want to use your own account:')
print('Set SERVER and run python mosfet.py.')
os._exit(0)
else:
print('No username or password provided, using burner minecraft account...')
EMAIL = 'moc.liamg@monortem'[::-1]
PASSWORD = '!8891anteR'[::-1]
auth_token = authentication.AuthenticationToken()
try:
auth_token.authenticate(EMAIL, PASSWORD)
except YggdrasilError as e:
print(e)
os._exit(0)
print("Logged in as %s..." % auth_token.username)
g.connection = Connection(SERVER, PORT, auth_token=auth_token)
g.chunks = ChunksManager(g.mcdata)
g.connection.connect()
g.chunks.register(g.connection)
g.chat = ChatManager(g)
g.game = game.Game(g)
g.world = world.World(g)
g.commands = commands.Commands(g)
try:
while not g.pos:
time.sleep(utils.TICK)
print('Player loaded.')
init(g)
g.game.close_window()
print('Initialized.')
while g.running:
tick(g)
global last_tick
sleep_time = utils.TICK + last_tick - time.time()
if sleep_time < 0: sleep_time = 0
time.sleep(sleep_time)
last_tick = time.time()
finally:
print('Removing listeners...')
g.connection.packet_listeners = []
g.connection.early_packet_listeners = []
g.connection.outgoing_packet_listeners = []
g.connection.early_outgoing_packet_listeners = []
print('Bot module loaded.')
def start_automation(global_state, initialize=True):
"""Start or replace the movement worker without replacing the connection."""
g = global_state
g.protocol_version = 777
if initialize:
g.time = 0.0
g.path = []
g.look_at = None
g.y_v = 0.0
g.y_a = 0.0
g.crawling = False
g.afk_timeout = 0.0
g.objects = g.get('objects') or {}
g.players = g.get('players') or {}
g.breaking = None
g.break_time = 0.0
g.item_lock = False
g.minimum_cache_slots = g.get('minimum_cache_slots', 27)
g.maximum_supply_slots = g.get('maximum_supply_slots', 33)
g.draining = False
g.dumping = None
g.dump_waiting = None
g.dump_dropped = 0
g.dump_total = 0
g.dump_error = None
g.drop_waiting = None
g.drain_slots = []
g.drain_waiting = None
g.drain_dropped = 0
g.drain_total = 0
g.drain_error = None
g.correction_count = g.get('correction_count', 0)
g.yaw = float(g.get('yaw', 0.0) or 0.0)
g.pitch = float(g.get('pitch', 0.0) or 0.0)
g.holding_slot = g.get('holding_slot', 0)
g.holding_hand = 0
g.chunks.loading = False
# Readiness below is based on nearby chunks, not unrelated view holes.
g.movement_paused = False
g.movement_status = 'idle'
g.movement_error = None
# Keep the active port constrained to the migrated job subset.
from mosfet.game import Game
from mosfet.job import JobStates
g.game = Game(g)
g.job = JobStates(g)
# Preserve active work during hot reloads, but give the worker all fields
# introduced by the currently migrated inventory controls.
g.setdefault('draining', False)
g.setdefault('dumping', None)
g.setdefault('dump_waiting', None)
g.setdefault('dump_dropped', 0)
g.setdefault('dump_total', 0)
g.setdefault('dump_error', None)
g.setdefault('drop_waiting', None)
g.setdefault('drain_slots', [])
g.setdefault('drain_waiting', None)
g.setdefault('drain_dropped', 0)
g.setdefault('drain_total', 0)
g.setdefault('drain_error', None)
g.setdefault('item_lock', False)
g.setdefault('minimum_cache_slots', 27)
g.setdefault('maximum_supply_slots', 33)
g.setdefault('holding_slot', 0)
g.setdefault('holding_hand', 0)
g.setdefault('time', 0.0)
old_stop = g.get('movement_stop_event')
old_worker = g.get('movement_thread')
if old_stop is not None:
old_stop.set()
if old_worker is not None and old_worker is not threading.current_thread():
old_worker.join(timeout=1.0)
stop_event = threading.Event()
g.movement_stop_event = stop_event
def run():
from mosfet import bot, utils
loaded_since = None
initial_world_ready = False
def set_status(status):
if g.movement_status != status:
print('[movement] status={}'.format(status), flush=True)
g.movement_status = status
while (not stop_event.is_set() and g.get('running', True) and
g.get('connection') and g.connection.connected):
started = time.monotonic()
now = time.monotonic()
if not initial_world_ready:
if not bot.missing_movement_chunks(g):
if loaded_since is None:
loaded_since = now
elif now - loaded_since >= 0.5:
initial_world_ready = True
else:
loaded_since = None
# Wait once for the initial position/chunk batch. After that,
# bot.tick() keeps requesting chunks at cache boundaries instead
# of gating itself forever on a momentary missing column.
movement_ready = initial_world_ready
job_active = (g.job is not None and g.job.state != g.job.idle)
has_work = (bool(g.path) or job_active or bool(g.breaking) or
bool(g.draining) or g.get('dumping') is not None or
bool(g.get('drop_waiting')))
if g.get('movement_paused'):
set_status('paused')
elif has_work and not movement_ready:
set_status('waiting_for_chunks')
elif has_work and movement_ready:
try:
reload_lock = g.get('reload_lock')
if reload_lock is None:
bot.tick(g)
else:
with reload_lock:
bot.tick(g)
except Exception:
traceback.print_exc()
g.movement_error = traceback.format_exc()
g.path = []
g.movement_status = 'error'
print('[movement] stopped after navigation error',
flush=True)
return
elif not has_work and not g.get('movement_paused'):
set_status('idle')
delay = max(0.0, utils.TICK - (time.monotonic() - started))
time.sleep(delay)
worker = threading.Thread(target=run, name='mosfet-movement',
daemon=True)
g.movement_thread = worker
worker.start()
return worker
def _remove_instance_listeners(connection, instance):
if connection is None or instance is None:
return
for name in ('packet_listeners', 'early_packet_listeners',
'outgoing_packet_listeners',
'early_outgoing_packet_listeners'):
listeners = getattr(connection, name, [])
listeners[:] = [listener for listener in listeners
if getattr(getattr(listener, 'callback', None),
'__self__', None) is not instance]
def refresh_runtime(global_state, reload_managers=False):
"""Rebind reloadable runtime adapters without touching the connection."""
g = global_state
connection = g.get('connection')
_remove_instance_listeners(connection, g.get('chat'))
_remove_instance_listeners(connection, g.get('game'))
if reload_managers:
_remove_instance_listeners(connection, g.get('entities'))
_remove_instance_listeners(connection, g.get('inventory'))
if reload_managers:
from mosfet.protocol import managers
if connection and g.get('chunks'):
current = g.chunks
replacement = managers.WorldState(current.min_y)
replacement.__dict__.update(current.__dict__)
g.chunks = replacement
g.world_state = replacement
connection.world_state = replacement
from mosfet.info import blocks as block_data
from mosfet.info import items as item_data
g.mcdata = DataManager()
g.entities = managers.EntityManager(g)
g.inventory = managers.InventoryManager(g)
# Keep data references in the reloaded legacy Game adapter current.
_ = (block_data, item_data)
from mosfet.protocol.modern import ModernChatManager
g.chat = ModernChatManager(g)
g.world = world.World(g)
g.game = game.Game(g)
state_name = getattr(g.get('job'), 'state', None)
state_name = getattr(state_name, '__name__', None)
g.job = job.JobStates(g)
if state_name and state_name != 'idle':
replacement_state = getattr(g.job, state_name, None)
if replacement_state is not None:
g.job.state = replacement_state
g.commands = commands.Commands(g)
if not reload_managers:
g.mcdata = DataManager()
start_automation(g, initialize=False)
print('[hot reload] runtime adapters refreshed; connection kept alive',
flush=True)
Reference in New Issue View Git Blame Copy Permalink