Описание протоколов взаимодействия с матч-контроллером Кайросс

Данные от матч-контроллера могут передаваться в двух режимах:

• по запросу (request, on-demand, pull mode)
• подписка/прослушивание (subscribe, streaming, push mode)

Доступные протоколы и форматы данных

Имя хоста и адрес матч-контроллера

В некоторых случаях вам может понадобиться адрес матч-контроллера в сети. Можно попробовать обратиться к нему по имени matchpad.local. Чтобы на Windows корректно работали адреса вида *.local (mDNS, Multicast DNS), нужно:

1. Поддержка mDNS в самой системе

   Windows 10 (с недавних версий) и Windows 11 уже имеют встроенную поддержку mDNS. Адреса hostname.local должны работать без дополнительного софта, если включены сетевые службы обнаружения.

2. Если не работает «из коробки»

   Обычно проблема в том, что на системе не установлен или отключён Bonjour (служба от Apple, используемая для mDNS).

   👉 Возможные варианты:

   • Установить Bonjour Print Services от Apple (или он автоматически ставится вместе с iTunes / iCloud).
   • Убедиться, что служба Bonjour Service (mDNSResponder.exe) запущена.

3. Настройки сети

   • mDNS работает через UDP порт 5353 на адрес мультикаста 224.0.0.251.
   • Нужно, чтобы брандмауэр Windows и антивирусы не блокировали этот порт.
   • В свойствах сетевого адаптера должно быть включено обнаружение сети.

4. Проверка

   После установки и запуска службы можно проверить доступность:

   ping matchpad.local
   

   или

   nslookup matchpad.local
   

5. Посмотреть IP-адрес на матч-контроллере

   Чтобы узнать IP-адреса, выданные контроллеру, откройте настройки подключения Wi-Fi, нажав на строку состояния. При этом не обязательно подключаться к Wi-Fi — на этой странице будут перечислены все выданные адреса.

UDP Multicast

Один из простых способов получать актуальную информацию о ходе игры — подписаться на мультикаст-рассылку по протоколу UDP.

• Адрес: 232.10.11.12
• Порт: 55000
• Формат данных: текстовый JSON в кодировке UTF-8

Общая структура пакета

Ниже приводятся типы событий и форматы данных для каждого из них.

Событие "tick" - время

• Описание: Обновление текущего времени таймера.
• Частота срабатывания:
  • Таймер остановлен: 1 раз в секунду
  • Таймер запущен: 10 раз в секунду

Событие "scoreboard" - данные на табло

• Описание: Обновление информации на табло (счёт, фолы, тайм-ауты и т.д.).
• Частота срабатывания: при изменении состояния, но не реже 1 раза каждые 30 секунд

Событие "updateLogo" - обновить логотип команды на табло

С помощью REST API можно загрузить логотип команды или фото игрока по его идентификатору используя URL http://matchpad.local/media/{mediaId}

Событие "goalScoredBy" - гол забил

• Описание: Отправляется при взятии ворот, содержит информацию об игроке, забившем гол, и при необходимости — об ассистентах. Используется для вывода сообщения на табло.
• Виды спорта: Все, где фиксируются голы (футбол, хоккей, гандбол и т. п.).
• Частота срабатывания: по факту события (не периодически).

Событие "playerStateChanged" - обновленная статистика игрока

• Описание: Передаёт обновлённые данные по отдельному игроку: очки, фолы, количество голов, штрафы и т. д.
• Виды спорта: Все командные виды, где ведётся индивидуальная статистика (например, баскетбол).
• Частота срабатывания: при каждом изменении показателей игрока (например, добавлен гол, фол, очко и т. п.).

Пример UDP-клиента на python и с++

В целях тестирования добавлен скрипт для воспроизведения UDP multicast дампа на Python udp_replay.py и дамп с данными udp_dump.jsonl. Поместите их в одну папку и запустите скрипт.

python udp_relay.py --help # выдаст опции для запуска

python udp_replay.py # воспроизведет дамп

python udp_replay.py --delay 0.5 --loop  # задержка 0.5с, зациклить

python udp_replay.py --dump my_dump.jsonl  # другой файл дампа

👉 Если не приходят пакеты, попробуйте заменить IFACE IP-адресом вашего компьютера из одной сети с матч-контроллером.

# udp_client_async.py
import asyncio
import socket
import struct
import json
import sys

MCAST_GRP = "232.10.11.12"
MCAST_PORT = 55000
BUFFER_SIZE = 65536

# Укажите здесь IP вашего локального интерфейса.
# Например: "192.168.1.42" или оставьте "0.0.0.0" (INADDR_ANY).
IFACE = "0.0.0.0"

class MulticastProtocol(asyncio.DatagramProtocol):
    def connection_made(self, transport):
        self.transport = transport
        sock = transport.get_extra_info("socket")
        print("transport ready, sock:", sock.getsockname())

    def datagram_received(self, data, addr):
        try:
            message = json.loads(data.decode("utf-8"))
            event_type = message.get("event", "unknown")
            event_data = message.get("data", message)
            print(f"📩 Новое событие: {event_type} | data={event_data}")
        except Exception as e:
            print("⚠️ Ошибка парсинга/обработки пакета:", e, "raw:", data)

    def error_received(self, exc):
        print("❌ Ошибка UDP:", exc)

    def connection_lost(self, exc):
        print("❌ Соединение закрыто", exc)

async def main():
    loop = asyncio.get_running_loop()

    # Создаем сокет и настраиваем опции до передачи в asyncio
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    # На macOS/новых Linux полезно разрешить REUSEPORT (если доступно)
    if hasattr(socket, "SO_REUSEPORT"):
        try:
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
        except OSError:
            pass

    # Привязываемся к порту на всех интерфейсах
    sock.bind(("", MCAST_PORT))

    # Подписываемся на мультикаст
    try:
        # Для macOS/Unix предпочтительнее 4s4s: group + local_if
        mreq = socket.inet_aton(MCAST_GRP) + socket.inet_aton(IFACE)
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
    except Exception as e:
        print("⚠️ Не удалось присоединиться к мультикаст-группе:", e)
        print("   Попробуй поставить IFACE = реальный IP интерфейса (например, ipconfig getifaddr en0).")
        # не выходим — иногда join не обязателен для теста

    sock.setblocking(False)

    transport, protocol = await loop.create_datagram_endpoint(
        lambda: MulticastProtocol(),
        sock=sock
    )

    print(f"🔌 Подключение к UDP Multicast {MCAST_GRP}:{MCAST_PORT} (IFACE={IFACE})...")

    try:
        # держим соединение бесконечно
        await asyncio.Event().wait()
    except asyncio.CancelledError:
        pass
    except KeyboardInterrupt:
        print("\n❌ Остановка клиента...")
    finally:
        transport.close()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        sys.exit(0)

// udp_client.cpp
// UDP Multicast клиент для Windows (Visual Studio)
// Компиляция: cl.exe udp_client.cpp /EHsc Ws2_32.lib
// или через Visual Studio: создать Console App и добавить этот файл

#include <iostream>
#include <string>
#include <winsock2.h>
#include <ws2tcpip.h>
#include <nlohmann/json.hpp> // нужно установить: vcpkg install nlohmann-json

#pragma comment(lib, "Ws2_32.lib")

using json = nlohmann::json;

const char* MCAST_GRP = "232.10.11.12";
const int MCAST_PORT = 55000;
const int BUFFER_SIZE = 65536;

class UdpMulticastClient {
private:
    SOCKET sock;
    struct sockaddr_in addr;
    char buffer[BUFFER_SIZE];

public:
    UdpMulticastClient() : sock(INVALID_SOCKET) {}

    ~UdpMulticastClient() {
        if (sock != INVALID_SOCKET) {
            closesocket(sock);
        }
        WSACleanup();
    }

    bool initialize() {
        // Инициализация Winsock
        WSADATA wsaData;
        if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) {
            std::cerr << "❌ WSAStartup failed" << std::endl;
            return false;
        }

        // Создаём UDP сокет
        sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
        if (sock == INVALID_SOCKET) {
            std::cerr << "❌ Socket creation failed: " << WSAGetLastError() << std::endl;
            return false;
        }

        // Разрешаем повторное использование адреса
        BOOL reuse = TRUE;
        if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, 
                      (char*)&reuse, sizeof(reuse)) < 0) {
            std::cerr << "⚠️ SO_REUSEADDR failed" << std::endl;
        }

        // Привязываемся к порту
        memset(&addr, 0, sizeof(addr));
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = htonl(INADDR_ANY);
        addr.sin_port = htons(MCAST_PORT);

        if (bind(sock, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
            std::cerr << "❌ Bind failed: " << WSAGetLastError() << std::endl;
            return false;
        }

        // Подписываемся на мультикаст группу
        struct ip_mreq mreq;
        InetPtonA(AF_INET, MCAST_GRP, &mreq.imr_multiaddr);
        mreq.imr_interface.s_addr = htonl(INADDR_ANY);

        if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, 
                      (char*)&mreq, sizeof(mreq)) < 0) {
            std::cerr << "❌ IP_ADD_MEMBERSHIP failed: " << WSAGetLastError() << std::endl;
            return false;
        }

        std::cout << "🔌 Подключение к UDP Multicast " 
                  << MCAST_GRP << ":" << MCAST_PORT << std::endl;
        return true;
    }

    void run() {
        std::cout << "📡 Ожидание событий... (Ctrl+C для выхода)" << std::endl;
        std::cout << std::string(50, '-') << std::endl;

        while (true) {
            memset(buffer, 0, BUFFER_SIZE);
            
            struct sockaddr_in sender;
            int senderLen = sizeof(sender);
            
            int bytesReceived = recvfrom(sock, buffer, BUFFER_SIZE - 1, 0,
                                        (struct sockaddr*)&sender, &senderLen);

            if (bytesReceived < 0) {
                std::cerr << "⚠️ recvfrom error: " << WSAGetLastError() << std::endl;
                continue;
            }

            if (bytesReceived == 0) {
                continue;
            }

            // Парсим JSON
            try {
                std::string jsonStr(buffer, bytesReceived);
                json message = json::parse(jsonStr);

                std::string eventType = message.value("event", "unknown");
                json eventData = message.value("data", json::object());

                std::cout << "📩 Новое событие: " << eventType 
                         << " | data=" << eventData.dump() << std::endl;

            } catch (const json::parse_error& e) {
                std::cerr << "⚠️ Ошибка парсинга JSON: " << e.what() << std::endl;
                std::cerr << "   raw: " << std::string(buffer, bytesReceived) << std::endl;
            } catch (const std::exception& e) {
                std::cerr << "⚠️ Ошибка: " << e.what() << std::endl;
            }
        }
    }
};

int main() {
    // Устанавливаем UTF-8 для консоли (для эмодзи)
    SetConsoleOutputCP(CP_UTF8);

    UdpMulticastClient client;

    if (!client.initialize()) {
        std::cerr << "❌ Не удалось инициализировать клиент" << std::endl;
        return 1;
    }

    try {
        client.run();
    } catch (const std::exception& e) {
        std::cerr << "❌ Ошибка: " << e.what() << std::endl;
        return 1;
    }

    return 0;
}

#!/usr/bin/env python3
"""
Скрипт для воспроизведения UDP multicast дампа (для тестирования).
Читает udp_dump.jsonl и отправляет события в мультикаст-группу.

python udp_relay.py --help

Воспроизведите дамп:
python udp_replay.py

С опциями:
python udp_replay.py --delay 0.5 --loop  # задержка 0.5с, зациклить
python udp_replay.py --dump my_dump.jsonl  # другой файл дампа
"""
import asyncio
import socket
import json
import sys
import argparse
from pathlib import Path

MCAST_GRP = "232.10.11.12"
MCAST_PORT = 55000
TTL = 2  # Time-to-live для мультикаста


async def replay_dump(
    dump_file: str,
    delay: float = 0.1,
    loop_replay: bool = False
):
    """
    Воспроизводит дамп UDP событий.
    
    Args:
        dump_file: Путь к файлу дампа (JSON Lines)
        delay: Задержка между отправкой пакетов (в секундах)
        loop_replay: Зациклить воспроизведение
    """
    if not Path(dump_file).exists():
        print(f"❌ Файл {dump_file} не найден!")
        sys.exit(1)

    # Создаём UDP сокет для отправки
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, TTL)

    print(f"🎬 Воспроизведение дампа: {dump_file}")
    print(f"📡 Отправка в {MCAST_GRP}:{MCAST_PORT}")
    print(f"⏱️  Задержка между пакетами: {delay}s")
    if loop_replay:
        print("🔁 Режим: зацикленное воспроизведение")
    print("-" * 50)

    try:
        while True:
            with open(dump_file, "r", encoding="utf-8") as f:
                count = 0
                for line in f:
                    line = line.strip()
                    if not line:
                        continue

                    try:
                        entry = json.loads(line)
                        event_type = entry.get("event", "unknown")
                        event_data = entry.get("data", {})

                        # Формируем исходное сообщение
                        message = {
                            "event": event_type,
                            "data": event_data,
                        }

                        # Отправляем в мультикаст
                        payload = json.dumps(message, ensure_ascii=False)
                        sock.sendto(
                            payload.encode("utf-8"),
                            (MCAST_GRP, MCAST_PORT)
                        )

                        count += 1
                        print(
                            f"📤 [{count}] Отправлено: "
                            f"{event_type} | {event_data}"
                        )

                        # Задержка между пакетами
                        await asyncio.sleep(delay)

                    except json.JSONDecodeError as e:
                        print(f"⚠️ Ошибка парсинга строки: {e}")
                        continue

            print(f"\n✅ Воспроизведено {count} событий")

            if not loop_replay:
                break

            print("\n🔁 Повтор воспроизведения...\n")
            await asyncio.sleep(1)

    except KeyboardInterrupt:
        print("\n❌ Остановка воспроизведения...")
    finally:
        sock.close()


def main():
    parser = argparse.ArgumentParser(
        description="Воспроизведение UDP multicast дампа"
    )
    parser.add_argument(
        "--dump",
        default="udp_dump.jsonl",
        help="Путь к файлу дампа (по умолчанию: udp_dump.jsonl)",
    )
    parser.add_argument(
        "--delay",
        type=float,
        default=0.1,
        help="Задержка между пакетами в секундах (по умолчанию: 0.1)",
    )
    parser.add_argument(
        "--loop",
        action="store_true",
        help="Зациклить воспроизведение",
    )

    args = parser.parse_args()

    try:
        asyncio.run(
            replay_dump(
                dump_file=args.dump,
                delay=args.delay,
                loop_replay=args.loop
            )
        )
    except KeyboardInterrupt:
        sys.exit(0)


if __name__ == "__main__":
    main()

Server-Sent Events (SSE)

Ещё один способ получать актуальную информацию — это подписка на события по протоколу SSE по адресу:
http://matchpad.local/api/v1/events

Вместо имени сервера matchpad.local можно использовать IP-адрес матч-контроллера.
Подписку можно проверить прямо в браузере, просто открыв указанную ссылку.

Протокол SSE идеально подходит для браузерных клиентов — например, для вывода информации о матче на HTML-странице. Именно этот механизм используется встроенным веб-сервером для отображения данных на видеотабло и SmartTV.

Пример открытой ссылки в браузере

event: scoreboard
data: {"sportSlug":"basketball","home":{"score":0,"fouls":0,"name":"ЦСКА","town":"Москва","teamId":"n7hitgakhv1kxhvy89ilvhdp","lineup":["2","8","9","11","41"],"timeouts":0},"away":{"score":0,"fouls":0,"name":"УНИКС","town":"Казань","teamId":"lxf3lhv3vc6byoj7q5yxqoqb","lineup":["2","4","5","12","15"],"timeouts":0},"period":"1","tournament":"Единая лига ВТБ","competitionId":"abs6b7s802kzra2gy6jed4km"}

event: updateLogo
data: {"team":"home","logo":{"id":"af5t865jqxyoebsot7jgil9t","width":147,"height":216},"primaryColor":"#d61a20","secondaryColor":"#003274","textColor":"#ffffff"}

event: updateLogo
data: {"team":"away","logo":{"id":"y2dny9a6jmj4zar666a0gtij","width":1820,"height":1820},"primaryColor":"#2faf6b","secondaryColor":"#d36d14","textColor":"#ffffff"}

event: tick
data: {"kind":"primary","view":"09:57.3","isRunning":false,"elapsed":2727,"timestamp":1759494388845,"name":"quarter","nibusId":1}

👉 Формат передаваемых данных полностью совпадает с форматом, описанным в разделе UDP Multicast.

Пример SSE-клиента на python

# sse_client_async.py
import asyncio
import aiohttp
import json

SSE_URL = "http://matchpad.local/api/v1/events"

async def sse_client(url):
    while True:
        try:
            print("🔌 Подключение к SSE...")
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as resp:
                    if resp.status != 200:
                        raise Exception(f"Ошибка подключения: {resp.status}")

                    event_type = "message"
                    data_buffer = []

                    async for line_bytes in resp.content:
                        line = line_bytes.decode("utf-8").strip()
                        if line.startswith("event:"):
                            event_type = line[6:].strip()
                        elif line.startswith("data:"):
                            data_buffer.append(line[5:].strip())
                        elif line == "":
                            # пустая строка — конец события
                            if data_buffer:
                                raw_data = "\n".join(data_buffer)
                                try:
                                    data = json.loads(raw_data)
                                except json.JSONDecodeError:
                                    data = raw_data
                                print(f"📩 Новое событие: {event_type} | data={data}")
                            # сброс
                            event_type = "message"
                            data_buffer = []

        except Exception as e:
            print("❌ Ошибка SSE:", e)

        await asyncio.sleep(5)
        print("♻️ Переподключение...")

async def main():
    await sse_client(SSE_URL)

if __name__ == "__main__":
    asyncio.run(main())

REST API

Полная справка в формате openapi/swagger по REST API доступна при подключении к матч-контроллеру с помощью браузера по адресу http://matchpad.local/api-docs. Вместо matchpad.local можно указать IP-адрес матч-контроллера.

Протокол NiBUS

NiBUS — проприетарный промышленный протокол фирмы Ната-Инфо. Скачать описание протокола NiBUS.pdf

👉 Полное описание протокола выходит за рамки данной публикации. Здесь рассмотрены только ключевые особенности, важные для интеграции матч-контроллера.

Особенностью протокола NiBUS является использование специальных двухбайтовых маркеров (служебных последовательностей), которые передаются без паузы между байтами и используются для управления доступом к каналу RS-485. Некоторые "умные" контроллеры и преобразователи, например устройства MOXA, могут некорректно обрабатывать такие маркеры из-за особенностей внутренней буферизации и автоматического управления направлением передачи, что приводит к искажению данных. Для стабильной работы рекомендуется использовать трансиверы, разработанные Ната-Инфо — Siolynx2 и C22. Скорость подключения к Siolynx2 — 115200 бод, для остальных устройств — 57600 бод.

Каждый кадр данных начинается с преамбулы 0x7e, затем идут адреса получателя и отправителя, сервисные поля, но первое, что имеет значение для анализа — это размер N полученного кадра, расположенный в байте со смещением 14 байт от начала кадра. После определения размера можно вычислить окончание кадра (17 + N), рассчитать контрольную сумму и сверить её с полученными данными.

Формат кадра

👉 ВАЖНО. Длинна данных в одном кадре ограниченна 238 байтами, и там могут располагаться сразу несколько информационных сообщений (NMS). Смотрите длину данных nms, чтобы определить границу текущеего сообщения, сразу за ним может находится следующее NMS-сообщение. Такое бывает, например, когда с основным временем передается время 24-секундника.

Стандартные типы данных

Контрольная сумма

Циклическая контрольная сумма подсчитывается (не включает преамбулу) по стандарту CRC16-ССITT. Делитель равен 1 0001 0000 0010 0001 = 0x1021. Начальное значение 0x0000. Хранится в формате big-endian.

Детализация информационных сообщений

👉 Вся текстовая информация в сообщениях представлена кодировкой Windows-1251, бинарные данные используют порядок байт (endianness) little-endian, формат BCD означает binary-coded decimal (например 38₁₀ хранится и передается как байт 0x38) .

Время

ID: 0x05
TYPE: VT_UI1 | VT_VECTOR (0x91)
DATA:

+0 байт - флаги:

+1 байт - идентификатор таймера:

для баскетбола:

для волейбола:

для гандбола:

для мини-футбола (футзал):

для хоккея:

+3 байта - минуты (BCD)

+4 байта - секунды (BCD)

+5 байт - доли секунд (BCD)

Счет команды хозяев

ID: 0x06
TYPE: VT_UI2 (0x12)
DATA: +0..1 байт - значение счета

Счет команды гостей

ID: 0x07
TYPE: VT_UI2 (0x12)
DATA: +0..1 байт - значение счета

Период

ID: 0x08
TYPE: VT_UI1 (0x11)
DATA: +0 байт - значение периода

Фолы команды хозяев

ID: 0x09
TYPE: VT_UI1 (0x11)
DATA: +0 байт - значение

Фолы команды гостей

ID: 0x0a
TYPE: VT_UI1 (0x11)
DATA: +0 байт - значение

Перерывы команды хозяев

ID: 0x0e
TYPE: VT_UI1 (0x11)
DATA: +0 байт - значение

Перерывы команды гостей

ID: 0x0f
TYPE: VT_UI1 (0x11)
DATA: +0 байт - значение

Количество игроков в команде

ID: 0x10
TYPE: VT_UI1 | VT_VECTOR (0x91)
DATA:

+0 байт - команда:

+1..2 байт - количество игроков

Информация о игроке

ID: 0x11
TYPE: VT_UI1 | VT_VECTOR (0x91)
DATA:

+0 байт - команда:

+1 байт - зарезервировано

+2 байт - индекс игрока (начинается с 0)

+3 байт - номер игрока (BCD)

+4 байт - роль игрока:

+5 байт - фамилия (не более 30 знаков, терминированная 0), страна (не более 20 знаков), допинформация (терминированная 0)

Пример: Петров А.\0РОССИЯ\0\0

Статистика по игроку

ID: 0x12
TYPE: VT_UI1 | VT_VECTOR (0x91)
DATA:

+0 байт - команда:

+1 байт - номер игрока (BCD)

+2 байт - индекс игрока (начинается с 0)

+3 байт - суммарное число очков (BCD)

+4 байт - cуммарное число фолов (BCD)

+5 байт - статус игрока:

Опционально:

+6 байт - количество 1-очковых бросков игрока (BCD)

+7 байт - количество 2-очковых бросков игрока (BCD)

+8 байт - количество 3-очковых бросков игрока (BCD)

Название команды хозяев

ID: 0x13
TYPE: VT_LPSTR (0x1e)
DATA: +0..M-1 байт - название команды, терминированное 0

Название команды гостей

ID: 0x14
TYPE: VT_LPSTR (0x1e)
DATA: +0..M-1 байт - название команды, терминированное 0

Город/страна команды хозяев

ID: 0x15
TYPE: VT_LPSTR (0x1e)
DATA: +0..M-1 байт - город/страна, терминированное 0

Город/страна команды гостей

ID: 0x16
TYPE: VT_LPSTR (0x1e)
DATA: +0..M-1 байт - город/страна, терминированное 0

Название турнира

ID: 0x17
TYPE: VT_LPSTR (0x1e)
DATA: +0..M-1 байт - название турнира, терминированное 0

Указатель владения мячом

ID: 0x18
TYPE: VT_UI1 (0x11)
DATA: +0 байт - значение

Вывод информационного сообщения

ID: 0x19
TYPE: VT_UI1 | VT_VECTOR (0x91)
DATA:

+0 байт - тип сообщения:

+1..M-1 байт - фаимлия игрока (не более 30 знаков, терминированная 0), номер игрока (не более 2 знаков), название команды (не более 20 знаков), допинформация (терминированная 0)

Пример: Петров А.\012\0ЦСКА\0\0

Переключение провайдера игры (смена вида спорта)

ID: 0x1b
TYPE: VT_UI2 (0x12)
DATA: +0..1 байт - вид спорта

Статистика по удалениям (Гандбол/Мини-футбол/Хоккей)

ID: 0x1c
TYPE: VT_UI1 | VT_VECTOR (0x91)
DATA:

+0 байт - количество строк (всего позиций 2*N, для 2х команд)

Строка 1 (хозяева)

+1 байт - флаги:

+2 байт - номер игрока (BCD)

+3 байт - минуты (BCD)

+4 байт - секунды (BCD)

+5 байт - зарезервировано

Всего N строк для хозяев, затем N строк для гостей: 5*(2*N) байт.

Статистика "Волейбол"

ID: 0x1d
TYPE: VT_UI1 | VT_VECTOR (0x91)
DATA:

+0 байт - счет команды хозяев в 1-партии

+1 байт - счет команды хозяев в 2-партии

+2 байт - счет команды хозяев в 3-партии

+3 байт - счет команды хозяев в 4-партии

+4 байт - счет команды хозяев в 5-партии

+5 байт - счет команды гостей в 1-партии

+6 байт - счет команды гостей в 2-партии

+7 байт - счет команды гостей в 3-партии

+8 байт - счет команды гостей в 4-партии

+9 байт - счет команды гостей в 5-партии

Количество замен в команде (Волейбол)

ID: 0x1e
TYPE: VT_UI1 | VT_VECTOR (0x91)
DATA:

+0 байт - команда:

+1 байт - количество замен

Пример NiBUS-парсера на Python

# nibus_parser.py
import serial
import struct
import sys

# Настройки порта
SERIAL_PORT = "COM3"  # Измените на ваш порт (например, "/dev/ttyUSB0" на Linux/macOS)
BAUD_RATE = 57600  # Укажите 115200, если используете Siolynx2

# Константы протокола
PREAMBLE = 0x7E
FRAME_TYPE = 0x01
SERVICE_TYPE = 0x18

# Типы данных NiBUS
VT_TYPES = {
    0x0B: ("VT_BOOL", 1),
    0x10: ("VT_I1", 1),
    0x02: ("VT_I2", 2),
    0x03: ("VT_I4", 4),
    0x14: ("VT_I8", 8),
    0x11: ("VT_UI1", 1),
    0x12: ("VT_UI2", 2),
    0x13: ("VT_UI4", 4),
    0x15: ("VT_UI8", 8),
    0x04: ("VT_R4", 4),
    0x05: ("VT_R8", 8),
    0x1E: ("VT_LPSTR", None),  # строка переменной длины
    0x07: ("VT_DATE", 10),
    0x91: ("VT_UI1|VT_VECTOR", None),  # вектор байтов
}


def calc_crc16(data):
    """Вычисление CRC16-CCITT (полином 0x1021, начальное значение 0x0000)"""
    crc = 0x0000
    for byte in data:
        crc ^= byte << 8
        for _ in range(8):
            if crc & 0x8000:
                crc = (crc << 1) ^ 0x1021
            else:
                crc = crc << 1
            crc &= 0xFFFF
    return crc


def parse_data(data_type, data_bytes):
    """Парсинг данных в зависимости от типа"""
    if data_type not in VT_TYPES:
        return f"Unknown type 0x{data_type:02x}", data_bytes.hex()
    
    type_name, size = VT_TYPES[data_type]
    
    if data_type == 0x1E:  # VT_LPSTR - строка
        try:
            # Ищем терминирующий ноль
            end = data_bytes.find(b'\x00')
            if end == -1:
                return type_name, data_bytes.decode('windows-1251', errors='replace')
            return type_name, data_bytes[:end].decode('windows-1251', errors='replace')
        except:
            return type_name, data_bytes.hex()
    
    elif data_type == 0x91:  # VT_UI1|VT_VECTOR - массив байтов
        return type_name, ' '.join(f'{b:02x}' for b in data_bytes)
    
    elif data_type == 0x0B:  # VT_BOOL
        return type_name, bool(data_bytes[0])
    
    elif data_type in [0x11, 0x10]:  # VT_UI1, VT_I1
        fmt = 'B' if data_type == 0x11 else 'b'
        return type_name, struct.unpack(fmt, data_bytes[:1])[0]
    
    elif data_type in [0x12, 0x02]:  # VT_UI2, VT_I2
        fmt = '<H' if data_type == 0x12 else '<h'
        return type_name, struct.unpack(fmt, data_bytes[:2])[0]
    
    elif data_type in [0x13, 0x03]:  # VT_UI4, VT_I4
        fmt = '<I' if data_type == 0x13 else '<i'
        return type_name, struct.unpack(fmt, data_bytes[:4])[0]
    
    elif data_type in [0x15, 0x14]:  # VT_UI8, VT_I8
        fmt = '<Q' if data_type == 0x15 else '<q'
        return type_name, struct.unpack(fmt, data_bytes[:8])[0]
    
    elif data_type == 0x04:  # VT_R4 - float
        return type_name, struct.unpack('<f', data_bytes[:4])[0]
    
    elif data_type == 0x05:  # VT_R8 - double
        return type_name, struct.unpack('<d', data_bytes[:8])[0]
    
    elif data_type == 0x07:  # VT_DATE - BCD дата/время
        return type_name, ' '.join(f'{b:02x}' for b in data_bytes[:10])
    
    return type_name, data_bytes.hex()

def parse_nibus_frame(frame):
    """Парсинг кадра NiBUS"""
    if len(frame) < 20:
        print("⚠️ Слишком короткий кадр")
        return False
    
    # Проверка преамбулы
    if frame[0] != PREAMBLE:
        print(f"⚠️ Неверная преамбула: 0x{frame[0]:02x}")
        return False
    
    # Размер кадра
    frame_size = frame[14]
    
    if len(frame) < 15 + frame_size + 2:
        print(f"⚠️ Неполный кадр (ожидается {15 + frame_size + 2}, получено {len(frame)})")
        return False
    
    # Тип кадра и сервиса
    frame_type = frame[15]
    service_type = frame[16]
    
    if frame_type != FRAME_TYPE:
        return False  # Пропускаем без вывода
    
    if service_type != SERVICE_TYPE:
        return False  # Пропускаем без вывода
    
    # ID, длина данных, тип данных
    msg_id = frame[17]
    data_len = frame[18]
    data_type = frame[19]
    
    # Данные
    data_start = 20
    data_end = data_start + data_len
    data_bytes = frame[data_start:data_end]
    
    # CRC
    crc_start = 15 + frame_size
    crc_received = struct.unpack('<H', frame[crc_start:crc_start + 2])[0]
    
    # Проверка CRC (без преамбулы)
    crc_calculated = int.from_bytes(calc_crc16(frame[1:crc_start]).to_bytes(2, 'big'), 'little')
    
    if crc_calculated != crc_received:
        print(f"⚠️ Ошибка CRC: вычислено 0x{crc_calculated:04x}, получено 0x{crc_received:04x}")
        return False
    
    # Парсинг данных
    type_name, parsed_data = parse_data(data_type, data_bytes)
    
    # Вывод
    print(f"📦 ID: 0x{msg_id:02x} | TYPE: {type_name} (0x{data_type:02x}) | DATA: {parsed_data}")
    return True

def main():
    print(f"🔌 Подключение к {SERIAL_PORT} на скорости {BAUD_RATE} бод...")
    
    try:
        ser = serial.Serial(
            port=SERIAL_PORT,
            baudrate=BAUD_RATE,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            timeout=1
        )
        
        print("✅ Порт открыт. Ожидание данных...")
        
        buffer = bytearray()
        
        while True:
            # Читаем данные
            if ser.in_waiting:
                chunk = ser.read(ser.in_waiting)
                buffer.extend(chunk)
            
            # Ищем преамбулу
            while len(buffer) >= 20:
                preamble_pos = buffer.find(PREAMBLE)
                
                if preamble_pos == -1:
                    # Преамбулы нет, очищаем буфер
                    buffer.clear()
                    break
                
                # Удаляем мусор до преамбулы
                if preamble_pos > 0:
                    buffer = buffer[preamble_pos:]
                
                # Проверяем, достаточно ли данных для определения размера
                if len(buffer) < 15:
                    break
                
                frame_size = buffer[14]
                total_size = 15 + frame_size + 2  # +2 для CRC
                
                # Ждём полного кадра
                if len(buffer) < total_size:
                    break
                
                # Извлекаем кадр
                frame = buffer[:total_size]
                buffer = buffer[total_size:]
                
                # Парсим кадр
                parse_nibus_frame(frame)
    
    except serial.SerialException as e:
        print(f"❌ Ошибка порта: {e}")
        sys.exit(1)
    except KeyboardInterrupt:
        print("\n❌ Остановка...")
        sys.exit(0)
    finally:
        if 'ser' in locals() and ser.is_open:
            ser.close()
            print("🔌 Порт закрыт")

if __name__ == "__main__":
    main()

COLOSSEO Universal Protocol

Этот протокол используется, если вы подключаете табло или часы производства COLOSSEO.

📄 Скачать описание протокола:
COLOSSEO Universal Protocol.pdf

⚠️ Обратите внимание: нельзя одновременно использовать оба протокола — NiBUS и COLOSSEO.
Необходимо выбрать только один.