BWU Python Scripting Framework

A Python scripting framework for bot automation, ported from JBotWithUsV2 (Java). Provides a clean API for writing scripts with dynamic loading, event-driven architecture, and a Dear ImGui UI with embedded live game streaming.

Features

• Dear ImGui UI — GLFW+OpenGL rendered command-line interface matching the Java framework
• Live Game Streaming — Embedded inline video stream from the game client via named pipe JPEG frames
• MessagePack RPC — Named pipe communication using MessagePack serialization (matching Java server exactly)
• Runtime Connection — Connect/disconnect to game pipes at runtime with auto-scan discovery
• 4K/HiDPI Support — Automatic DPI scaling for high-resolution monitors
• Script Lifecycle — on_start() -> on_loop() -> on_stop() with threaded execution
• Task Framework — Priority-based task execution via TaskScript
• Entity System — Fluent query builders for NPCs, Players, Scene Objects, Ground Items
• Inventory System — Backpack, Bank, Equipment containers with item queries
• Event Bus — Type-safe pub/sub for game events (ticks, actions, chat, var changes)
• Inter-Script Communication — MessageBus (async messaging) + SharedState (shared key-value store)
• Configuration — Typed config fields with persistent JSON storage and ImGui config panel
• Hot Reload — Drop .py files into scripts/ and scan without restart
• Human-like Delays — Gaussian delays, micro-breaks, probability utilities

Requirements

• Python 3.11+
• Windows (named pipe communication requires pywin32)

Quick Start

pip install -r requirements.txt
python main.py

Dependencies

imgui[glfw]>=2.0.0
PyOpenGL>=3.1.7
glfw>=2.6.0
Pillow>=10.0.0
pywin32>=306
msgpack>=1.0.0

Commands

Command	Description
connect [pipe]	Connect to game pipe (auto-scans if no name given)
disconnect	Disconnect from game pipe
ping	Test connection with ping/pong
scan	Scan for available game pipes
scripts	List available scripts
start <name>	Start a script by name
stop <name>	Stop a running script
reload	Rescan scripts directory
stream start [quality] [fps] [w] [h]	Start live game video stream
stream stop	Stop active stream
screenshot	Capture a game screenshot
config <name>	Open script configuration panel
metrics	Show RPC call metrics
clear	Clear output buffer
help	Show available commands
exit	Quit the application

Writing Scripts

Basic Script

from api.bot_script import BotScript, script_manifest
from api.script_context import ScriptContext
from api.entities.npc import Npcs
from api.util.humanize import Humanize


@script_manifest(name="My Script", version="1.0", author="Me", description="Does things")
class MyScript(BotScript):

    def on_start(self, ctx: ScriptContext) -> None:
        self.ctx = ctx
        self.npcs = Npcs(ctx.api)
        ctx.log.info("Started!")

    def on_loop(self) -> int:
        npc = self.npcs.query(name="Chicken").visible().nearest()
        if npc:
            npc.interact(option=1)
        return Humanize.loop_delay(600)

    def on_stop(self) -> None:
        self.ctx.log.info("Stopped!")

Task-Based Script

from api.bot_script import script_manifest
from api.script.task import Task
from api.script.task_script import TaskScript


@script_manifest(name="My Task Script", version="1.0", author="Me")
class MyTaskScript(TaskScript):

    def setup_tasks(self) -> None:
        self.add_task(HighPriorityTask(self.ctx))
        self.add_task(LowPriorityTask(self.ctx))

Tasks are checked in priority order each loop. The first task whose validate() returns True gets its execute() called.

Configuration

from api.config.config_field import ConfigField
from api.config.script_config import ScriptConfig

class MyScript(BotScript):
    def get_config_fields(self) -> list[ConfigField]:
        return [
            ConfigField.string_field("target", "Target NPC", "Chicken"),
            ConfigField.int_field("delay", "Loop Delay (ms)", 600, 100, 5000),
            ConfigField.bool_field("verbose", "Verbose Logging", True),
            ConfigField.choice_field("mode", "Mode", ["Combat", "Skilling", "Gathering"]),
        ]

    def on_config_update(self, config: ScriptConfig) -> None:
        self.target = config.get_str("target", "Chicken")
        self.delay = config.get_int("delay", 600)

Events

from api.event.events import ChatMessageEvent, TickEvent

def on_start(self, ctx: ScriptContext) -> None:
    ctx.events.subscribe(ChatMessageEvent, self.on_chat)
    ctx.events.subscribe(TickEvent, self.on_tick)

def on_chat(self, event: ChatMessageEvent) -> None:
    ctx.log.info("Chat: %s", event.message.text)

Entity Queries

npcs = Npcs(ctx.api)
objects = SceneObjects(ctx.api)
backpack = Backpack(ctx.api)

# Fluent query builder
npc = npcs.query().named("Guard").visible().within_distance(10).nearest()

# Scene objects with option filter
tree = objects.query().name_contains("Tree").with_option("Chop down").nearest()

# Inventory queries
has_logs = backpack.contains(name="Logs")
log_count = backpack.count(name="Logs")
free = backpack.free_slots()

Project Structure

├── api/                        # Public API (scripts depend on this)
│   ├── bot_script.py           # BotScript ABC + @script_manifest
│   ├── game_api.py             # RPC game interaction interface
│   ├── script_context.py       # Context injected into scripts
│   ├── client.py               # Client/ClientProvider
│   ├── config/                 # ConfigField, ScriptConfig
│   ├── constants/              # InterfaceIds, ActionTypes, SkillId
│   ├── entities/               # Npc, Player, SceneObject, GroundItem
│   ├── event/                  # EventBus + built-in event types
│   ├── inventory/              # Backpack, Bank, Equipment
│   ├── isc/                    # MessageBus, SharedState
│   ├── log/                    # BotLogger
│   ├── model/                  # Entity, Coordinate, GameAction, types, etc.
│   ├── query/                  # EntityFilter, InventoryFilter, ComponentFilter
│   ├── script/                 # Task, TaskScript
│   └── util/                   # Humanize, Conditions
├── core/                       # Runtime implementation
│   ├── runtime/                # ScriptLoader, ScriptRunner, ScriptProfiler, EventDispatcher
│   ├── config/                 # ConfigStore (JSON persistence)
│   ├── rpc/                    # RpcClient (named pipe MessagePack RPC)
│   └── pipe/                   # PipeScanner, StreamPipeReader
├── ui/                         # Dear ImGui application
│   ├── app.py                  # Main window, command handling, RunnerManager
│   ├── theme.py                # Dark theme matching Java ImGuiTheme
│   ├── output_buffer.py        # AnsiOutputBuffer with ANSI SGR parsing
│   ├── output_line.py          # OutputLine data model (TEXT, IMAGE, PROGRESS, STREAM)
│   ├── texture_manager.py      # Thread-safe OpenGL texture lifecycle
│   ├── stream_manager.py       # Video stream lifecycle (start/stop/display)
│   └── imgui_config_panel.py   # Script config floating window
├── scripts/                    # User scripts (drop .py files here)
│   ├── example_script.py
│   └── example_task_script.py
├── requirements.txt
└── main.py                     # Entry point

Game Client Connection

The framework communicates with the game client via named pipe (\\.\pipe\BotWithUs) using MessagePack serialization with length-prefixed framing. Connection is managed at runtime through the connect command — the app auto-scans for available pipes or accepts a pipe name directly. If not connected, scripts load but game API calls return empty results.

Architecture

• Single-lock RPC — Matches the Java RpcClient: one lock serializes all pipe I/O, the calling thread reads its own response and dispatches interleaved events
• Background reader — Polls PeekNamedPipe for events when no RPC call is active
• Stream pipes — start_stream RPC returns a separate named pipe for JPEG frame delivery
• Event pipeline — RPC events are decoded from MessagePack, converted to typed events by EventDispatcher, and published on the EventBus for script subscribers