import os import glob import re import requests import time import threading from pythonosc import udp_client from flask import Flask, jsonify from flask_cors import CORS from pypresence import Presence, ActivityType app = Flask(__name__) CORS(app) last_known_health = "" last_known_map = "" current_map_name = "reqMaster" last_file_size = 0 last_file_path = "" display_output = "" last_overtime_trigger = 0 def trigger_damage_flash(): # Flashes damage indicator on avatar (change "Damage" to whatever you want to flash in your avatar) try: udp_client.SimpleUDPClient("127.0.0.1", 9000).send_message("/avatar/parameters/Damage", True) time.sleep(0.05) udp_client.SimpleUDPClient("127.0.0.1", 9000).send_message("/avatar/parameters/Damage", False) except Exception as e: print(f"OSC Emote Error: {e}") def check_and_trigger(clean_line): global last_known_health, last_known_map, current_map_name, last_overtime_trigger if "SLASHCO Game setup." in clean_line: # Makes health and map visible last_known_health = "HP: [10]" last_known_map = f"Map: [{current_map_name}]" health_match = re.search(r"Current user health: (\d+)", clean_line) # Grabs health for html and Discord RPC if health_match: last_known_health = f"HP: [{health_match.group(1)}]" damage_match = re.search(r"DMG TAKEN; raw: (\w+)", clean_line) # Output damage log to console if damage_match: threading.Thread(target=trigger_damage_flash, daemon=True).start() if "Player Deathed." in clean_line: last_known_health = "Deathed" map_match = re.search(r"Selected landing spot on map (\w+)", clean_line) # Grabs map name for html and Discord RPC if map_match: raw_map = map_match.group(1) current_map_name = re.sub(r'([A-Z])', r' \1', raw_map).strip() last_known_map = f"Map: [{current_map_name}]" if "Game Overtime Chase." in clean_line: # Displays OVERTIME on html last_overtime_trigger = time.time() if "SLASHCO Client STOP GAME" in clean_line: # Clears text on html last_known_health = "" last_known_map = "" triggers = [ # Uses YouTube Music Client https://github.com/pear-devs/pear-desktop (["Player Deathed."], "Player Deathed. PLAY", "http://localhost:26538/api/v1/play"), # Play music at death (["SLASHCO Client STOP GAME"], "SLASHCO Client STOP GAME PLAY", "http://localhost:26538/api/v1/play"), # Play music at game end (["SLASHCO Game setup."], "SLASHCO Game setup. PAUSE", "http://localhost:26538/api/v1/pause"), # Pause game at game start ] for keywords, display_name, target_url in triggers: if any(word in clean_line for word in keywords): try: payload = {"event": "LogTriggered", "message": clean_line, "type": display_name} requests.post(target_url, json=payload, timeout=1) except Exception as e: print(f"{display_name} failed: {e}") def process_logs(): global last_file_size, last_file_path, display_output path = r"C:\Users\Riletin\AppData\LocalLow\VRChat\VRChat" # -----------------------------------↓ list_of_files = glob.glob(os.path.join(path, 'output_log_*.txt')) # searches for latest log txt file if not list_of_files: return latest_file = max(list_of_files, key=os.path.getctime) if latest_file != last_file_path: last_file_path = latest_file last_file_size = os.path.getsize(latest_file) try: current_size = os.path.getsize(latest_file) if current_size > last_file_size: with open(latest_file, "rb") as f: f.seek(last_file_size) new_chunk = f.read().decode('utf-8', errors='ignore') last_file_size = current_size lines = new_chunk.strip().splitlines() timestamp_regex = r'^\d{4}\.\d{2}\.\d{2}\s\d{2}:\d{2}:\d{2}\s\w+\s+-\s+' # removes timestamp from logs filters = ["[API]", "[EOSManager]", "[AssetBundleDownloadManager]"] # Filter out logs including for raw_line in lines: if any(f.lower() in raw_line.lower() for f in filters): continue clean_line = re.sub(timestamp_regex, '', raw_line) check_and_trigger(clean_line) with open(latest_file, "rb") as f: f.seek(0, os.SEEK_END) # grab the latest log f.seek(max(0, f.tell() - 4096)) end_chunk = f.read().decode('utf-8', errors='ignore').strip().splitlines() display_lines = [] for l in end_chunk: if not any(f.lower() in l.lower() for f in ["[api]", "[eosmanager]", "[assetbundledownloadmanager]"]): display_lines.append(re.sub(r'^\d{4}.*?\s-\s', '', l)) display_output = "\n".join(display_lines[-3:]) # appened's new logs / amount of logs to display in html except Exception as e: print(f"Log: {e}") def fast_log_monitor(): # makes log faster to output in html while True: process_logs() time.sleep(0.1) def discord_rpc(): # Discord Rich Presence try: RPC = Presence("1472107031874895956") RPC.connect() except Exception as e: print(f"Discord: {e}") return while True: try: process_logs() RPC.update( details=last_known_health.strip() if last_known_health else "In Lobby", state=last_known_map.strip() if last_known_map else "initializing...", large_image="slashco", activity_type=ActivityType.COMPETING ) time.sleep(5) except Exception as e: print(f"Main: {e}") time.sleep(5) # http://127.0.0.1:27210/@app.route @app.route('/ControlPanel') # VR Overlay Controler in html def ctrl_pnl(): with open("ControlPanel.html", "r") as f: return f.read() @app.route('/vr') # VR Overlay def vr(): with open("vr.html", "r") as f: return f.read() @app.route('/set_map/') # use in ControlPanel def set_map(name): global last_known_map last_known_map = f"Map: [{name}]" return "OK" @app.route('/test_flash') # use in ControlPanel def test_flash(): threading.Thread(target=trigger_damage_flash, daemon=True).start() return "Damage" @app.route('/test_overtime') # use in ControlPanel def test_overtime(): global last_overtime_trigger last_overtime_trigger = time.time() return "Overtime" @app.route('/info') # Json API def info(): process_logs() return jsonify({ "line": display_output, "health": last_known_health, "map": last_known_map, "overtime_stamp": last_overtime_trigger }) @app.route('/') # Use in OBS FPS Limit 25fps def slashco(): with open("SlashCo.html", "r") as f: return f.read() if __name__ == '__main__': threading.Thread(target=discord_rpc, daemon=True).start() app.run(host='127.0.0.1', port=27210, debug=False)