initial work, ported some basic logic out of aiocraft

This commit is contained in:
əlemi 2021-11-24 02:47:35 +01:00
parent d2fa232ee9
commit 15db4bc210
8 changed files with 257 additions and 0 deletions

2
requirements.txt Normal file
View file

@ -0,0 +1,2 @@
apscheduler
aiocraft

27
setup.py Normal file
View file

@ -0,0 +1,27 @@
from setuptools import setup, find_packages
with open("requirements.txt") as f:
requirements = f.read().split("\n")
setup(
name='treepuncher',
version='0.0.1',
description='An hackable Minecraft client, built with aiocraft',
url='https://github.com/alemidev/treepuncher',
author='alemi',
author_email='me@alemi.dev',
license='MIT',
packages=find_packages(),
package_data = {
'treepuncher': ['py.typed'],
},
install_requires=requirements,
classifiers=[
'Development Status :: 1 - Planning',
'Intended Audience :: Developers',
'License :: OSI Approved :: MIT License',
'Operating System :: POSIX :: Linux',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3.8',
],
)

1
treepuncher/__init__.py Normal file
View file

@ -0,0 +1 @@
from .treepuncher import Treepuncher

0
treepuncher/__main__.py Normal file
View file

View file

@ -0,0 +1 @@
from .chat import ChatEvent

View file

@ -0,0 +1,54 @@
import re
from typing import Optional
from enum import Enum
from aiocraft.util.helpers import parse_chat
CHAT_MESSAGE_MATCHER = re.compile(r"<(?P<usr>[A-Za-z0-9_]+)> (?P<msg>.+)")
REMOVE_COLOR_FORMATS = re.compile(r"§[0-9a-z]")
WHISPER_MATCHER = re.compile(r"(?:to (?P<touser>[A-Za-z0-9_]+)( |):|(?P<fromuser>[A-Za-z0-9_]+) whispers( |):|from (?P<from9b>[A-Za-z0-9_]+):) (?P<txt>.+)", flags=re.IGNORECASE)
JOIN_LEAVE_MATCHER = re.compile(r"(?P<usr>[A-Za-z0-9_]+) (?P<action>joined|left)( the game|)$", flags=re.IGNORECASE)
class MessageType(Enum):
CHAT = "chat"
WHISPER = "whisper"
#COMMAND = "cmd"
#DEATH = "death" # ???? TODO!
JOIN = "join"
LEAVE = "leave"
SYSTEM = "system"
class ChatEvent:
text : str
type : MessageType
user : str
message : str
def __init__(self, text:str):
self.text = REMOVE_COLOR_FORMATS.sub("", parse_chat(text))
self.user = ""
self.message= ""
self.type = MessageType.SYSTEM
self._parse()
def _parse(self):
match = CHAT_MESSAGE_MATCHER.search(self.text)
if match:
self.type = MessageType.CHAT
self.user = match["usr"]
self.message = match["msg"]
return
match = WHISPER_MATCHER.search(self.text)
if match:
self.type = MessageType.WHISPER
self.user = match["touser"] or match["fromuser"] or match["from9b"]
self.message = match["txt"]
return
match = JOIN_LEAVE_MATCHER.search(self.text)
if match:
self.type = MessageType.JOIN if match["action"] == "join" else MessageType.LEAVE
self.user = match["usr"]
return

0
treepuncher/py.typed Normal file
View file

172
treepuncher/treepuncher.py Normal file
View file

@ -0,0 +1,172 @@
import re
from typing import List, Dict, Union, Optional
from enum import Enum
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from aiocraft.client import MinecraftClient
from aiocraft.mc.packet import Packet
from aiocraft.mc.definitions import Difficulty, Dimension, Gamemode, Position
from aiocraft.mc.proto.play.clientbound import (
PacketRespawn, PacketLogin, PacketPosition, PacketUpdateHealth, PacketExperience,
PacketAbilities, PacketChat as PacketChatMessage
)
from aiocraft.mc.proto.play.serverbound import PacketTeleportConfirm, PacketClientCommand, PacketChat
from .events import ChatEvent
from .events.chat import MessageType
REMOVE_COLOR_FORMATS = re.compile(r"§[0-9a-z]")
class BotEvents(Enum):
DIED = 0
class Treepuncher(MinecraftClient):
in_game : bool
gamemode : Gamemode
dimension : Dimension
difficulty : Difficulty
hp : float
food : float
xp : float
lvl : int
total_xp : int
slot : int
# TODO inventory
position : Position
# TODO world
# TODO player abilities
# walk_speed : float
# fly_speed : float
# flags : int
scheduler : AsyncIOScheduler
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.in_game = False
self.gamemode = Gamemode.SURVIVAL
self.dimension = Dimension.OVERWORLD
self.difficulty = Difficulty.HARD
self.hp = 20.0
self.food = 20.0
self.xp = 0.0
self.lvl = 0
self.slot = 0
self.position = Position(0, 0, 0)
self._register_handlers()
self.scheduler = AsyncIOScheduler()
# self.scheduler.add_job(job, "interval", seconds=3)
self.scheduler.start(paused=True)
async def start(self):
await super().start()
self.scheduler.resume()
async def stop(self, force:bool=False):
self.scheduler.pause()
await super().stop(force=force)
def on_chat(self, msg_type:Union[str, MessageType] = None):
if isinstance(msg_type, str):
msg_type = MessageType(msg_type)
def wrapper(fun):
async def process_chat_packet(packet:PacketChatMessage):
msg = ChatEvent(packet.message)
if not msg_type or msg.type == msg_type:
return await fun(msg)
return self.register(PacketChatMessage, process_chat_packet)
return wrapper
def on_death(self):
def wrapper(fun):
return self.register(BotEvents.DIED, fun)
return wrapper
async def write(self, packet:Packet, wait:bool=False):
await self.dispatcher.write(packet, wait)
async def chat(self, message:str, wait:bool=False):
await self.dispatcher.write(
PacketChat(
self.dispatcher.proto,
message=message
),
wait=wait
)
def _register_handlers(self):
@self.on_packet(PacketRespawn)
async def on_player_respawning(packet:PacketRespawn):
self.gamemode = Gamemode(packet.gamemode)
self.dimension = Dimension(packet.dimension)
self.difficulty = Difficulty(packet.difficulty)
if self.difficulty != Difficulty.PEACEFUL \
and self.gamemode != Gamemode.SPECTATOR:
self.in_game = True
else:
self.in_game = False
self._logger.info(
"Reloading world: %s (%s) in %s",
self.dimension.name,
self.difficulty.name,
self.gamemode.name
)
@self.on_packet(PacketLogin)
async def player_joining_cb(packet:PacketLogin):
self.gamemode = Gamemode(packet.gameMode)
self.dimension = Dimension(packet.dimension)
self.difficulty = Difficulty(packet.difficulty)
if self.difficulty != Difficulty.PEACEFUL \
and self.gamemode != Gamemode.SPECTATOR:
self.in_game = True
else:
self.in_game = False
self._logger.info(
"Joined world: %s (%s) in %s",
self.dimension.name,
self.difficulty.name,
self.gamemode.name
)
@self.on_packet(PacketPosition)
async def player_rubberband_cb(packet:PacketPosition):
self._logger.info("Position synchronized")
self.position = Position(packet.x, packet.y, packet.z)
await self.dispatcher.write(
PacketTeleportConfirm(
self.dispatcher.proto,
teleportId=packet.teleportId
)
)
@self.on_packet(PacketUpdateHealth)
async def player_hp_cb(packet:PacketUpdateHealth):
if packet.health != self.hp and packet.health <= 0:
self._logger.info("Dead, respawning...")
await self.dispatcher.write(
PacketClientCommand(self.dispatcher.proto, actionId=0) # respawn
)
self.run_callbacks(BotEvents.DIED)
self.hp = packet.health
self.food = packet.food
@self.on_packet(PacketExperience)
async def player_hp_cb(packet:PacketExperience):
self.xp = packet.experienceBar
self.lvl = packet.level
self.total_xp = packet.totalExperience