minecraft-bot/game.py

import re
import time
import importlib

from panda3d.core import LPoint3f

from minecraft.networking.packets import Packet, clientbound, serverbound

from protocol.packets import TimeUpdatePacket, SetSlotPacket, PlayerDiggingPacket, BlockBreakAnimationPacket, AcknowledgePlayerDiggingPacket

import utils
importlib.reload(utils)
import path
importlib.reload(path)
import blocks
importlib.reload(blocks)

class Game:
    def __init__(self, global_state):
        self.g = global_state
        self.l = self.g.local_state

        register = self.g.connection.register_packet_listener
        register(self.handle_block_change, clientbound.play.BlockChangePacket)
        register(self.handle_join_game, clientbound.play.JoinGamePacket)
        register(self.handle_position_and_look, clientbound.play.PlayerPositionAndLookPacket)
        register(self.handle_time_update, TimeUpdatePacket)
        register(self.handle_set_slot, SetSlotPacket)
        register(self.handle_break_animation, BlockBreakAnimationPacket)
        register(self.handle_break_ack, AcknowledgePlayerDiggingPacket)

        self.g.chat.set_handler(self.handle_chat)

    def handle_join_game(self, packet):
        print('Connected.')
        print(packet)
        self.g.info = packet
        self.g.eid = packet.entity_id

    def handle_block_change(self, packet):
        l = self.g.local_state

        if packet.block_state_id == blocks.SOUL_TORCH:
            try:
                l.goal = LPoint3f(x=packet.location[0], y=packet.location[1], z=packet.location[2])
                print('new waypoint:', l.goal)

                start = time.time()
                solution = path.Pathfinder(self.g.chunks).astar(utils.pint(self.g.pos), utils.pint(l.goal))
                if solution:
                    solution = list(solution)
                    l.path = solution
                    #l.jobstate.state = l.jobstate.stop
                    print(len(solution))
                    print(solution)
                    print(round(time.time() - start, 3), 'seconds')
                else:
                    print('No path found')
                    #say(connection, 'No path found')

                #l.y_v = 10.0
                #l.y_a = -36.0
            except BaseException as e:
                import traceback
                print(traceback.format_exc())

    def handle_position_and_look(self, packet):
        print('pos and look:')
        print(packet)
        p = LPoint3f(x=packet.x, y=packet.y, z=packet.z)
        self.g.pos = p

    def handle_chat(self, message):
        source, text = message
        reply = None

        match = re.match(r'<(\w+)> (.*)', text)
        if match:
            sender, text = match.groups()
        else:
            return

        if text.startswith('! '):
            text = text[2:]
        elif text.startswith('!'):
            text = text[1:]
        else:
            return

        if ' ' in text:
            command = text.split(' ', 1)[0]
            data = text.split(' ', 1)[1]
        else:
            command = text

        if command == 'ping':
            reply = 'pong'

        if command == 'echo' and data:
            reply = data

        if command == 'respawn':
            packet = serverbound.play.ClientStatusPacket()
            packet.action_id = serverbound.play.ClientStatusPacket.RESPAWN
            self.g.connection.write_packet(packet)
            reply = 'ok'

        if command == 'pos':
            reply = str(utils.pint(self.g.pos))[1:-1]

        if command == 'afk':
            reply = '/afk'

        if command == 'error':
            reply = 'ok'
            raise

        if command == 'break':
            self.break_block((616, 78, 496))
            reply = 'ok'

        if reply:
            print(reply)
            self.g.chat.send(reply)

    def handle_time_update(self, packet):
        l = self.g.local_state
        l.time = packet.time_of_day % 24000

    def handle_set_slot(self, packet):
        print(packet)
        if packet.window_id == 0:
            self.g.inv[packet.slot] = packet.slot_data

    def break_block(self, location):
        bid = self.g.chunks.get_block_at(*location)
        if bid != 0:
            packet = PlayerDiggingPacket()
            packet.status = 0
            packet.location = location
            packet.face = 1
            self.g.connection.write_packet(packet)

            self.l.breaking = location
            self.l.break_time = time.time() + utils.break_time(bid)

    def break_finish(self):
        packet = PlayerDiggingPacket()
        packet.status = 2
        packet.location = self.l.breaking
        packet.face = 1
        self.g.connection.write_packet(packet)
        self.l.breaking = None


    def handle_break_animation(self, packet):
        print(packet)

    def handle_break_ack(self, packet):
        print(packet)

    def animate(self):
        packet = serverbound.play.AnimationPacket()
        packet.hand = packet.HAND_MAIN
        self.g.connection.write_packet(packet)

    def tick(self):
        if self.l.breaking:
            self.animate()

            if time.time() >= self.l.break_time - 2*utils.TICK:
                self.break_finish()