Created
April 27, 2023 08:08
-
-
Save catboy1357/ff382800d2773c6fe330a5f8d8445fee to your computer and use it in GitHub Desktop.
A simple script made to log all the OSC data from VRChat into a file.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from pythonosc.dispatcher import Dispatcher | |
from pythonosc.osc_server import BlockingOSCUDPServer | |
from os import path | |
from time import time | |
from datetime import datetime | |
class OSCLogger: | |
# Print message to console when the class is initiated | |
print("Starting! use Ctrl+C to exit") | |
# Set up constants | |
FILENAME = "Log.txt" | |
FILEPATH = path.exists(FILENAME) | |
# Constructor method | |
def __init__(self, ip = "127.0.0.1", port = 9001, update_rate = 0.5, update_limit = 5): | |
# Set up instance variables | |
self.last_addresses = [] | |
self.last_update = 0 | |
self.update_rate = update_rate # in seconds | |
self.limit_addresses_len = update_limit # messages limited before before update | |
# Check for log file and clear it if it exists | |
self.file_check() | |
# Set up OSC server and dispatcher | |
dispatcher = Dispatcher() | |
dispatcher.set_default_handler(self.rate_limit) | |
self.server = BlockingOSCUDPServer((ip, port), dispatcher) | |
# Method to check if log file exists and clear it if it does | |
def file_check(self): | |
if self.FILEPATH: | |
print(f"The file {self.FILENAME} exists in directory and has been cleared.") | |
else: | |
print(f"The file {self.FILENAME} did not exist, a new file was created.") | |
# Get current date and time | |
time = datetime.now() | |
time.strftime("%Y-%m-%d %H:%M:%S") | |
# Write init message to log file | |
with open(self.FILENAME, "w") as f: | |
f.write(f"START {time}\n") | |
# Method to log incoming OSC messages to file and console | |
def logger(self, address, *args): | |
print(f"GET {address}: {args}") | |
# Get path to log file and append message to file | |
log_file = path.join(path.dirname(__file__), self.FILENAME) | |
with open(log_file, "a") as f: | |
f.write(f"Address: {address:<40} Args: {args}\n") | |
# Method to rate limit incoming OSC messages and call logger method | |
def rate_limit(self, address, *args): | |
now = time() | |
# If the time elapsed since the last update is less than the update rate | |
if now - self.last_update < self.update_rate: | |
# If the current address is in the list of last addresses, return and don't log | |
if address in self.last_addresses: | |
return | |
else: | |
# clear the list of last addresses after time elapsed | |
self.last_addresses = [] | |
self.last_update = now | |
self.last_addresses.append(address) | |
# If the list of last addresses exceeds the limit, remove the oldest address | |
if len(self.last_addresses) > self.limit_addresses_len: | |
self.last_addresses.pop(0) | |
# Log the current address and arguments | |
self.logger(address, *args) | |
# Method to start OSC server and listen for incoming messages | |
def run(self): | |
try: | |
self.server.serve_forever() | |
# Catch Ctrl+C interrupt and print exit message to console | |
except KeyboardInterrupt: | |
print("Exiting...") | |
if __name__ == '__main__': | |
# Create OSCLogger object and start server | |
# You can change how this updates using OSCLogger update_rate or update_limit | |
server = OSCLogger() | |
server.run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment