Integrate OBS and Shure Microflex LEDs and mute control

shure_led_control.py

import obspython as obs
import socket
import threading

# --- Defaults ---
DEFAULT_IP = "192.168.1.100"
DEFAULT_PORT = 2202  # Fixed port for Shure devices
DEFAULT_MAPPING = "Mic 1:1, Mic 2:2"

# --- State ---
TOGGLE_MUTE_FROM_BUTTON = True
SET_LED_ON_MUTE = True
TELNET_HOST = DEFAULT_IP
TELNET_PORT = DEFAULT_PORT
CHANNEL_MAP = {}  # OBS source name -> channel number
REVERSE_CHANNEL_MAP = {}  # channel number -> OBS source name

# --- Telnet Listener Thread ---
stop_telnet_thread = threading.Event()
telnet_thread = None

def telnet_listener():
    global TELNET_HOST, TELNET_PORT
    obs.script_log(obs.LOG_INFO, f"Connecting to Telnet server at {TELNET_HOST}:{TELNET_PORT}...")
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        sock.connect((TELNET_HOST, TELNET_PORT))
        obs.script_log(obs.LOG_INFO, "Connected to Telnet server.")
        sock.settimeout(1.0)

        buffer = ""
        while not stop_telnet_thread.is_set():
            try:
                data = sock.recv(1024)
                if not data:
                    break
                buffer += data.decode('ascii', errors='ignore')
                while '<' in buffer and '>' in buffer:
                    start = buffer.index('<')
                    end = buffer.index('>', start)
                    message = buffer[start+1:end].strip()
                    buffer = buffer[end+1:]
                    handle_telnet_message(message)
            except socket.timeout:
                continue
            except Exception as e:
                obs.script_log(obs.LOG_WARNING, f"Telnet recv error: {e}")
                break
        try:
            sock.shutdown(socket.SHUT_RDWR)
        except Exception:
            pass
        sock.close()
    except Exception as e:
        obs.script_log(obs.LOG_WARNING, f"Telnet connection error: {e}")

def handle_telnet_message(message):
    if "BUTTON_STS" in message or "LED_STATUS" in message:
        obs.script_log(obs.LOG_INFO, f"Received Telnet message: {message}")
    if message.startswith("REP") and "BUTTON_STS" in message:
        parts = message.split()
        try:
            ch_index = parts.index("REP") + 1
            channel = int(parts[ch_index])
            status = parts[-1]  # ON or OFF
            source_name = REVERSE_CHANNEL_MAP.get(channel)
            if source_name and status == "OFF":  # Only act on BUTTON_STS OFF
                source = obs.obs_get_source_by_name(source_name)
                if source and TOGGLE_MUTE_FROM_BUTTON:
                    is_muted = obs.obs_source_muted(source)
                    obs.obs_source_set_muted(source, not is_muted)
                    if SET_LED_ON_MUTE:
                        send_telnet_command(channel, not is_muted)
                    obs.obs_source_release(source)
                    obs.obs_source_release(source)
        except Exception as e:
            obs.script_log(obs.LOG_WARNING, f"Error parsing telnet message: {e}")

# --- Telnet Command ---
def send_telnet_command(channel, muted):
    try:
        command = f"< SET {channel} LED_STATUS {'ON OF' if muted else 'OF ON'} >\r\n"
        with socket.create_connection((TELNET_HOST, TELNET_PORT), timeout=2) as sock:
            sock.sendall(command.encode('ascii'))
    except Exception as e:
        obs.script_log(obs.LOG_WARNING, f"Telnet error for channel {channel}: {e}")

# --- Parse Mapping String ---
def parse_channel_map(mapping_str):
    parsed = {}
    reverse = {}
    entries = mapping_str.split(',')
    for entry in entries:
        if ':' in entry:
            source, ch = entry.strip().split(':', 1)
            try:
                ch_num = int(ch.strip())
                source_clean = source.strip()
                parsed[source_clean] = ch_num
                reverse[ch_num] = source_clean
            except ValueError:
                obs.script_log(obs.LOG_WARNING, f"Invalid channel number for '{entry.strip()}'")
    return parsed, reverse

# --- Mute/Unmute Callback ---
def on_source_mute(calldata):
    source = obs.calldata_source(calldata, "source")
    if source:
        source_name = obs.obs_source_get_name(source)
        if source_name in CHANNEL_MAP:
            channel = CHANNEL_MAP[source_name]
            muted = obs.obs_source_muted(source)
            if SET_LED_ON_MUTE:
                send_telnet_command(channel, muted)

# --- OBS Hooks ---
def connect_signals():
    for source_name in CHANNEL_MAP:
        source = obs.obs_get_source_by_name(source_name)
        if source:
            handler = obs.obs_source_get_signal_handler(source)
            obs.signal_handler_connect(handler, "mute", on_source_mute)
            obs.signal_handler_connect(handler, "unmute", on_source_mute)
            obs.obs_source_release(source)

# --- OBS Script Lifecycle ---
def script_load(settings):
    connect_signals()

def script_update(settings):
    global TOGGLE_MUTE_FROM_BUTTON, SET_LED_ON_MUTE, TELNET_HOST, CHANNEL_MAP, REVERSE_CHANNEL_MAP, telnet_thread, stop_telnet_thread
    global TELNET_HOST, CHANNEL_MAP, REVERSE_CHANNEL_MAP, telnet_thread, stop_telnet_thread

    TOGGLE_MUTE_FROM_BUTTON = obs.obs_data_get_bool(settings, "toggle_mute")
    SET_LED_ON_MUTE = obs.obs_data_get_bool(settings, "set_led")

    TELNET_HOST = obs.obs_data_get_string(settings, "telnet_ip")
    channel_map_str = obs.obs_data_get_string(settings, "channel_map")
    CHANNEL_MAP, REVERSE_CHANNEL_MAP = parse_channel_map(channel_map_str)

    stop_telnet_thread.clear()
    telnet_thread = threading.Thread(target=telnet_listener, daemon=True)
    telnet_thread.start()

    connect_signals()

def script_unload():
    global stop_telnet_thread, telnet_thread
    stop_telnet_thread.set()
    if telnet_thread is not None:
        telnet_thread.join(timeout=2)

def script_defaults(settings):
    obs.obs_data_set_default_bool(settings, "toggle_mute", True)
    obs.obs_data_set_default_bool(settings, "set_led", True)
    obs.obs_data_set_default_string(settings, "telnet_ip", DEFAULT_IP)
    obs.obs_data_set_default_string(settings, "channel_map", DEFAULT_MAPPING)

def script_properties():
    props = obs.obs_properties_create()
    obs.obs_properties_add_bool(props, "toggle_mute", "Toggle OBS Mute From Mic Button")
    obs.obs_properties_add_bool(props, "set_led", "Update LED Based On OBS Mute")
    obs.obs_properties_add_text(props, "telnet_ip", "Shure Device IP", obs.OBS_TEXT_DEFAULT)
    obs.obs_properties_add_text(props, "channel_map", "Channel Map (e.g. Mic 1:1, Mic 2:2)", obs.OBS_TEXT_DEFAULT)
    return props

def script_description():
    return "Controls Shure Microflex LEDs via Telnet based on OBS mute state.\nAlso listens for microphone button presses to update mute state in OBS. Port is fixed to 2202."