oscpy 다운로드 - oscpy 소스 코드 다운로드

oscpy

기타 카테고리

1.0.0

다운로드

OSCPy

CI는 Github Checks에 의해 수행됩니다. 빌드 상태는 현재 커밋을 참조하세요.

Python2/3용 OSC의 최신 구현입니다.

OSC 란 무엇입니까?

OpenSoundControl은 UDP 기반 네트워크 프로토콜로, 이름에서 알 수 있듯이 시간에 민감한 메시지를 빠르게 전달하도록 설계되었으며 MIDI를 대체하도록 설계되었지만 다른 상황에도 잘 적용됩니다. 프로토콜은 사용하기 간단하며, OSC 주소는 http URL처럼 보이고 string, float, int 등과 같은 다양한 기본 유형을 허용합니다. 기본적으로 오버헤드가 적은 http POST로 생각할 수 있습니다.

OpenSoundControl.org에서 OSC에 대해 자세히 알아볼 수 있습니다.

목표

python2.7/3.6+ 호환성(필요한 경우 python3 측에서 더 완화될 수 있지만 2.7 이전 버전은 지원되지 않음)
빠른
사용하기 쉬운
견고함(잘못된 메시지의 경우 의미 있는 오류를 반환하고, 올바른 메시지에 대해 항상 올바른 작업을 수행하며, 기본적으로 콜백에서 발생한 예외를 가로채고 기록함)
관심사 분리(메시지 구문 분석과 통신)
동기 및 비동기 호환성(스레드, asyncio, 트리오…)
깔끔하고 읽기 쉬운 코드

특징

OSC 데이터 유형/메시지/번들을 직렬화하고 구문 분석합니다.
소켓(INET 또는 UNIX)을 열고 해당 소켓의 osc 주소에 콜백을 바인딩하는 스레드 기반 UDP 서버
간단한 클라이언트

설치하다

pip install oscpy

용법

서버(스레드)

 from oscpy . server import OSCThreadServer
from time import sleep

def callback ( * values ):
    print ( "got values: {}" . format ( values ))

osc = OSCThreadServer ()  # See sources for all the arguments

# You can also use an *nix socket path here
sock = osc . listen ( address = '0.0.0.0' , port = 8000 , default = True )
osc . bind ( b'/address' , callback )
sleep ( 1000 )
osc . stop ()  # Stop the default socket

osc . stop_all ()  # Stop all sockets

# Here the server is still alive, one might call osc.listen() again

osc . terminate_server ()  # Request the handler thread to stop looping

osc . join_server ()  # Wait for the handler thread to finish pending tasks and exit

또는 데코레이터 API를 사용할 수 있습니다.

서버(스레드)

 from oscpy . server import OSCThreadServer
from time import sleep

osc = OSCThreadServer ()
sock = osc . listen ( address = '0.0.0.0' , port = 8000 , default = True )

@ osc . address ( b'/address' )
def callback ( * values ):
    print ( "got values: {}" . format ( values ))

sleep ( 1000 )
osc . stop ()

서버는 클라이언트이기도 합니다. 즉, 메시지를 보내고 다른 서버의 메시지에 응답할 수 있습니다.

 from oscpy . server import OSCThreadServer
from time import sleep

osc_1 = OSCThreadServer ()
osc_1 . listen ( default = True )

@ osc_1 . address ( b'/ping' )
def ping ( * values ):
    print ( "ping called" )
    if True in values :
        cont . append ( True )
    else :
        osc_1 . answer ( b'/pong' )

osc_2 = OSCThreadServer ()
osc_2 . listen ( default = True )

@ osc_2 . address ( b'/pong' )
def pong ( * values ):
    print ( "pong called" )
    osc_2 . answer ( b'/ping' , [ True ])

osc_2 . send_message ( b'/ping' , [], * osc_1 . getaddress ())

timeout = time () + 1
while not cont :
    if time () > timeout :
        raise OSError ( 'timeout while waiting for success message.' )

서버(비동기) (해야 할 일!)

 from oscpy . server import OSCThreadServer

with OSCAsyncServer ( port = 8000 ) as OSC :
    for address , values in OSC . listen ():
       if address == b'/example' :
            print ( "got {} on /example" . format ( values ))
       else :
            print ( "unknown address {}" . format ( address ))

고객

 from oscpy . client import OSCClient

address = "127.0.0.1"
port = 8000

osc = OSCClient ( address , port )
for i in range ( 10 ):
    osc . send_message ( b'/ping' , [ i ])

유니코드

기본적으로 서버와 클라이언트는 osc 주소와 osc 문자열에 대해 유니코드 문자열이 아닌 바이트(인코딩된 문자열)를 사용합니다. 그러나 encoding 매개변수를 전달하여 문자열을 자동으로 인코딩하고 디코딩할 수 있으므로 콜백은 유니코드 문자열(python2의 경우 유니코드, python3의 경우 str)을 가져옵니다.

 osc = OSCThreadServer ( encoding = 'utf8' )
osc . listen ( default = True )

values = []

@ osc . address ( u'/encoded' )
def encoded ( * val ):
    for v in val :
        assert not isinstance ( v , bytes )
    values . append ( val )

send_message (
    u'/encoded' ,
    [ u'hello world' , u'ééééé ààààà' ],
    * osc . getaddress (), encoding = 'utf8' )