How to use the oscpy.parser.format_message function in oscpy

To help you get started, we’ve selected a few oscpy examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github kivy / oscpy / tests / test_parser.py View on Github external
def tests_format_message():
    for message in message_1, message_2:
        source, msg, result = message
        msg = struct.pack('>%iB' % len(msg), *msg)
        assert format_message(*source)[0] == msg
github kivy / oscpy / tests / test_parser.py View on Github external
def test_read_message_wrong_address():
    msg, stat = format_message(b'test', [])
    with raises(ValueError, match="doesn't start with a '/'") as e:
        address, tags, values, size = read_message(msg)
github kivy / oscpy / tests / performances.py View on Github external
[1.2345, 1.2345, 10000000000.],
    list(range(500)),
]

print("#" * 80)
print("format/parse test")

for i, pattern in enumerate(patterns):
    print("*" * 100)
    print(f"pattern: {i}")
    n = 0
    timeout = time() + DURATION

    while time() < timeout:
        n += 1
        p, s = format_message(b'/address', pattern)

    size = len(p) / 1000
    print(
        f"formated message {n} times ({n / DURATION}/s) "
        f"({n * size / DURATION:.2f}MB/s)"
    )

    n = 0
    timeout = time() + DURATION

    while time() < timeout:
        n += 1
        read_message(p)

    print(
        f"parsed message {n} times ({n / DURATION}/s) "
github kivy / oscpy / tests / performances.py View on Github external
osc.bind(b'/count', count, sock=sock)
    for i, pattern in enumerate(patterns):
        for safer in (False, True):
            timeout = time() + DURATION
            sent = 0
            received = 0

            while time() < timeout:
                send_message(
                    b'/count', pattern, address, port, sock=SOCK, safer=safer
                )
                sent += 1
            sleep(10e-9)

            size = len(format_message(b'/count', pattern)[0]) / 1000.

            print(
                f"{i}: safe: {safer}\t",
                f"sent:\t{sent}\t({sent / DURATION}/s)\t({sent * size / DURATION:.2f}MB/s)\t"                  # noqa
                f"received:\t{received}\t({received / DURATION}/s)\t({received * size / DURATION:.2f}MB/s)\t"  # noqa
                f"loss {((sent - received) / sent) * 100}%"
            )

    if family == 'unix':
        os.unlink(address)
    osc.stop_all()
github kivy / oscpy / tests / test_parser.py View on Github external
def test_format_wrong_types():
    with raises(TypeError):
        format_message(b'/test', values=[u'test'])
github kivy / oscpy / tests / test_parser.py View on Github external
def test_format_encoding():
    s = u'éééààà'
    with raises(TypeError):
        read_message(format_message('/test', [s])[0])

    assert read_message(format_message('/test', [s], encoding='utf8')[0])[2][0] == s.encode('utf8')  # noqa
    assert read_message(format_message('/test', [s], encoding='utf8')[0], encoding='utf8')[2][0] == s  # noqa

    with raises(UnicodeEncodeError):
        format_message('/test', [s], encoding='ascii')  # noqa

    with raises(UnicodeDecodeError):
        read_message(format_message('/test', [s], encoding='utf8')[0], encoding='ascii')  # noqa

    assert read_message(
        format_message('/test', [s], encoding='utf8')[0],
        encoding='ascii', encoding_errors='ignore'
    )[2][0] == ''

    assert read_message(
github kivy / oscpy / tests / test_parser.py View on Github external
def tests_format_message_null_terminated_address():
    for message in message_1, message_2:
        source, msg, result = message
        source = source[0] + b'\0', source[1]
        msg = struct.pack('>%iB' % len(msg), *msg)
        assert format_message(*source)[0] == msg
github kivy / oscpy / tests / test_parser.py View on Github external
def test_format_encoding():
    s = u'éééààà'
    with raises(TypeError):
        read_message(format_message('/test', [s])[0])

    assert read_message(format_message('/test', [s], encoding='utf8')[0])[2][0] == s.encode('utf8')  # noqa
    assert read_message(format_message('/test', [s], encoding='utf8')[0], encoding='utf8')[2][0] == s  # noqa

    with raises(UnicodeEncodeError):
        format_message('/test', [s], encoding='ascii')  # noqa

    with raises(UnicodeDecodeError):
        read_message(format_message('/test', [s], encoding='utf8')[0], encoding='ascii')  # noqa

    assert read_message(
        format_message('/test', [s], encoding='utf8')[0],
        encoding='ascii', encoding_errors='ignore'
    )[2][0] == ''

    assert read_message(
        format_message('/test', [s], encoding='utf8')[0],
        encoding='ascii', encoding_errors='replace'
    )[2][0] == u'������������'
github kivy / oscpy / oscpy / client.py View on Github external
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        send_message(b'/test', [], '192.168.0.1', 8000, sock=sock, safer=True)

        # unix sockets work on linux and osx, and over unix platforms,
        # but not windows
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        send_message(b'/some/address', [1, 2, 3], b'/tmp/sock')

    """
    if platform != 'win32' and sock.family == socket.AF_UNIX:
        address = ip_address
    else:
        address = (ip_address, port)

    message, stats = format_message(
        osc_address, values, encoding=encoding,
        encoding_errors=encoding_errors
    )

    sock.sendto(message, address)
    if safer:
        sleep(10e-9)

    return stats