如何在 OBS 中将 python-osc 作为脚本运行？

编程问答 2022-05-30

问题描述

我希望/需要将此代码作为脚本在 OBS 内运行。我把这个 OSC Sender for OBS 当作它的名字所说的那样：它发送 OSC 消息。但是，在代码内部，对服务器部分进行了注释，因为它不起作用……开箱即用。需要使用 TextureCube CubeMap: register(t0); SamplerState TexSampler : register(s0); float4 main(LightingPixelShaderInput input) : SV_Target { float4 cubeTexture = CubeMap.Sample(TexSampler,input.worldnormal); //light calculations float3 finalColour = (gAmbientColour + diffuseLights) * cubeTexture.rgb + (specularLights) * cubeTexture.a; return float4(finalColour,1.0f); } 运行 OBS。

我修改了代码但是（未注释的行）但是，当服务器打开时，OBS 仍然无法使用，被阻止。所以，我尝试了 python-osc 的“异步服务器”-“并发模式”。我以 its wiki 中的示例（它在控制台中工作）为例，并与其他脚本混合。我做了一个很好的代码......但它使一些奇怪的事情。这是代码：

LD_PRELOAD=/path/libpython3.7.so

我不是程序员，所以 asyncio 对我来说是一团糟。我不明白。但我知道它做什么以及我希望它做什么。

加载脚本后（在 """ """ # osc data import argparse import random import time import math import asyncio from pythonosc import osc_message_builder from pythonosc import udp_client from pythonosc.dispatcher import dispatcher from pythonosc.osc_server import AsyncIOOSCUDPServer client = None server = None #obs ! import obspython as obs pleaseLog = False host = None serverPort = None # transport = None async def loop(): """Example main loop that only runs for 10 iterations before finishing""" for i in range(10): print(f"Loop {i}") await asyncio.sleep(.5) async def init_main(): # global transport server = AsyncIOOSCUDPServer((host,serverPort),dispatcher,asyncio.get_event_loop()) transport,protocol = await server.create_serve_endpoint() # Create datagram endpoint and start serving await loop() # Enter main loop of program transport.close() # Clean up serve endpoint def handleOSC(*args): for i in args: print(i) # defines script description def script_description(): return '''Send OSC data when source is activated if source name begins with /''' # defines user properties def script_properties(): global props props = obs.obs_properties_create() obs.obs_properties_add_text(props,"host","Host IP",obs.OBS_TEXT_DEFAULT) obs.obs_properties_add_int(props,"clientPort","Host port",1,400000,1) obs.obs_properties_add_bool(props,"logoscOutput","Log OSC output") obs.obs_properties_add_int(props,"serverPort","Listen port",1) return props def script_defaults(settings): obs.obs_data_set_default_string(settings,"127.0.0.1") obs.obs_data_set_default_int(settings,10000) obs.obs_data_set_default_int(settings,10001) def source_activated(cd): global pleaseLog source = obs.calldata_source(cd,"source") if source is not None: name = obs.obs_source_get_name(source) if name[0] == "/": client.send_message(name,1) if (pleaseLog): print("send " + name) def script_load(settings): global dispatcher global host sh = obs.obs_get_signal_handler() obs.signal_handler_connect(sh,"source_activate",source_activated) dispatcher = dispatcher() dispatcher.map("/*",handleOSC) # def script_unload(): # global transport # print(f'script_unload(settings)') # transport.close() # Clean up serve endpoint def script_update(settings): global host global client global server global clientPort global serverPort global pleaseLog pleaseLog = obs.obs_data_get_bool(settings,"logoscOutput") host = obs.obs_data_get_string(settings,"host") clientPort = obs.obs_data_get_int(settings,"clientPort") serverPort = obs.obs_data_get_int(settings,"serverPort") # Client client = udp_client.SimpleUDPClient(host,clientPort) print("target set to "+host+":"+str(clientPort)+"") # Server print("serving in "+host+":"+str(serverPort)+"") asyncio.run(init_main()) 中），它打开服务器并运行 script_update(settings) 函数。此函数是一个 5 秒循环，用于打印一些文本。如果我从 PureData 连接到正确的端口并发送一些 OSC 消息，这些消息将到达 OBS...同时，什么都没有，OBS 被阻止，脚本日志中也没有任何内容。

如果我在控制台中运行 python-osc asyncio 示例代码，当循环正在执行时，每个 loop()、Loop 0 等，以及每条消息到达，并且所有消息都打印在正确的时间。

我应该如何让这段代码工作？我需要打开该服务器并同时运行更多代码。我正在像使用游戏引擎一样使用 OBS。

解决方法

最后，我没有使用 Asyncio。取而代之的是，我使用 Blocking Server 方法，但在 threading.Thread() 函数内部。目前，它以这种方式正常工作。

卸载或重新加载脚本时关闭端口很重要。

Oro、Treblig_Punisher（来自 OBS Discor 的用户）帮助我实现了这一目标。

"""

"""
# osc data

import argparse,random,time,math,threading

from pythonosc import osc_message_builder
from pythonosc import udp_client

from pythonosc import dispatcher
from pythonosc import osc_server

targetIp = "127.0.0.1"
targetPort = 10000
serverPort = 10008

client = None
server = None

#obs !

import obspython as obs
pleaseLog = False

def handleOSC(address,args,data):
    print (address)
    print (args)
    print (data)

# defines script description 
def script_description():
    return '''Send and receive OSC messages'''

# defines user properties
def script_properties():
    #global props 
    props = obs.obs_properties_create()
    obs.obs_properties_add_text(props,"host","Host IP",obs.OBS_TEXT_DEFAULT)
    obs.obs_properties_add_int(props,"port","Host port",1,400000,1)
    obs.obs_properties_add_bool(props,"logOscOutput","Log OSC output")
    obs.obs_properties_add_int(props,"serverPort","Listen port",1)
    return props

def script_defaults(settings):
    obs.obs_data_set_default_string(settings,targetIp)
    obs.obs_data_set_default_int(settings,targetPort)
    obs.obs_data_set_default_int(settings,serverPort)

# Cuando se activa un source
def source_activated(cd):
    global pleaseLog
    source = obs.calldata_source(cd,"source")
    if source is not None:
        name = obs.obs_source_get_name(source)
        
        current_scene = obs.obs_scene_from_source(obs.obs_frontend_get_current_scene())
        scene_item = obs.obs_scene_find_source(current_scene,name)
        boolean = obs.obs_sceneitem_visible(scene_item)
        
        print(boolean)
        if name[0] == "/":
            client.send_message(name,boolean)
            if (pleaseLog):
                print("send " + name + boolean)

def script_load(settings):
    global despachante
    
    sh = obs.obs_get_signal_handler()
    obs.signal_handler_connect(sh,"source_activate",source_activated)

    despachante = dispatcher.Dispatcher()
    despachante.map("/*",handleOSC)

def script_unload():
    global server
    server.server_close()

def script_update(settings):
    global host
    global port
    global client
    global server
    global pleaseLog

    pleaseLog = obs.obs_data_get_bool(settings,"logOscOutput")
    host = obs.obs_data_get_string(settings,"host")
    port = obs.obs_data_get_int(settings,"port")

    # Client
    client = udp_client.SimpleUDPClient(host,port)
    print("target set to "+host+":"+str(port)+"")

    # Server
    serverPort = obs.obs_data_get_int(settings,"serverPort")
    try:
        server.server_close()
    except:
        print('*Server aun no creado')
        # raise
    server = osc_server.BlockingOSCUDPServer(("127.0.0.1",serverPort),despachante)
    t = threading.Thread(target=server_th,args=([settings]))
    t.daemon = True
    t.start()

    # Loop every 1000ms
    # obs.timer_remove(funcion_loop)
    # if and source_name != "":
        # obs.timer_add(funcion_loop,1000)

# Threading function
def server_th(settings):
    print(f'Serving on {server.server_address}')
    server.serve_forever()  # Blocks forever
obsoscpythonpython-3.xpython-asyncio