You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
144 lines
4.9 KiB
144 lines
4.9 KiB
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
OSC-2-BAT
|
|
--------------------------
|
|
v0.1.0
|
|
--------------------------
|
|
|
|
OSC server that receives commands on /script/* and runs batch files
|
|
|
|
-i [IP ADDRESS] --ip [IP ADDRESS] IP Address to listen on
|
|
-p [PORT] --port [PORT] Port to listen on
|
|
-s [FOLDER] --scripts [SCRIPTS] Folder containing batch script files
|
|
-l [LEVEL] --log-level [LEVEL] Log Level [CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET]
|
|
-nl --no-logging Disable logging
|
|
|
|
"""
|
|
|
|
"""-------------------------------------------------------------------------------------------
|
|
-- IMPORTS
|
|
-------------------------------------------------------------------------------------------"""
|
|
|
|
# ARGUMENT PARSING
|
|
import argparse
|
|
|
|
# PYTHON OSC
|
|
from pythonosc.dispatcher import Dispatcher
|
|
from pythonosc.osc_server import BlockingOSCUDPServer
|
|
|
|
# SUBPROCESS
|
|
import os
|
|
import subprocess
|
|
from subprocess import Popen
|
|
|
|
# LOGGING
|
|
import logging
|
|
import datetime
|
|
|
|
"""-------------------------------------------------------------------------------------------
|
|
-- ARGUMENT PARSING
|
|
-------------------------------------------------------------------------------------------"""
|
|
|
|
parser = argparse.ArgumentParser()
|
|
|
|
parser.add_argument("-i", "--ip", dest="ip", default="127.0.0.1", help="IP Address to listen on")
|
|
parser.add_argument("-p", "--port", dest="port", default="1337", type=int, help="Port to listen on")
|
|
parser.add_argument("-s", "--scripts", dest="scripts", default=".", help="Folder containing batch script files")
|
|
parser.add_argument("-l", "--log-level", dest="loglevel", default='info', help="Log Level [CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET]")
|
|
parser.add_argument("-nl", "--no-logging", dest="logaction", action='store_false', help="Disable logging")
|
|
|
|
args = parser.parse_args()
|
|
|
|
server_ip = args.ip
|
|
server_port = args.port
|
|
scripts_path = args.scripts
|
|
logging_level = args.loglevel
|
|
|
|
batch_files = []
|
|
|
|
"""-------------------------------------------------------------------------------------------
|
|
-- LOGGING
|
|
-------------------------------------------------------------------------------------------"""
|
|
|
|
# LOGGING FILE LOCATION
|
|
log_filename = "osc-2-script." + datetime.datetime.now().strftime('%m-%d-%Y') + ".log"
|
|
|
|
# SET LEVEL
|
|
return_loglevel_error = False
|
|
if logging_level.upper() not in list(logging._nameToLevel.keys()):
|
|
return_loglevel_error = True
|
|
logging_level = 'WARNING'
|
|
|
|
if args.logaction:
|
|
logging.basicConfig(filename=log_filename, level=logging_level.upper(), format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%d-%b-%y %H:%M:%S')
|
|
|
|
if return_loglevel_error:
|
|
print(f"{args.loglevel.upper()} not a valid logging level - defaulting to WARNING")
|
|
if args.logaction:
|
|
logging.warning(f"{args.loglevel.upper()} not a valid logging level - defaulting to WARNING")
|
|
|
|
print(f"Logging level set to {logging_level}")
|
|
if args.logaction:
|
|
logging.info(f"Logging level set to {logging_level}")
|
|
print(f"Logging to file: {log_filename}")
|
|
if args.logaction:
|
|
logging.info(f"Logging to file: {log_filename}")
|
|
|
|
"""-------------------------------------------------------------------------------------------
|
|
-- SERVER - HANDLER FUNCTIONS
|
|
-------------------------------------------------------------------------------------------"""
|
|
|
|
# DEFAULT HANDLER
|
|
|
|
def default_handler(address, *args):
|
|
print(f"DEFAULT - {address}: {args}")
|
|
|
|
# SCRIPT HANDLER
|
|
|
|
def script_handler(address, *args):
|
|
|
|
if args.logaction:
|
|
logging.debug("Running Script Handler")
|
|
|
|
script_to_run = address.split("/")
|
|
|
|
script_name = script_to_run[-1]
|
|
script_location = script_to_run[2:-1]
|
|
|
|
print(f"SCRIPT LOCATION: {script_location} - SCRIPT NAME {script_name}")
|
|
if args.logaction:
|
|
logging.debug(f"SCRIPT LOCATION: {script_location} - SCRIPT NAME: {script_name}")
|
|
|
|
file_path = os.path.join(scripts_path, *script_location, script_name)
|
|
|
|
if os.path.isfile(file_path):
|
|
print(f"RUNNING SCRIPT: {file_path}")
|
|
if args.logaction:
|
|
logging.info(f"RUNNING SCRIPT: {file_path}")
|
|
Popen(file_path,creationflags=subprocess.CREATE_NEW_CONSOLE)
|
|
else:
|
|
print(f"FILE NOT FOUND: {file_path}")
|
|
if args.logaction:
|
|
logging.error(f"FILE NOT FOUND: {file_path}")
|
|
|
|
"""-------------------------------------------------------------------------------------------
|
|
-- SERVER
|
|
-------------------------------------------------------------------------------------------"""
|
|
|
|
# DISPATCHER
|
|
dispatcher = Dispatcher()
|
|
|
|
dispatcher.map("/script/*", script_handler)
|
|
dispatcher.set_default_handler(default_handler)
|
|
|
|
# SERVER START
|
|
if __name__ == '__main__':
|
|
|
|
print(f"Starting server on interface {server_ip} port {server_port}")
|
|
if args.logaction:
|
|
logging.info(f"Starting server on interface {server_ip} port {server_port}")
|
|
|
|
server = BlockingOSCUDPServer((server_ip, server_port), dispatcher)
|
|
server.serve_forever() |