testReadArchiveAndSendToQueue.py - master - afedorov1971/apkdk-tests

testReadArchiveAndSendToQueue.py
65 строк3 KB
Alexander Fedorov
добавлен аргумент для чтения и посылки данных без задержек
20 апр 2025, 18:55
20 апр 2025, 18:5533f2fe2
import argparse
import time
import pika
from imsat import archive
from imsat import utils

parser = argparse.ArgumentParser(description='Чтение файла архива и посылка пакетов в виде датаграмм')
parser.add_argument('--archive', help='Имя файла архива', required='true')
parser.add_argument('--server', help='RabbitMQ сервер', default='localhost')
parser.add_argument('--port', help='Номер порта RabbitMQ сервера', default=5672)
parser.add_argument('--user', help='Имя пользователя для подключения', default='imsat')
parser.add_argument('--password', help='Пароль для подключения', default='imsat')
parser.add_argument('--nodelay', help='Выполнять или нет задержку перед посылкой данных', default=False)
args = parser.parse_args()

server_address = (args.server, args.port)

credentials = pika.PlainCredentials(args.user, args.password)
connection = pika.BlockingConnection(pika.ConnectionParameters(host=args.server,
                                                               port=args.port,
                                                               credentials=credentials))
channel = connection.channel()

channel.exchange_declare(exchange='imsat-messages', exchange_type='fanout')

print('Пакеты данных посылаются на сервер {0}, порт {1}'.format(args.server, args.port))

package_counter = 1
prevWriteTime = None


def on_read_package(data_package: archive.DatagramPackage, write_time: int):
    global package_counter
    global prevWriteTime
    global channel

    if prevWriteTime is None:
        print('Время {0}:'.format(utils.time_extended_to_string(write_time)))
    else:
        if not args.nodelay:
            diff_in_secs = round((write_time - prevWriteTime) / 1000000, 3)
            if diff_in_secs > 0.1:
                print('Waiting for {0} secs...'.format(diff_in_secs))
                time.sleep(diff_in_secs)

    print('Пакет: {0}, Инфо={1}'.format(package_counter, data_package.to_string()))

    # Посылаем пакет в очередь сообщений
    channel.basic_publish(exchange='imsat-messages', routing_key='', body=data_package.to_bytes())

    prevWriteTime = write_time

    package_counter += 1


# открываем архив и читаем записанные пакеты
try:
    archive.read_archive(args.archive, on_read_package)

except Exception as ex:
    print("Ошибка чтения архива {0}".format(ex))

print('Всего послано {0} пакетов данных.'.format(package_counter - 1))

connection.close()