Untitled
unknown
plain_text
10 months ago
18 kB
20
Indexable
import asyncio
import sys
import tkinter as tk
import win32api
import win32con
import win32gui
import logging
import time
import threading
import struct
import random
import queue
from pymem import Pymem
from pymem.exception import MemoryReadError
from tkinter import messagebox
# ------------------------------------------------------------
# Dictionary to track last logged times for throttling
# ------------------------------------------------------------
last_logged = {}
# ------------------------------------------------------------
# Configuration Constants
# ------------------------------------------------------------
PROCESS_NAME = "PathOfExileSteam.exe"
BASE_POINTER_OFFSET = 0x040A61A8
# Memory Offsets
OFFSETS_CUR_HP = [0x38, 0x0, 0x80, 0x290, 0x1D8]
OFFSETS_MAX_HP = [0x60, 0x68, 0x40, 0x20, 0x1F0, 0x2C]
OFFSETS_CUR_MP = [0x98, 0x70, 0x20, 0x50, 0x360]
OFFSETS_MAX_MP = [0x60, 0x38, 0x20, 0x48, 0x10, 0x1F0, 0x2C]
# Potion Keys
HP_KEY = "1"
MP_KEY = "2"
# Thresholds
HP_THRESHOLD = 1500
MP_THRESHOLD = 150
# Cooldowns
HP_FLASK_COOLDOWN = 1.0
MP_FLASK_COOLDOWN = 1.0
# Intervals
CHECK_INTERVAL = 0.25
# Rehook & Popup
REHOOK_COOLDOWN = 2
POPUP_COOLDOWN = 30
# ------------------------------------------------------------
# Logger Setup
# ------------------------------------------------------------
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(message)s')
logger = logging.getLogger()
# Global state
pm = None
base_pointer = None
game_window = None
cur_hp_addr = None
max_hp_addr = None
last_max_hp = None
cur_mp_addr = None
max_mp_addr = None
last_max_mp = None
last_hp_flask_time = 0
last_mp_flask_time = 0
last_rehook_attempt = 0
last_popup_time = 0
popup_queue = queue.Queue()
stop_event = asyncio.Event()
# Locks for flask cooldown times
hp_flask_lock = asyncio.Lock()
mp_flask_lock = asyncio.Lock()
# ------------------------------------------------------------
# Helper Functions
# ------------------------------------------------------------
def log_throttled(logger, level, message, key=None, cooldown=10):
now = time.time()
key = key or message
last_time = last_logged.get(key, 0)
if now - last_time > cooldown:
last_logged[key] = now
getattr(logger, level)(message)
def find_dma_address(pm, base, offsets, label=""):
if pm is None:
log_throttled(logger, "warning", "Pymem instance not initialized.", key="pymem_not_init", cooldown=30)
return False
try:
addr = pm.read_longlong(base)
log_throttled(logger, "debug", f"[{label}] Base: 0x{base:X} → 0x{addr:X}", key=f"dma_base_{label}", cooldown=5)
for i, offset in enumerate(offsets[:-1]):
if addr == 0:
log_throttled(logger, "warning", f"[{label}] Null pointer encountered at step {i}", key=f"null_ptr_{label}_{i}", cooldown=20)
return None
prev = addr
addr = pm.read_longlong(addr + offset)
log_throttled(logger, "debug", f"[{label}] Step {i}: 0x{prev + offset:X} → 0x{addr:X}", key=f"dma_step_{label}_{i}", cooldown=5)
final_address = addr + offsets[-1]
log_throttled(logger, "debug", f"[{label}] Final address: 0x{final_address:X}", key=f"dma_final_{label}", cooldown=5)
return final_address
except MemoryReadError as e:
log_throttled(logger, "warning", f"[{label}] DMA address calculation failed: {e}", key=f"dma_fail_{label}", cooldown=15)
return None
def press_key(window, key):
try:
win32api.PostMessage(window, win32con.WM_KEYDOWN, ord(key), 0)
win32api.PostMessage(window, win32con.WM_KEYUP, ord(key), 0)
return True
except Exception as e:
log_throttled(logger, "error", f"Error pressing key {key}: {e}", key="press_key_error", cooldown=30)
return False
def show_error_popup_async(message):
global last_popup_time
now = time.time()
if now - last_popup_time < POPUP_COOLDOWN:
log_throttled(logger, "info", "Skipping popup to avoid spam.", key="popup_skip", cooldown=POPUP_COOLDOWN)
return
last_popup_time = now
popup_queue.put(message)
def process_popup_queue(root):
try:
while True:
message = popup_queue.get_nowait()
messagebox.showerror("Memory Access Error", message)
except queue.Empty:
pass
root.after(100, lambda: process_popup_queue(root))
def find_window_by_partial_title(partial_title):
def enum_windows_callback(hwnd, hwnd_list):
title = win32gui.GetWindowText(hwnd)
if partial_title.lower() in title.lower():
hwnd_list.append(hwnd)
hwnd_list = []
win32gui.EnumWindows(enum_windows_callback, hwnd_list)
return hwnd_list[0] if hwnd_list else None
def is_game_running():
try:
hwnd = find_window_by_partial_title("Path of Exile")
if not hwnd:
log_throttled(logger, "error", "Game window not found. Please start the game and try again.", key="game_window_not_found", cooldown=60)
return False
return True
except Exception as e:
log_throttled(logger, "error", f"Error checking game window: {e}", key="game_window_check_error", cooldown=60)
return False
def is_game_active(pm):
if pm is None:
log_throttled(logger, "warning", "Pymem instance not initialized.", key="pymem_not_init_active", cooldown=30)
return False
try:
current_hp = pm.read_int(cur_hp_addr)
current_mp = pm.read_int(cur_mp_addr)
return not (current_hp == 0 and current_mp == 0)
except MemoryReadError as e:
log_throttled(logger, "warning", f"Memory read failed in is_game_active(): {e}", key="mem_read_fail_active", cooldown=20)
log_throttled(logger, "info", "Attempting to rehook pointers due to read failure...", key="rehook_attempt_active", cooldown=REHOOK_COOLDOWN)
safe_rehook(pm)
return False
def dump_pointer_info(pm):
if pm is None:
log_throttled(logger, "warning", "Pymem instance not initialized.", key="pymem_not_init_dump", cooldown=30)
return False
try:
with open("pointer_debug.txt", "w") as f:
f.write(f"Cur HP Addr: {hex(cur_hp_addr) if cur_hp_addr else 'None'}\n")
f.write(f"Max HP Addr: {hex(max_hp_addr) if max_hp_addr else 'None'}\n")
f.write(f"Cur MP Addr: {hex(cur_mp_addr) if cur_mp_addr else 'None'}\n")
f.write(f"Max MP Addr: {hex(max_mp_addr) if max_mp_addr else 'None'}\n")
except Exception as e:
log_throttled(logger, "warning", f"Failed to dump pointers: {e}", key="dump_pointer_fail", cooldown=60)
def rehook_pointers(pm):
global cur_hp_addr, max_hp_addr, cur_mp_addr, max_mp_addr, last_max_hp, last_max_mp
if pm is None:
log_throttled(logger, "warning", "Pymem instance not initialized.", key="pymem_not_init_rehook", cooldown=30)
return False
try:
temp_cur_hp = find_dma_address(pm, base_pointer, OFFSETS_CUR_HP, label="CUR_HP")
temp_max_hp = find_dma_address(pm, base_pointer, OFFSETS_MAX_HP, label="MAX_HP")
temp_cur_mp = find_dma_address(pm, base_pointer, OFFSETS_CUR_MP, label="CUR_MP")
temp_max_mp = find_dma_address(pm, base_pointer, OFFSETS_MAX_MP, label="MAX_MP")
if None in (temp_cur_hp, temp_max_hp, temp_cur_mp, temp_max_mp):
raise MemoryReadError(0, 4, "One or more DMA addresses are invalid (None).")
cur_hp_val = pm.read_int(temp_cur_hp)
max_hp_val = pm.read_int(temp_max_hp)
cur_mp_val = pm.read_int(temp_cur_mp)
max_mp_val = pm.read_int(temp_max_mp)
if not (0 <= cur_hp_val <= max_hp_val):
raise ValueError(f"Current HP value {cur_hp_val} out of bounds (0 to {max_hp_val})")
if not (0 <= cur_mp_val <= max_mp_val):
raise ValueError(f"Current MP value {cur_mp_val} out of bounds (0 to {max_mp_val})")
cur_hp_addr = temp_cur_hp
max_hp_addr = temp_max_hp
cur_mp_addr = temp_cur_mp
max_mp_addr = temp_max_mp
last_max_hp = max_hp_val
last_max_mp = max_mp_val
log_throttled(logger, "info", f"Rehook successful. HP ptr: 0x{cur_hp_addr:X}, MP ptr: 0x{cur_mp_addr:X}", key="rehook_success", cooldown=30)
log_throttled(logger, "info", f"HP: {cur_hp_val}/{max_hp_val}, MP: {cur_mp_val}/{max_mp_val}", key="rehook_values", cooldown=30)
except Exception as e:
log_throttled(logger, "error", f"Failed to rehook memory: {e}", key="rehook_fail", cooldown=60)
show_error_popup_async("Failed to rehook memory. You may need to restart the script or game.")
dump_pointer_info(pm)
def validate_pointers(pm):
global last_max_hp, last_max_mp
if pm is None:
log_throttled(logger, "warning", "Pymem instance not initialized.", key="pymem_not_init_validate", cooldown=30)
return False
try:
cur_hp = pm.read_int(cur_hp_addr)
max_hp = pm.read_int(max_hp_addr)
cur_mp = pm.read_int(cur_mp_addr)
max_mp = pm.read_int(max_mp_addr)
if last_max_hp != max_hp:
log_throttled(logger, "info", f"Max HP changed from {last_max_hp} to {max_hp}. Rehooking.", key="max_hp_change", cooldown=30)
last_max_hp = max_hp
safe_rehook(pm)
if last_max_mp != max_mp:
log_throttled(logger, "info", f"Max MP changed from {last_max_mp} to {max_mp}. Rehooking.", key="max_mp_change", cooldown=30)
last_max_mp = max_mp
safe_rehook(pm)
if not (0 <= cur_hp <= max_hp):
log_throttled(logger, "warning", f"Corrupt HP: cur_hp={cur_hp}, max_hp={max_hp}, rehooking.", key="corrupt_hp", cooldown=30)
safe_rehook(pm)
return None
if not (0 <= cur_mp <= max_mp):
log_throttled(logger, "warning", f"Corrupt MP: cur_mp={cur_mp}, max_mp={max_mp}, rehooking.", key="corrupt_mp", cooldown=30)
safe_rehook(pm)
return None
return (cur_hp, max_hp, cur_mp, max_mp)
except MemoryReadError as e:
log_throttled(logger, "warning", f"Validation read error: {e}. Rehooking.", key="validation_read_error", cooldown=30)
safe_rehook(pm)
return None
def safe_rehook(pm):
global last_rehook_attempt
if time.time() - last_rehook_attempt >= REHOOK_COOLDOWN:
last_rehook_attempt = time.time()
try:
rehook_pointers(pm)
except MemoryReadError as e:
log_throttled(logger, "error", f"Rehook failed: {e}", key="safe_rehook_fail", cooldown=60)
async def read_memory_with_retry(pm, addr, retries=3, delay=1):
if pm is None:
raise RuntimeError("Pymem instance not initialized.")
for attempt in range(retries):
try:
return pm.read_int(addr)
except MemoryReadError as e:
log_throttled(logger, "warning", f"Memory read failed (Attempt {attempt + 1}/{retries}): {e}", key="memory_read_fail", cooldown=10)
if attempt == retries - 1:
log_throttled(logger, "info", "Triggering pointer rehook due to repeated read failures...", key="memory_read_rehook", cooldown=60)
safe_rehook(pm)
await asyncio.sleep(delay)
raise MemoryReadError(addr, struct.calcsize('i'), "Failed after retries")
def get_random_delay(base, variance=0.15):
return base + random.uniform(-variance, variance)
# ------------------------------------------------------------
# Potion Routines
# ------------------------------------------------------------
async def confirm_hp_flask(current_hp, max_hp):
readings = []
for _ in range(6):
try:
hp_val = await read_memory_with_retry(pm, cur_hp_addr)
except MemoryReadError:
hp_val = 0
readings.append(hp_val)
await asyncio.sleep(0.3)
new_hp = max(readings)
if new_hp >= max_hp - 500:
log_throttled(logger, "info", f"HP flask confirmed: {current_hp} → {new_hp}", key="hp_flask_confirmed", cooldown=30)
else:
log_throttled(logger, "warning", f"HP flask ineffective: {current_hp} → {new_hp}", key="hp_flask_ineffective", cooldown=30)
async def confirm_mp_flask(current_mp, max_mp):
readings = []
for _ in range(6):
try:
mp_val = await read_memory_with_retry(pm, cur_mp_addr)
except MemoryReadError:
mp_val = 0
readings.append(mp_val)
await asyncio.sleep(0.3)
new_mp = max(readings)
if new_mp >= max_mp - 10:
log_throttled(logger, "info", f"MP flask confirmed: {current_mp} → {new_mp}", key="mp_flask_confirmed", cooldown=30)
else:
log_throttled(logger, "warning", f"MP flask ineffective: {current_mp} → {new_mp}", key="mp_flask_ineffective", cooldown=30)
async def hp_routine():
global last_hp_flask_time
while not stop_event.is_set():
try:
vals = validate_pointers(pm)
if vals is None:
await asyncio.sleep(CHECK_INTERVAL)
continue
current_hp, max_hp, _, _ = vals
if current_hp == 0 and max_hp == 0:
await asyncio.sleep(CHECK_INTERVAL)
continue
now = time.time()
async with hp_flask_lock:
if current_hp < HP_THRESHOLD and (now - last_hp_flask_time) > HP_FLASK_COOLDOWN:
log_throttled(logger, "info", f"HP below threshold ({current_hp} < {HP_THRESHOLD}), pressing {HP_KEY}", key="hp_flask_press", cooldown=HP_FLASK_COOLDOWN)
press_key(game_window, HP_KEY)
last_hp_flask_time = now
await confirm_hp_flask(current_hp, max_hp)
await asyncio.sleep(get_random_delay(CHECK_INTERVAL))
except Exception as e:
log_throttled(logger, "error", f"Error in hp_routine: {e}", key="hp_routine_error", cooldown=30, exc_info=True)
await asyncio.sleep(1)
async def mp_routine():
global last_mp_flask_time
while not stop_event.is_set():
try:
vals = validate_pointers(pm)
if vals is None:
await asyncio.sleep(CHECK_INTERVAL)
continue
_, _, current_mp, max_mp = vals
if current_mp == 0 and max_mp == 0:
await asyncio.sleep(CHECK_INTERVAL)
continue
now = time.time()
async with mp_flask_lock:
if current_mp < MP_THRESHOLD and (now - last_mp_flask_time) > MP_FLASK_COOLDOWN:
log_throttled(logger, "info", f"MP below threshold ({current_mp} < {MP_THRESHOLD}), pressing {MP_KEY}", key="mp_flask_press", cooldown=MP_FLASK_COOLDOWN)
press_key(game_window, MP_KEY)
last_mp_flask_time = now
await confirm_mp_flask(current_mp, max_mp)
await asyncio.sleep(get_random_delay(CHECK_INTERVAL))
except Exception as e:
log_throttled(logger, "error", f"Error in mp_routine: {e}", key="mp_routine_error", cooldown=30, exc_info=True)
await asyncio.sleep(1)
# ------------------------------------------------------------
# Initialization and Entry
# ------------------------------------------------------------
def init():
global pm, base_pointer, game_window, cur_hp_addr, max_hp_addr, cur_mp_addr, max_mp_addr
if not is_game_running():
raise RuntimeError("Game not running.")
try:
pm = Pymem(PROCESS_NAME)
base_pointer = pm.base_address + BASE_POINTER_OFFSET
game_window = find_window_by_partial_title("Path of Exile")
if not game_window:
raise RuntimeError("Game window not found.")
safe_rehook(pm)
if None in (cur_hp_addr, max_hp_addr, cur_mp_addr, max_mp_addr):
log_throttled(logger, "error", "Failed to initialize memory pointers.", key="init_fail", cooldown=60)
show_error_popup_async(
"Critical memory pointers could not be found.\nPlease restart the script or wait for an update."
)
raise RuntimeError("Critical memory pointers not found.")
max_hp = pm.read_int(max_hp_addr)
max_mp = pm.read_int(max_mp_addr)
log_throttled(logger, "info", f"Initialized. Max HP: {max_hp}, Max MP: {max_mp}", key="init_success", cooldown=60)
except Exception as e:
log_throttled(logger, "error", f"Initialization failed: {e}", key="init_error", cooldown=60, exc_info=True)
show_error_popup_async("Initialization error. Please restart the game.")
raise
async def start_script():
try:
init()
except Exception as e:
log_throttled(logger, "error", f"Startup initialization failed: {e}", key="startup_fail", cooldown=60)
return
tasks = [
asyncio.create_task(hp_routine()),
asyncio.create_task(mp_routine())
]
try:
await stop_event.wait()
finally:
for task in tasks:
task.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
# ------------------------------------------------------------
# Main Entry Point
# ------------------------------------------------------------
if __name__ == "__main__":
root = tk.Tk()
root.withdraw()
root.after(100, lambda: process_popup_queue(root))
try:
asyncio.run(start_script())
except Exception as e:
log_throttled(logger, "error", f"Unexpected error: {e}", key="unexpected_error", cooldown=60, exc_info=True)
show_error_popup_async(f"Unexpected error: {e}")
finally:
log_throttled(logger, "info", "Shutting down gracefully.", key="shutdown", cooldown=60)
try:
if root:
root.quit()
root.destroy()
except Exception as e:
log_throttled(logger, "warning", f"Failed to destroy Tkinter root: {e}", key="tkinter_destroy_fail", cooldown=60)
Editor is loading...
Leave a Comment