Mercurial > code > home > repos > mineflayer

comparison termbanator.py @ 0:f570713a7d31 default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
start termbanator
author drewp@bigasterisk.com
date Sun, 25 Jun 2023 20:01:38 -0700
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:f570713a7d31
1 import datetime
2 import os
3 import pdb
4 import subprocess
5 import sys
6 import time
7 import logging
8 import traceback
9 from typing import Any, Optional, Tuple
10
11 from javascript import On, Once, console, require
12
13 logging.basicConfig(
14 level=logging.INFO,
15 format="%(asctime)s [%(levelname)s] [line %(lineno)s] %(message)s")
16 log = logging.getLogger()
17
18 mineflayer = require("mineflayer", "latest")
19 Vec3 = require("vec3").Vec3
20 registry = require('prismarine-registry')('1.19.3')
21 Block = require("prismarine-block")(registry)
22
23 host, port, rconPort = ('bang', 25665, 25675)
24
25 mfbot = mineflayer.createBot({
26 "host": host,
27 "port": port,
28 "username": 'termbanator',
29 'auth': 'offline',
30 'version': '1.19.3',
31 })
32 log.info('created')
33
34
35 class Player:
36
37 def __init__(self, player):
38 self._p = player
39 if self._p is None:
40 raise TypeError(f'player was {player}')
41 self.username = self._p.username
42
43 def __repr__(self):
44 return f'Player({self.username})'
45
46 def position(self) -> Tuple[float, float, float]:
47 if self._p.entity is None:
48 return getDistantPlayerPos(self.username)
49 pos = self._p.entity.position
50 return (round(pos.x, 2), round(pos.y, 2), round(pos.z, 2))
51
52
53 def getDistantPlayerPos(username) -> Tuple[float, float, float]:
54 out = subprocess.check_output([
55 "/home/drewp/Downloads/mcrcon/mcrcon", "-H", host, "-P",
56 str(rconPort), "-p", "111", f"/data get entity @p[name={username}] Pos"
57 ],
58 encoding='utf8')
59 words = out.split("the following entity data: ")[1].split('[')[1].split(
60 ']')[0].split(',')
61 return tuple(round(float(v.rstrip('d')), 2) for v in words)
62
63
64 class Bot:
65
66 def __init__(self, mineflayer_bot: Any):
67 self._bot = mineflayer_bot
68 self.me = Player(self._bot)
69 mcData = require('minecraft-data')(self._bot.version)
70 self.pf_module = require('mineflayer-pathfinder')
71 self._bot.loadPlugin(self.pf_module.pathfinder)
72
73 self.movements = self.pf_module.Movements(self._bot, mcData)
74 self.movements.allow1by1towers = True
75 self.movements.canDig = False
76 self.movements.scafoldingBlocks.push(
77 self._bot.registry.itemsByName['dirt'].id)
78 self.movements.maxDropDown = 300
79 self.movements.allowFreeMotion = True
80 self.movements.canOpenDoors = True
81
82 self._bot.removeAllListeners('chat')
83
84 self.walkStartTime = 0
85 self.waitUntil = time.time() + 2
86 self.target: Optional[Player] = None
87 self.announcedToTarget = False
88
89 def chat(self, txt):
90 log.info(f'say {txt!r}')
91 self._bot.chat(txt)
92
93 def teleport(self, x, y, z):
94 self._bot.chat(f'/tp {x} {y} {z}')
95
96 def setPathfinderGoal(self, x, y, z, rng):
97 pos = self.me.position()
98 log.info(f'now at {pos}, pathfinding to {x} {y} {z} {rng=}')
99 goal = self.pf_module.goals.GoalNear(x, y, z, rng)
100 self._bot.pathfinder.setMovements(self.movements)
101 self._bot.pathfinder.setGoal(goal)
102 self.walkStartTime = time.time()
103
104 def isMoving(self) -> bool:
105 return self._bot.pathfinder.isMoving()
106
107 def stopPathfinder(self):
108 log.info("Stopping pathfinder")
109 self._bot.pathfinder.stop()
110
111 def ban(self, player: Player, reason: str = "good night"):
112 self.chat(f'/ban {player.username} {reason}')
113
114 def quit(self):
115 self._bot.quit('done')
116 os.kill(os.getpid(), 15)
117
118 ########## below here doesn't use self._bot ##########
119
120 def onLogin(self):
121 self.chat("Beware the termbanator")
122 # self.teleport(-160, 200, 280)
123
124 def walkTo(self, player: Player):
125 self.target = player
126 self.announcedToTarget = False
127 log.info(f'walk to {self.target}')
128 pos = player.position()
129 self.setPathfinderGoal(pos[0], pos[1], pos[2], rng=3)
130
131 def stillWalking(self, walkMinTime=3):
132 if self.walkStartTime and time.time(
133 ) < self.walkStartTime + walkMinTime:
134 return True
135 return self.isMoving()
136
137 def update(self, waitBetweenTargets=9):
138 try:
139 now = time.time()
140
141 if self.waitUntil > 0 and now < self.waitUntil:
142 log.debug(f'waiting {round(self.waitUntil-now, 1)} more sec')
143 return
144 self.waitUntil = 0
145
146 if self.stillWalking():
147 pos = self.me.position()
148 log.info(f'still moving to {self.target}: {pos=}')
149 return
150
151 self.stopPathfinder()
152
153 if self.target:
154 if not self.announcedToTarget:
155 minsLeft = round(
156 (self.getTodayEndTime() - time.time()) / 60, 1)
157
158 if minsLeft > 0:
159 mins = 'minutes' if minsLeft != 1 else 'minute'
160 self.chat(
161 f'hi {self.target.username} - {minsLeft} {mins} left'
162 )
163 else:
164 self.ban(self.target)
165
166 self.announcedToTarget = True
167 self.sleepFor(waitBetweenTargets)
168 return
169
170 log.info('follow next player')
171
172 next = players.after(self.target)
173 try:
174 self.walkTo(next)
175 except ValueError as e:
176 log.warning(repr(e))
177 self.sleepFor(8)
178 except EveryoneGone:
179 bot.quit()
180 except Exception as e:
181 traceback.print_exc()
182 log.warning(repr(e))
183 self.sleepFor(60)
184
185 def getTodayEndTime(self) -> float:
186 current_date = datetime.datetime.now().date()
187 end_datetime = datetime.datetime.combine(current_date,
188 datetime.time(19, 0))
189
190 return end_datetime.timestamp()
191
192 def sleepFor(self, sec):
193 log.info(f'sleeping for {sec}')
194 self.waitUntil = time.time() + sec
195
196
197 class EveryoneGone(ValueError):
198 pass
199
200
201 class Players:
202
203 def __init__(self, mineflayer_bot: Any):
204 self._b = mineflayer_bot
205
206 def byName(self, name) -> Player:
207 # only works for 'nearby' players- useless
208 return Player(self._b.players[name])
209
210 def after(self, player: Optional[Player]) -> Player:
211 otherPlayers = [p for p in self._b.players if p != 'termbanator']
212 log.info(f'{otherPlayers=}')
213 if not otherPlayers:
214 log.info('everyone is gone')
215 raise EveryoneGone()
216 # import pdb;pdb.set_trace()
217 prevPlayer = player.username if player else None
218
219 try:
220 i = otherPlayers.index(prevPlayer)
221 except (AttributeError, ValueError):
222 i = 0
223 i = (i + 1) % len(otherPlayers)
224
225 return self.byName(otherPlayers[i])
226
227
228 bot = Bot(mfbot)
229 players = Players(mfbot)
230
231
232 @On(mfbot, "login")
233 def login(this):
234 bot.onLogin()
235
236
237 @On(mfbot, 'chat')
238 def handleMsg(this, sender, message, *args):
239 if sender and (sender != 'termbanator'):
240 pass
241
242
243 @On(mfbot, 'time')
244 def onTime(this):
245 bot.update()
246
247
248 @On(mfbot, 'message')
249 def message(emitter, msg, *args):
250 log.info(f'got msg event {msg.toString()}')