import json import math import mimetypes import os import subprocess import sys import threading import time import urllib.request from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn class ThreadedHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True from urllib.parse import urlparse PORT = 7777 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) COORDS_FILE = os.path.join(BASE_DIR, ".last_coords.json") DAEMON_SCRIPT = os.path.join(BASE_DIR, "location_daemon.py") DEFAULT_COORDS = {"lat": 33.4484, "lng": -112.0740} def load_coords(): try: with open(COORDS_FILE, "r") as f: data = json.load(f) if "lat" in data and "lng" in data: print(f"Resuming from last position: {data['lat']:.5f}, {data['lng']:.5f}") return data except Exception: pass return dict(DEFAULT_COORDS) def save_coords(lat, lng): try: with open(COORDS_FILE, "w") as f: json.dump({"lat": lat, "lng": lng}, f) except Exception: pass current_coords = load_coords() last_result = {"method": None, "ok": False, "error": "not tried yet"} _daemon_proc = None _daemon_lock = threading.Lock() _cached_rsd_host = None _cached_rsd_port = None _rsd_cache_time = 0 RSD_CACHE_TTL = 30 def get_rsd(): global _cached_rsd_host, _cached_rsd_port, _rsd_cache_time now = time.time() if _cached_rsd_host and (now - _rsd_cache_time) < RSD_CACHE_TTL: return _cached_rsd_host, _cached_rsd_port try: with urllib.request.urlopen("http://127.0.0.1:49151/", timeout=3) as r: tunnels = json.loads(r.read()) if isinstance(tunnels, dict) and tunnels: entry = list(tunnels.values())[0] if isinstance(entry, list) and entry: t = entry[0] host = t.get("tunnel-address") port = t.get("tunnel-port") if host and port: _cached_rsd_host = str(host) _cached_rsd_port = int(port) _rsd_cache_time = now return _cached_rsd_host, _cached_rsd_port except Exception: pass return None, None def start_daemon(): global _daemon_proc, _cached_rsd_host, _cached_rsd_port host, port = get_rsd() if not host: print(" [!] No RSD tunnel found — daemon not started", flush=True) return False try: proc = subprocess.Popen( [sys.executable, DAEMON_SCRIPT, host, str(port)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1 ) ready_line = proc.stdout.readline().strip() if ready_line == "ready": _daemon_proc = proc print(f" [✓] Location daemon connected via RSD {host}:{port}", flush=True) return True else: print(f" [!] Daemon startup failed: {ready_line}", flush=True) proc.kill() return False except Exception as e: print(f" [!] Could not start daemon: {e}", flush=True) return False def send_via_daemon(lat, lng): global _daemon_proc, last_result with _daemon_lock: if _daemon_proc is None or _daemon_proc.poll() is not None: print(" [~] Daemon not running — restarting...", flush=True) start_daemon() if _daemon_proc is None: return False try: _daemon_proc.stdin.write(f"{lat},{lng}\n") _daemon_proc.stdin.flush() response = _daemon_proc.stdout.readline().strip() if response == "ok": last_result = {"method": "DVT daemon (persistent)", "ok": True, "error": None} return True else: print(f" [!] Daemon response: {response}", flush=True) return False except Exception as e: print(f" [!] Daemon write failed: {e}", flush=True) _daemon_proc = None return False def send_location_cli(lat, lng): global last_result, _cached_rsd_host, _cached_rsd_port host, port = get_rsd() if host and port: try: r = subprocess.run( ["pymobiledevice3", "developer", "dvt", "simulate-location", "set", "--rsd", host, str(port), "--", str(lat), str(lng)], capture_output=True, text=True, timeout=5 ) if r.returncode == 0: last_result = {"method": "dvt --rsd (cli)", "ok": True, "error": None} return True except (subprocess.TimeoutExpired, Exception): _cached_rsd_host = None last_result = {"method": None, "ok": False, "error": "all methods failed"} return False import random def add_gps_jitter(lat, lng): meter_lat = 1.0 / 111320.0 meter_lng = 1.0 / (111320.0 * max(abs(math.cos(math.radians(lat))), 0.001)) jitter_m = 2.0 return ( lat + random.uniform(-jitter_m, jitter_m) * meter_lat, lng + random.uniform(-jitter_m, jitter_m) * meter_lng ) def set_location(lat, lng): jlat, jlng = add_gps_jitter(lat, lng) if not send_via_daemon(jlat, jlng): send_location_cli(jlat, jlng) _target_lat = current_coords.get("lat") _target_lng = current_coords.get("lng") _target_updated = time.time() _target_lock = threading.Lock() _last_sent_lat = None _last_sent_lng = None def location_loop(): global _last_sent_lat, _last_sent_lng while True: with _target_lock: lat = _target_lat lng = _target_lng if lat is None or lng is None: time.sleep(0.1) continue if lat != _last_sent_lat or lng != _last_sent_lng: set_location(lat, lng) _last_sent_lat = lat _last_sent_lng = lng else: time.sleep(0.1) def set_target(lat, lng): global _target_lat, _target_lng, _target_updated with _target_lock: _target_lat = lat _target_lng = lng _target_updated = time.time() class Handler(BaseHTTPRequestHandler): def handle_error(self, request, client_address): import sys as _sys exc = _sys.exc_info()[1] if isinstance(exc, (BrokenPipeError, ConnectionResetError)): return super().handle_error(request, client_address) def log_message(self, format, *args): pass def send_json(self, data, status=200): body = json.dumps(data, indent=2).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def do_GET(self): path = urlparse(self.path).path if path == "/coords": return self.send_json(current_coords) if path == "/status": return self.send_json({"ok": True, "spoof": last_result}) if path in ("/", "/index.html"): path = "/index.html" file_path = os.path.join(BASE_DIR, path.lstrip("/")) if os.path.isfile(file_path): mime, _ = mimetypes.guess_type(file_path) with open(file_path, "rb") as f: body = f.read() self.send_response(200) self.send_header("Content-Type", mime or "application/octet-stream") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) else: self.send_response(404) self.end_headers() def do_POST(self): global current_coords if urlparse(self.path).path == "/move": length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) try: data = json.loads(body) lat = float(data["lat"]) lng = float(data["lng"]) current_coords = {"lat": lat, "lng": lng} save_coords(lat, lng) set_target(lat, lng) self.send_json({"ok": True, "lat": lat, "lng": lng}) except Exception as e: self.send_json({"error": str(e)}, 400) else: self.send_response(404) self.end_headers() if __name__ == "__main__": print(f"PoGo Mover running at http://localhost:{PORT}") print(" Starting location daemon...", flush=True) start_daemon() threading.Thread(target=location_loop, daemon=True).start() server = ThreadedHTTPServer(("localhost", PORT), Handler) server.serve_forever()