tanner/minecraft-bot
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Port nether wart farming to 26.3

Browse Source
This commit is contained in:
Tanner
2026-11-07 17:25:51 +00:00
parent 9251edc36a
commit 95f4aa4821
7 changed files with 237 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mosfet/bot.py
+2 -1
View File
@@ -62,7 +62,8 @@ FARMING_PITCH = 45.0
def pitch_lock_for_job(job_state):
state = getattr(job_state, 'state', None) if job_state else None
return (FARMING_PITCH if getattr(state, '__name__', None) == 'farm_crop'
return (FARMING_PITCH
if getattr(state, '__name__', None) in ('farm_crop', 'farm_wart')
else None)
mosfet/commands.py
+1 -1
View File
@@ -73,7 +73,7 @@ class Commands:
command in unsupported or
(command == 'farm' and
not data.lower().startswith('crop') and
data.lower().strip() != 'wood')):
data.lower().strip() not in ('wood', 'wart'))):
reply = 'not migrated to Minecraft 26.3 yet'
print('Reply:', reply)
self._send_reply(reply, private, reply_target)
mosfet/info/blocks.py
+1 -1
View File
@@ -78,7 +78,7 @@ SAPLINGS = sorted(name for name in MCD_BLOCKS
if name.endswith('_sapling') or name.endswith('_propagule'))
BEDS = sorted(name for name in MCD_BLOCKS if name.endswith('_bed'))
INDEXED = ['chest', 'trapped_chest', 'emerald_block', 'barrel'] + BEDS + [
'wheat', 'potatoes', 'carrots', 'beetroots',
'wheat', 'potatoes', 'carrots', 'beetroots', 'nether_wart',
] + LOGS
LOG_IDS = get_set(LOGS)
mosfet/job.py
+13 -1
View File
@@ -270,11 +270,12 @@ class LegacyJobStates:
class JobStates(LegacyJobStates):
"""26.3 facade for migrated gather-wood and crop workflows."""
"""26.3 facade for migrated gathering and farming workflows."""
def __init__(self, global_state):
self.g = global_state
self.gather_wood_states = gather_wood.GatherWoodStates(global_state)
self.gather_crop_states = gather_crop.GatherCropStates(global_state)
self.gather_wart_states = gather_wart.GatherWartStates(global_state)
self.grab_supplies_states = grab_supplies.GrabSuppliesStates(
global_state)
self.clear_leaves_states = clear_leaves.ClearLeavesStates(global_state)
@@ -308,6 +309,17 @@ class JobStates(LegacyJobStates):
self.cache_items_states,
]
def farm_wart(self):
self.sleep_with_bed_states.silent = True
self.cache_items_states.silent = True
items.set_wanted({items.NETHERWART_ID})
return [
self.gather_wart_states,
self.sleep_with_bed_states,
self.eat_food_states,
self.cache_items_states,
]
def farm_crop(self):
self.cache_items_states.silent = True
self.sleep_with_bed_states.silent = True
mosfet/jobs/gather_wart.py
+72 -52
View File
@@ -1,52 +1,52 @@
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 import utils
from mosfet.info import blocks
from mosfet.info import items
from mosfet.info import mcdata
from mosfet.info import mobs
class GatherWartStates:
BREAK_TIMEOUT = 5.0
ITEM_TIMEOUT = 8.0
PLACE_TIMEOUT = 3.0
def idle(self):
return None
def init(self):
self.state = self.find_new_wart
def _blacklist_wart(self):
if self.wart is not None and self.wart not in self.bad_warts:
self.bad_warts.append(self.wart)
def find_new_wart(self):
print('Finding new wart...')
w = self.g.world
p = utils.pint(self.g.pos)
world = self.g.world
center = utils.pint(self.g.pos)
mature_wart = max(blocks.NETHERWART_IDS)
for wart in w.find_blocks_3d(p, [mature_wart], 50, 20, True):
if (wart not in self.bad_warts and
w.work_position_allowed(wart)):
break
else: # for
candidates = world.find_blocks_indexed(
center, {mature_wart}, 50)
candidates = world.filter_work_positions(candidates)
wart = next((position for position in candidates
if position not in self.bad_warts), None)
if wart is None:
print('No good warts left, aborting.')
self.state = self.cleanup
return
print('Found wart:', wart)
self.wart = wart
self.type_id = mature_wart
self.state = self.nav_to_wart
def nav_to_wart(self):
w = self.g.world
p = utils.pint(self.g.pos)
navpath = w.path_to_place(p, self.wart)
world = self.g.world
position = utils.pint(self.g.pos)
navpath = world.path_to_place(position, self.wart)
if navpath:
print('Going to wart', self.wart)
@@ -55,8 +55,7 @@ class GatherWartStates:
self.state = self.going_to_wart
else:
print('Cant get to it, blacklisting')
time.sleep(0.1)
self.bad_warts.append(wart)
self._blacklist_wart()
self.state = self.find_new_wart
def going_to_wart(self):
@@ -65,63 +64,84 @@ class GatherWartStates:
self.state = self.break_wart
def break_wart(self):
self.g.game.break_block(self.wart)
self.wait_time = 0.5
self.state = self.wait
current = self.g.world.block_at(*self.wart)
if current != self.type_id:
print('Wart changed before break; searching again')
self._blacklist_wart()
self.state = self.find_new_wart
return
def wait(self):
# wait for the item
if self.wait_time > 0:
self.wait_time -= utils.TICK
if self.g.game.break_block(self.wart):
self.break_deadline = time.monotonic() + self.BREAK_TIMEOUT
self.state = self.wait_break
else:
print('Break request skipped; blacklisting', self.wart)
self._blacklist_wart()
self.state = self.find_new_wart
def wait_break(self):
current = self.g.world.block_at(*self.wart)
if current != self.type_id:
print('Wart break confirmed at {}: {} -> {}'.format(
self.wart, self.type_id, current))
self.item_deadline = time.monotonic() + self.ITEM_TIMEOUT
self.item_wait_logged = False
self.state = self.select_wart
elif time.monotonic() > self.break_deadline:
print('Wart did not break; blacklisting', self.wart)
self._blacklist_wart()
self.state = self.find_new_wart
def select_wart(self):
p = utils.pint(self.g.pos)
if self.g.game.select_item([items.NETHERWART_ID]):
self.state = self.wait_select
self.wait_time = 0.5
else:
print('Aborting planting, no wart')
self.state = self.cleanup
return
if time.monotonic() > self.item_deadline:
print('No nether wart collected; skipping', self.wart)
self._blacklist_wart()
self.state = self.find_new_wart
elif not self.item_wait_logged:
print('Waiting for nether wart in inventory/hotbar')
self.item_wait_logged = True
def wait_select(self):
# wait a bit to select
if self.wait_time > 0:
self.wait_time -= utils.TICK
else:
if self.g.get('holding') == items.NETHERWART_ID:
self.state = self.place_wart
def place_wart(self):
p = utils.pint(self.g.pos)
self.g.game.place_block(p, BlockFace.TOP)
print('Placed wart')
soul_sand = utils.padd(self.wart, path.BLOCK_BELOW)
self.g.game.place_block(soul_sand, BlockFace.TOP)
self.place_deadline = time.monotonic() + self.PLACE_TIMEOUT
self.state = self.wait_place
self.wait_time = 0.5
def wait_place(self):
# wait a bit for chunk data to update
if self.wait_time > 0:
self.wait_time -= utils.TICK
else:
current = self.g.world.block_at(*self.wart)
if current in blocks.NETHERWART_IDS and current != self.type_id:
print('Nether wart replanted at', self.wart)
self.state = self.cleanup
elif time.monotonic() > self.place_deadline:
print('Nether wart did not replant; blacklisting', self.wart)
self._blacklist_wart()
self.state = self.find_new_wart
def cleanup(self):
self.g.look_at = None
self.state = self.done
def done(self):
# never gets ran, placeholder
return None
def __init__(self, global_state):
self.g = global_state
self.state = self.idle
self.wart = None
self.type_id = None
self.bad_warts = []
self.wait_time = 0
self.break_deadline = None
self.item_deadline = None
self.place_deadline = None
self.item_wait_logged = False
def run(self):
self.state()
tests/test_farming_pitch.py
+6 -1
View File
@@ -6,15 +6,20 @@ from mosfet.bot import FARMING_PITCH, pitch_lock_for_job, update_pitch
class TestFarmingPitch(unittest.TestCase):
def test_only_crop_farming_locks_pitch(self):
def test_crop_and_wart_farming_lock_pitch(self):
def farm_crop():
pass
def farm_wart():
pass
def sleep():
pass
self.assertEqual(
pitch_lock_for_job(Munch(state=farm_crop)), FARMING_PITCH)
self.assertEqual(
pitch_lock_for_job(Munch(state=farm_wart)), FARMING_PITCH)
self.assertIsNone(pitch_lock_for_job(Munch(state=sleep)))
self.assertIsNone(pitch_lock_for_job(None))
tests/test_gather_wart.py
+142
View File
@@ -0,0 +1,142 @@
import unittest
from munch import Munch
from mosfet.commands import Commands
from mosfet.info import blocks, items
from mosfet.job import JobStates
from mosfet.jobs.gather_wart import GatherWartStates
class GatherWartTests(unittest.TestCase):
def test_nether_wart_states_are_indexed_for_patch_search(self):
self.assertTrue(blocks.NETHERWART_IDS.issubset(blocks.INDEXED_IDS))
def test_farm_wart_command_is_enabled_and_keeps_job_sequence(self):
old_wanted = items.WANTED_ITEMS
job = JobStates(Munch())
class FakeChat:
def __init__(self):
self.sent = []
def set_handler(self, _handler):
pass
def send(self, message):
self.sent.append(message)
chat = FakeChat()
g = Munch(name='mosfet1', chat=chat, job=job,
inv={0: Munch(item_id=next(iter(items.BED_IDS)))})
try:
Commands(g).handle_chat(('CHAT', 'tester',
'tester: 1farm wart'))
self.assertEqual(chat.sent, ['ok'])
self.assertIs(job.state.__func__, job.farm_wart.__func__)
self.assertIs(job.farm_wart()[0], job.gather_wart_states)
self.assertIn(items.NETHERWART_ID, items.WANTED_ITEMS)
finally:
items.WANTED_ITEMS = old_wanted
def test_finds_only_indexed_mature_wart_allowed_by_work_signs(self):
position = (5, 64, 8)
mature = max(blocks.NETHERWART_IDS)
class FakeWorld:
def __init__(self):
self.search = None
self.filter_input = None
def find_blocks_indexed(self, center, block_ids, distance):
self.search = (center, block_ids, distance)
return [position]
def filter_work_positions(self, positions):
self.filter_input = list(positions)
return list(positions)
world = FakeWorld()
state = GatherWartStates(Munch(world=world, pos=(1.5, 64, 1.5)))
state.find_new_wart()
self.assertEqual(world.search, ((1, 64, 1), {mature}, 50))
self.assertEqual(world.filter_input, [position])
self.assertEqual(state.wart, position)
self.assertIs(state.state.__func__, state.nav_to_wart.__func__)
def test_harvests_confirms_replant_on_soul_sand_below_wart(self):
position = (5, 64, 8)
mature = max(blocks.NETHERWART_IDS)
unripe = min(blocks.NETHERWART_IDS)
class FakeWorld:
def __init__(self):
self.state_id = mature
def block_at(self, *_position):
return self.state_id
class FakeGame:
def __init__(self, global_state):
self.g = global_state
self.break_calls = []
self.place_calls = []
def break_block(self, target):
self.break_calls.append(target)
return True
def select_item(self, item_ids):
self.g.holding = items.NETHERWART_ID
return items.NETHERWART_ID in item_ids
def place_block(self, target, face):
self.place_calls.append((target, face))
world = FakeWorld()
g = Munch(world=world, pos=position, holding=0, look_at=None)
game = FakeGame(g)
g.game = game
state = GatherWartStates(g)
state.wart = position
state.type_id = mature
state.break_wart()
self.assertIs(state.state.__func__, state.wait_break.__func__)
self.assertEqual(game.break_calls, [position])
world.state_id = blocks.AIR
state.wait_break()
self.assertIs(state.state.__func__, state.select_wart.__func__)
state.select_wart()
state.wait_select()
self.assertIs(state.state.__func__, state.place_wart.__func__)
state.place_wart()
support = (position[0], position[1] - 1, position[2])
self.assertEqual(game.place_calls, [(support, 1)])
world.state_id = unripe
state.wait_place()
self.assertIs(state.state.__func__, state.cleanup.__func__)
def test_unreachable_wart_is_blacklisted_without_undefined_variable(self):
position = (5, 64, 8)
class FakeWorld:
@staticmethod
def path_to_place(_start, _target):
return None
state = GatherWartStates(Munch(
world=FakeWorld(), pos=(1, 64, 1), path=[]))
state.wart = position
state.nav_to_wart()
self.assertIn(position, state.bad_warts)
self.assertIs(state.state.__func__, state.find_new_wart.__func__)
if __name__ == '__main__':
unittest.main()