dehlak/clod
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: tony starkified clod

Browse Source
This commit is contained in:
dehlak
2026-02-15 14:01:45 +01:00
parent 786f259b8e
commit 79665a0acc
6 changed files with 227 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
minecraft-mcp/src/minecraft_mcp/__pycache__/server.cpython-314.pyc
BIN
View File
Binary file not shown.
minecraft-mcp/src/minecraft_mcp/server.py
+97 -15
View File
@@ -24,6 +24,9 @@ _last_build_time: float = 0.0
_BUILD_SESSION_TIMEOUT = 3.0 # seconds of inactivity before session ends
_build_session_origin: tuple[float, float, float] | None = None
_watcher_task: asyncio.Task | None = None
_build_bounds: dict | None = None # {"min_x", "max_x", "min_y", "max_y", "min_z", "max_z"}
_build_flying: bool = False # whether current session uses flying mode
_build_fly_xz: tuple[float, float] | None = None # fixed XZ while flying
async def _request(method: str, path: str, **kwargs) -> Any:
@@ -67,6 +70,10 @@ async def _look_at_pos(x: float, y: float, z: float) -> None:
await _request("POST", "/look_at", json={"x": x, "y": y, "z": z})
async def _set_bot_flying(flying: bool) -> None:
await _request("POST", "/set_flying", json={"flying": flying})
async def _get_bot_position() -> tuple[float, float, float] | None:
data = await _request("GET", "/status")
if isinstance(data, dict) and "position" in data:
@@ -78,8 +85,42 @@ async def _get_bot_position() -> tuple[float, float, float] | None:
def _compute_viewing_position(
build_x: float, build_y: float, build_z: float,
bot_x: float, bot_y: float, bot_z: float,
bounds: dict | None = None,
) -> tuple[float, float, float]:
"""Compute a spot 5 blocks from the build, on the side the bot is on."""
"""Compute a viewing spot outside the build volume.
When *bounds* is provided, position 3 blocks outside the nearest edge
of the build footprint (on the side the bot is already on).
Otherwise fall back to 5 blocks from the center point.
"""
if bounds is not None:
min_x = bounds["min_x"]
max_x = bounds["max_x"]
min_z = bounds["min_z"]
max_z = bounds["max_z"]
center_x = (min_x + max_x) / 2.0
center_z = (min_z + max_z) / 2.0
dx = bot_x - center_x
dz = bot_z - center_z
if abs(dx) < 0.01 and abs(dz) < 0.01:
dx = 1.0
# Pick the axis the bot is farthest on
if abs(dx) >= abs(dz):
# Position outside X edge
if dx >= 0:
vx = max_x + 3.5
else:
vx = min_x - 3.5
vz = center_z
else:
# Position outside Z edge
vx = center_x
if dz >= 0:
vz = max_z + 3.5
else:
vz = min_z - 3.5
return (vx, build_y, vz)
dx = bot_x - build_x
dz = bot_z - build_z
dist = math.hypot(dx, dz)
@@ -90,35 +131,67 @@ def _compute_viewing_position(
return (build_x + dx * scale, build_y, build_z + dz * scale)
async def _ensure_build_session(x: float, y: float, z: float) -> None:
async def _ensure_build_session(
x: float, y: float, z: float,
bounds: dict | None = None,
flying: bool = False,
) -> None:
"""Start or continue a build session, managing state/teleport/look."""
global _build_session_active, _last_build_time, _build_session_origin
global _watcher_task
global _watcher_task, _build_bounds, _build_flying, _build_fly_xz
async with _build_session_lock:
_last_build_time = time.monotonic()
if _build_session_active:
# Session already running — just look at the new target
# Session already running
_build_session_origin = (x, y, z)
await _look_at_pos(x, y, z)
if _build_flying and _build_fly_xz is not None:
# Flying mode: keep XZ fixed, only update Y to rise with layers
fx, fz = _build_fly_xz
await _teleport_bot(fx, y, fz)
# Look at the center of the build at the current layer height
if _build_bounds:
cx = (_build_bounds["min_x"] + _build_bounds["max_x"]) / 2.0
cz = (_build_bounds["min_z"] + _build_bounds["max_z"]) / 2.0
await _look_at_pos(cx, y, cz)
else:
await _look_at_pos(x, y, z)
else:
await _look_at_pos(x, y, z)
return
# --- Start a new session ---
_build_session_active = True
_build_session_origin = (x, y, z)
_build_bounds = bounds
_build_flying = flying
await _set_bot_state("building")
pos = await _get_bot_position()
if pos is not None:
bx, by, bz = pos
dist = math.sqrt((bx - x) ** 2 + (by - y) ** 2 + (bz - z) ** 2)
if dist > 16:
vx, vy, vz = _compute_viewing_position(x, y, z, bx, by, bz)
await _teleport_bot(vx, vy, vz)
await _look_at_pos(x, y, z)
if bounds is not None and flying:
# Compute safe position outside build volume
vx, vy, vz = _compute_viewing_position(x, y, z, bx, by, bz, bounds=bounds)
_build_fly_xz = (vx, vz)
await _set_bot_flying(True)
await _teleport_bot(vx, y, vz)
else:
_build_fly_xz = None
dist = math.sqrt((bx - x) ** 2 + (by - y) ** 2 + (bz - z) ** 2)
if dist > 16:
vx, vy, vz = _compute_viewing_position(x, y, z, bx, by, bz)
await _teleport_bot(vx, vy, vz)
if bounds is not None:
cx = (bounds["min_x"] + bounds["max_x"]) / 2.0
cz = (bounds["min_z"] + bounds["max_z"]) / 2.0
await _look_at_pos(cx, y, cz)
else:
await _look_at_pos(x, y, z)
# Start the watcher if not already running
if _watcher_task is None or _watcher_task.done():
@@ -128,6 +201,7 @@ async def _ensure_build_session(x: float, y: float, z: float) -> None:
async def _build_session_watcher() -> None:
"""Background task that ends the build session after inactivity."""
global _build_session_active, _build_session_origin
global _build_bounds, _build_flying, _build_fly_xz
while True:
await asyncio.sleep(1.0)
@@ -136,9 +210,14 @@ async def _build_session_watcher() -> None:
return
elapsed = time.monotonic() - _last_build_time
if elapsed >= _BUILD_SESSION_TIMEOUT:
if _build_flying:
await _set_bot_flying(False)
await _set_bot_state("idle")
_build_session_active = False
_build_session_origin = None
_build_bounds = None
_build_flying = False
_build_fly_xz = None
return
@@ -366,12 +445,13 @@ async def build_schematic(name: str, x: int, y: int, z: int) -> str:
layers = info.get("layers", [])
fmt = info.get("format", "unknown")
num_layers = len(layers)
bounds = info.get("bounds") # may be None for empty schematics
if num_layers == 0:
return _fmt({"ok": True, "message": "Schematic is empty, nothing to build.", "placed": 0})
# 2. Start build session and announce
await _ensure_build_session(x, y, z)
# 2. Start build session with flying + bounds and announce
await _ensure_build_session(x, y, z, bounds=bounds, flying=True)
await _request("POST", "/chat", json={
"message": f"Starting build of {name}! {total_blocks} blocks, {num_layers} layers."
})
@@ -384,8 +464,8 @@ async def build_schematic(name: str, x: int, y: int, z: int) -> str:
layer_y = layer["y"]
block_count = layer["block_count"]
# Keep build session alive and look at current layer
await _ensure_build_session(x, layer_y, z)
# Keep build session alive; flying mode rises with each layer
await _ensure_build_session(x, layer_y, z, bounds=bounds, flying=True)
# Check if this layer crosses a milestone threshold
progress = (i + 1) / num_layers
@@ -408,6 +488,7 @@ async def build_schematic(name: str, x: int, y: int, z: int) -> str:
}, timeout=15.0)
if isinstance(result, dict) and "error" in result:
await _set_bot_flying(False)
await _request("POST", "/chat", json={
"message": f"Build failed at layer {i + 1}/{num_layers}! Error: {result['error']}"
})
@@ -426,7 +507,8 @@ async def build_schematic(name: str, x: int, y: int, z: int) -> str:
if i < num_layers - 1:
await asyncio.sleep(0.4)
# 4. Announce completion
# 4. Land and announce completion
await _set_bot_flying(False)
await _request("POST", "/chat", json={
"message": f"Build complete! Placed {total_placed} blocks in {num_layers} layers."
})