Загрузка oscpy - Загрузка исходного кода oscpy

oscpy

Другие категории

1.0.0

Скачать

OSCPy

CI выполняется с помощью проверок Github, статус сборки смотрите в текущем коммите.

Современная реализация OSC для Python2/3.

Что такое ОСК.

OpenSoundControl — это сетевой протокол на основе UDP, предназначенный для быстрой отправки чувствительных ко времени сообщений. Как следует из названия, он был разработан как замена MIDI, но хорошо применим в других ситуациях. Протокол прост в использовании, адреса OSC выглядят как URL-адреса http и принимают различные базовые типы, такие как строка, число с плавающей запятой, целое число и т. д. По сути, его можно рассматривать как http POST с меньшими накладными расходами.

Вы можете узнать больше об OSC на OpenSoundControl.org.

Цели

совместимость с python2.7/3.6+ (при необходимости можно ослабить поддержку Python3, но версии до версии 2.7 поддерживаться не будут)
быстрый
простой в использовании
надежный (возвращает значимые ошибки в случае неверных сообщений, всегда поступает правильно с правильными сообщениями и по умолчанию перехватывает + регистрирует исключения, вызванные обратными вызовами)
разделение задач (анализ сообщений и общение)
совместимость синхронизации и асинхронности (threads, asyncio, trio…)
чистый и легко читаемый код

Функции

сериализовать и анализировать типы данных OSC/сообщения/пакеты
udp-сервер на основе потоков для открытия сокетов (INET или UNIX) и привязки обратных вызовов к адресам OSC на них
простой клиент

Установить

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.' )

Сервер (асинхронный) (TODO!)

 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 ])

Юникод

По умолчанию сервер и клиент принимают байты (закодированные строки), а не строки Unicode, как для адресов OSC, так и для строк OSC. Однако вы можете передать параметр encoding , чтобы ваши строки автоматически кодировались и декодировались ими, поэтому ваши обратные вызовы будут получать строки в Юникоде (unicode в python2, str в python3).

 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' )

( u добавлены здесь для ясности).

интерфейс командной строки

OSCPy предоставляет утилиту «oscli», помогающую при отладке:

oscli dump для прослушивания сообщений и их дампа
oscli send для отправки сообщений или пакетов на сервер

См. oscli -h для получения дополнительной информации.

ПОЛУЧИЛОСЬ

None значения не допускаются при сериализации.
Сокеты типа Unix еще не должны существовать, когда вы их прослушиваете().

TODO

реальная поддержка меток времени (в настоящее время поддерживается только опциональное удаление поздних пакетов, без задержки в будущем пакетов с метками времени)
поддержка дополнительных типов аргументов
реализация асинхронно-ориентированного сервера
примеры и документация