LOGO Support Package

#!/usr/bin/python
# -*- coding: utf-8 -*-

import socket
import struct
from typing import List

# Абстракция для взаимодействия {docs_engee} и LOGOS
class EngeeSim:
    def __init__(self):
        # Сокеты для связи
        self.sock_comm = None  # сокет для отправки данных (LOGOS -> {docs_engee})
        self.sock_snd = None   # сокет для приема данных ({docs_engee} -> LOGOS)
        # Параметры модели {docs_engee}
        self.engee_model_time = 0.0
        self.engee_sample_time = 0.0

    def __del__(self):
        # При удалении объекта закрываем сокеты
        if self.sock_comm:
            self.sock_comm.close()
        if self.sock_snd:
            self.sock_snd.close()

    def init_intercommunicate(self):
        # Инициализация UDP-соединения.
        self.sock_comm = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock_snd = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock_snd.settimeout(0.05)
        # Порт 6849 — передача данных из LOGOS в {docs_engee}
        self.sock_comm.connect(("127.0.0.1", 6849))
        # Порт 6848 — прием данных от {docs_engee}
        self.sock_snd.bind(("127.0.0.1", 6848))
        # Отправляем служебное сообщение для установления связи
        self.sock_comm.send(b"INIT")

    def cosimulate(self, logos_model_time: float, signals: List[float]):
        # Обмен сигналами между LOGOS и {docs_engee} на каждом шаге расчета.
        # Проверяем синхронизацию по времени модели
        if logos_model_time <= engee_sim.engee_model_time + engee_sim.engee_sample_time:
            return []

        try:
            # We accept data from {docs_engee}
            recv_data = self.sock_snd.recv(2048)
except socket.timeout:
# If there is no data, we return an empty list
            return []

        # Decompressing the data as a double (8 bytes)
        engee_data = []
        for i in range(0, len(recv_data), 8):
            chunk = recv_data[i:i+8]
            engee_data.append(struct.unpack('d', chunk)[0])

        # The first two values are the step and the {docs_engee} model time
        engee_sim.engee_sample_time = engee_data[0]
        engee_sim.engee_model_time = engee_data[1]
        engee_data = engee_data[2:]

        # Packing signals from LOGOS to send to {docs_engee}
        packed_data = b""
        for sig in signals:
            packed_data += struct.pack("d", sig)

        # Sending data to {docs_engee}
        self.sock_comm.send(packed_data)
        return engee_data

##################################################################
# Initializing data transfer between {docs_engee} and LOGOS
engee_sim = EngeeSim()
engee_sim.init_intercommunicate()
##################################################################