import neopixel import array import os import socket import RPi.GPIO as gpio import math import subprocess import sys import time from threading import Thread from multiprocessing import Process from PyQt5.QtCore import * from PyQt5.QtWidgets import * from PyQt5.QtGui import * import board import json global autotest_client, flage_remove_cV, counter_client_play_video, counter_setting_request, \ number_LED, DIR, STEP, ENA, DEGSTEP, CW, CCW, MICROSTEP, FACTOR, MAX_SPEED_DELAY_small_dist, \ START_DELAY_small_dist, FIN_DELAY_small_dist, MAX_SPEED_DELAY_long_dist, START_DELAY_long_dist, \ FIN_DELAY_long_dist, HALL, board_PIXEL, brightness, line_script_Wait, R_wait, G_wait, B_wait, \ R2_wait, G2_wait, B2_wait, initial_Xaxis, initial_Yaxis, size_Xaxis, size_Yaxis, pixel, connected_socket global base_path, navigation_directory, event_directory, weather_video_path, video_path BUFFER_SIZE = 16384 message_client_dict = "" # number_LED = 26 # Количество пикселей светодиодной ленты # screenplay = 0 # pixel = None # DIR = 20 # Номер пина, который отвечает за направление вращения # STEP = 21 # Номер пина, который отвечает за шаг вращения # ENA = 22 # Номер пина, который отвечает за включение и отключение драйвера # CW = 0 # Вращение по часовой стрелке # CCW = 1 # Вращение против часовой стрелки # SPR = 200 # Шагов за оборот (*250*360/1.8) # DEGSTEP = 1.8 # Значение угла одного шага двигателя # SYSCOF = 50 * 8 # SYSCOF = 50 * 8 # Коэфициент системы. Количество оборотов двигателя для 360 градусов системы # MAX_SPEED_DELAY_small_dist = 0.000020 # Настройка максимальной скорости. Больще значение, тем меньше скорость # MAX_SPEED_DELAY_long_dist = 0.000015 # Настройка максимальной скорости. Больще значение, тем меньше скорость # DELAYSTEPPER = 0.00010 # START_DELAY_small_dist = 0.00002 # Настройка начальной скорости маленькой дистанции # FIN_DELAY_small_dist = 0.0010 # Настройка конечной скорости маленькой дистанции # START_DELAY_long_dist = 0.000014 # Настройка начальной скорости большой дистанции # FIN_DELAY_long_dist = 0.00003 # Настройка конечной скорости большой дистанции # HALL = 2 # Номер пина, к которому подключен датчик Холла # flage_remove_cV = False # autotest_client = False # necessarystep = 0 # Здесь определяются функции и классы для обработки команд от клиента setting_client = [] # Массив с командами от клиента. Элемент массива это стока с набором значений. filepath_settinh_client = [] setting = None # Команда от клиента setting_priority = None # Приоритетная команда от клиента objectName = [] # Названия объектов server_socket = None connected_socket = None thread_check_message_client = None filepath = None brightness = 0 counter_client_play_video = None message_play_video = None def get_base_path(): if getattr(sys, 'frozen', False): # Если приложение собрано в исполняемый файл base_path = os.path.dirname(os.path.dirname(os.path.dirname(sys.executable))) else: # Если приложение запущено как скрипт base_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) return base_path # Получаем абсолютный путь к директории, где находится скрипт base_path = get_base_path() search_directory = os.path.join(base_path, "data", "Videos") # Формируем путь к папке createdNavigation внутри директории VDNH navigation_directory = os.path.join(base_path, 'data', 'Videos', 'VDNH', 'createdNavigation') # Формируем путь к папке createdEvent внутри директории VDNH event_directory = os.path.join(base_path, 'data', 'Videos', 'VDNH', 'createdEvent') # Формируем путь к файлу pogoda.mp4 внутри папки createdVideos weather_video_path = os.path.join(base_path, 'data', 'Videos', 'VDNH', 'createdVideos', 'pogoda.mp4') print(f'Базовый путь: {base_path}') print(f'Путь к папке с видео: {search_directory}') def new_socket(attempt=0): global server_socket, connected_socket def TCP_IP_PORT_BUFFER(): global TCP_IP, TCP_PORT, BUFFER_SIZE try: IP_name = socket.gethostname() print("Host name:", IP_name) TCP_IP = (([ip for ip in socket.gethostbyname_ex(socket.gethostname())[2] if not ip.startswith("127.")] or [ [(s.connect(("8.8.8.8", 53)), s.getsockname()[0], s.close()) for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]]) + ["no IP found"])[0] print("TCP IP:", TCP_IP) TCP_PORT = 5000 except OSError: time.sleep(3) gpio.cleanup() TCP_IP_PORT_BUFFER() try: TCP_IP_PORT_BUFFER() if server_socket: server_socket.close() # Закрыть существующее соединение if connected_socket: connected_socket.close() server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind((TCP_IP, TCP_PORT)) print("Server socket successfully created and bound.") server_socket.listen(1) print("Server socket is now listening for connections.") connected_socket = server_socket return server_socket except OSError: time.sleep(1) if server_socket: server_socket.close() if connected_socket: connected_socket.close() return new_socket() def onClient(message, client_socket=connected_socket): # Отправление сообщения клиенту global connected_socket from led_controller import CLEAR if client_socket == None: client_socket = connected_socket try: connected_socket.send((message + '\n').encode()) except Exception as e: print('Ошибка при работе с клиентом:', e) # pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() print("close onClient") if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) # except ConnectionResetError: # except OSError: def setting_server(client_socket=connected_socket): global connected_socket, pixel, counter_setting_request, DEGSTEP, counter_setting_request, \ autotest_client, flage_remove_cV, counter_client_play_video, counter_setting_request, \ DIR, STEP, ENA, DEGSTEP, CW, CCW, MICROSTEP, FACTOR, MAX_SPEED_DELAY_small_dist, \ START_DELAY_small_dist, FIN_DELAY_small_dist, MAX_SPEED_DELAY_long_dist, START_DELAY_long_dist, \ FIN_DELAY_long_dist, HALL, board_PIXEL, number_LED, brightness, line_script_Wait, R_wait, G_wait, B_wait, R2_wait, \ G2_wait, B2_wait, initial_Xaxis, initial_Yaxis, size_Xaxis, size_Yaxis, pixel, SPR, SYSCOF, specialNavigationRequests from led_controller import CLEAR try: GPIOboard = [board.D12, board.D18, board.D21, board.D10] connected_socket = client_socket setting_data = b'' # Накапливаем данные в байтовом буфере while True: chunk = client_socket.recv(BUFFER_SIZE) if not chunk: break # Если данные закончились, выходим из цикла setting_data += chunk # Добавляем прочитанные байты # Проверяем, достаточно ли данных для декодирования в UTF-8 try: decoded_data = setting_data.decode('utf-8') # Попытка декодировать данные json.loads(decoded_data) # Попытка загрузить JSON break # Если успех, выходим из цикла except UnicodeDecodeError: continue # Продолжаем читать данные, если возникла ошибка декодирования except json.JSONDecodeError: continue # Продолжаем читать данные, если JSON не полный # if not setting_data.endswith('}'): # setting_data += '}' # Добавляем закрывающую кавычку if setting_data: decoded_data = setting_data.decode('utf-8') settings_dict = json.loads(setting_data) autotest_client = settings_dict.get("autoTest", False) flage_remove_cV = settings_dict.get("flageremovecV", False) counter_client_play_video = int(settings_dict.get("counterClientPlayVideo", 0)) counter_setting_request = int(settings_dict.get("countersettingrequest", 0)) number_LED = int(settings_dict.get("numberLED", 0)) DIR = int(settings_dict.get("directionPin", 0)) STEP = int(settings_dict.get("stepPin", 0)) ENA = int(settings_dict.get("enablePin", 0)) DEGSTEP = float(settings_dict.get("degstep", 0)) CW = int(settings_dict.get("CW", 0)) CCW = int(settings_dict.get("CCW", 0)) MICROSTEP = int(settings_dict.get("microstep", 0)) FACTOR = int(settings_dict.get("factor", 0)) MAX_SPEED_DELAY_small_dist = float(settings_dict.get("speedDelay_sd", 0)) START_DELAY_small_dist = float(settings_dict.get("startSpeedDelay_sd", 0)) FIN_DELAY_small_dist = float(settings_dict.get("finSpeedDelay_sd", 0)) MAX_SPEED_DELAY_long_dist = float(settings_dict.get("finSpeedDelay_ld", 0)) START_DELAY_long_dist = float(settings_dict.get("startSpeedDelay_ld", 0)) FIN_DELAY_long_dist = float(settings_dict.get("finSpeedDelay_sd", 0)) HALL = int(settings_dict.get("hallPin", 0)) board_PIXEL = int(settings_dict.get("pixelBoard", 0)) brightness = int(settings_dict.get("brightness", 0)) line_script_Wait = int(settings_dict.get("lineScriptWait", 0)) colorWait = settings_dict.get("colorWait", "#000000") secondColorWait = settings_dict.get("secondColorWait", "#000000") initial_Xaxis = int(settings_dict.get("initXaxis", 0)) initial_Yaxis = int(settings_dict.get("initYaxis", 0)) size_Xaxis = int(settings_dict.get("sizeXaxis", 0)) size_Yaxis = int(settings_dict.get("sizeYaxis", 0)) specialNavigationRequests = settings_dict.get("specialNavigationRequests", []) number_LED = int(number_LED) DIR = int(DIR) STEP = int(STEP) ENA = int(ENA) CW = int(CW) CCW = int(CCW) DEGSTEP = float(DEGSTEP) SPR = 360 / DEGSTEP MICROSTEP = int(MICROSTEP) FACTOR = int(FACTOR) SYSCOF = FACTOR * MICROSTEP MAX_SPEED_DELAY_small_dist = float(MAX_SPEED_DELAY_small_dist) START_DELAY_small_dist = float(START_DELAY_small_dist) FIN_DELAY_small_dist = float(FIN_DELAY_small_dist) MAX_SPEED_DELAY_long_dist = float(MAX_SPEED_DELAY_long_dist) START_DELAY_long_dist = float(START_DELAY_long_dist) FIN_DELAY_long_dist = float(FIN_DELAY_long_dist) HALL = int(HALL) board_PIXEL = int(board_PIXEL) PIXEL_PIN = GPIOboard[board_PIXEL] brightness = int(brightness) line_script_Wait = int(line_script_Wait) R_wait = int(colorWait[1:3], 16) G_wait = int(colorWait[3:5], 16) B_wait = int(colorWait[5:7], 16) R2_wait = int(secondColorWait[1:3], 16) G2_wait = int(secondColorWait[3:5], 16) B2_wait = int(secondColorWait[5:7], 16) initial_Xaxis = int(initial_Xaxis) initial_Yaxis = int(initial_Yaxis) size_Xaxis = int(size_Xaxis) size_Yaxis = int(size_Yaxis) pixel = neopixel.NeoPixel(PIXEL_PIN, number_LED) # Значение pixel можно установить здесь, если это необходимо onClient("Настройки получены и обработаны", client_socket) else: print("close setting_server") pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) except ConnectionResetError: print("close setting_server") if pixel: pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) except OSError: print("close onClient") if pixel: pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) return autotest_client, flage_remove_cV, counter_client_play_video, counter_setting_request, \ DIR, STEP, ENA, DEGSTEP, CW, CCW, MICROSTEP, FACTOR, MAX_SPEED_DELAY_small_dist, \ START_DELAY_small_dist, FIN_DELAY_small_dist, MAX_SPEED_DELAY_long_dist, START_DELAY_long_dist, \ FIN_DELAY_long_dist, HALL, board_PIXEL, number_LED, brightness, line_script_Wait, R_wait, G_wait, B_wait, R2_wait, \ G2_wait, B2_wait, initial_Xaxis, initial_Yaxis, size_Xaxis, size_Yaxis, pixel, SPR, SYSCOF, specialNavigationRequests def get_specialNavigationRequests(): global specialNavigationRequests print(f'Массив с особыми видео: {specialNavigationRequests}') return specialNavigationRequests def setting_request(): # Обработка команд от клиента from video_player import find_file from led_controller import CLEAR global reqType, necessarystep, screenplay, color, color2, setting, connected_socket, namefile, setting_client, DEGSTEP try: counter = 0 while counter <= counter_setting_request: if len(setting_client) > 0: # start_thread() setting = setting_client[0] message_client = setting_client[0] print(f"Команда в обработке: {setting}") start_thread(message_client) try: reqType, deg, namefile, screenplay, R, G, B, R2, G2, B2 = setting.split() deg = int(deg) necessarystep = deg / DEGSTEP necessarystep = math.trunc(necessarystep) screenplay = int(screenplay) R = int(color[1:3], 16) G = int(color[3:5], 16) B = int(color[5:7], 16) R2 = int(color2[1:3], 16) G2 = int(color2[3:5], 16) B2 = int(color2[5:7], 16) break except ValueError: # Если не удается распаковать значения, просто продолжайте выполнение цикла print(f"Ошибка при распаковке значений из {setting}. Пропускаем.") join_thread() break time.sleep(1) counter = counter + 1 if counter > counter_setting_request: print("close setting_request") # pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) except ConnectionResetError: print("close setting_request") # pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) except OSError: print("close onClient") # pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) return reqType, deg, namefile, screenplay, R, G, B, R2, G2, B2, necessarystep, setting def get_setting_client(): global setting_client return setting_client def filepath_setting_request(): # Обработка названия видеофайла для воспроизведения от клиента from video_player import find_file global filepath, filepath_settinh_client if len(filepath_settinh_client) > 0: namefile = filepath_settinh_client[0] if namefile == '0': return 0 else: file_to_find = f'{namefile}.mp4' filepath = find_file(file_to_find) # поиск пути файла для воспроизведения print(f'filepath {filepath}') return filepath def client_request(): # Обработка видео from led_controller import CLEAR global message_play_video, message_client, connected_socket, color, color2, setting_client counter_of_itterations = 0 try: while True: if counter_of_itterations == 10: onClient('Ping') message_client = connected_socket.recv(BUFFER_SIZE).decode() if message_client: # Попытка декодировать message_client как JSON try: command_data = json.loads(message_client) except json.decoder.JSONDecodeError: # Если не удается декодировать как JSON, предполагаем, что это строка 'Старт видео' if message_client == 'Старт видео': set_message_play_video('Старт видео') print(f"Получена команда: {message_play_video}") onClient("Команда получена и обработана") else: print("Не удалось декодировать как JSON и не является 'Старт видео'") os.execv(sys.executable, [sys.executable] + sys.argv) continue try: # Получение значений из JSON-объекта reqType = command_data.get('reqType', '') deg = command_data.get('deg', 0) namefile = command_data.get('namefile', '') screenplay = command_data.get('screenplay', 0) color = command_data.get('hexcolour', '#000000') color2 = command_data.get('hexcolour2', '#000000') R = int(color[1:3], 16) G = int(color[3:5], 16) B = int(color[5:7], 16) R2 = int(color2[1:3], 16) G2 = int(color2[3:5], 16) B2 = int(color2[5:7], 16) # Добавление данных в список setting_client setting_client.append(f"{reqType} {deg} {namefile} {screenplay} {R} {G} {B} {R2} {G2} {B2}") counter_of_itterations += 1 except json.decoder.JSONDecodeError: pass onClient("Команда получена и обработана") continue except ConnectionResetError: print("close client_request") pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) except OSError: print("close onClient") pixel.fill(CLEAR) time.sleep(3) gpio.cleanup() if connected_socket: connected_socket.close() # Закрыть сокет if server_socket: server_socket.close() os.execv(sys.executable, [sys.executable] + sys.argv) def get_setting_client(): global setting_client return setting_client def get_message_play_video(): global message_play_video return message_play_video def set_message_play_video(new_message_play_video): global message_play_video message_play_video = new_message_play_video def check_message_client(message_client): from video_player import add_scroll_text global filepath_settinh_client, parts, namefile print('Открыли поток thread_check_message_client') parts = message_client.split() print(parts) print('=================================================================') if len(parts) == 10 and parts[2].startswith('objectNav_'): new_parts = parts[2][len('objectNav_'):] text_name = new_parts[:30] # Строим полный путь к файлу file_path = os.path.join(navigation_directory, f'{text_name}.mp4') print(f'Директория с видео: {file_path}') # Проверяем, существует ли файл if os.path.isfile(file_path): print(f'Файл {text_name}.mp4 найден.') else: print(f'Файл {text_name}.mp4 не найден.') subprocess.run(add_scroll_text('object_navigation', new_parts)) # if os.path.isfile(os.path.join(f'/home/blade/Videos/VDNH/createdNavigation/{text_name}.mp4')): # print(f'Файл: {text_name} имеется') # else: # print(f'Компонент: {new_parts}') # subprocess.run(add_scroll_text('object_navigation', new_parts)) parts[2] = text_name print(f'parts[2] {parts[2]}') message_client = ' '.join(parts) print(f'message_client {message_client}') if len(parts) == 10 and parts[2].startswith('objectEvent_'): # Убираем префикс и получаем основную часть new_parts = parts[2][len('objectEvent_'):] # Находим позицию второго нижнего подчёркивания # Если подчеркиваний меньше двух, берем весь new_parts underscore_positions = [pos for pos, char in enumerate(new_parts) if char == '_'] # Находим позицию первого нижнего подчёркивания first_underscore_position = new_parts.find('_') if first_underscore_position != -1: new_str = new_parts[:first_underscore_position] # Обрезаем до первого подчёркивания else: new_str = new_parts # Используем полную строку, если подчёркиваний нет # Строим полный путь к файлу file_path = os.path.join(event_directory, f'{new_str}.mp4') print(f'Полный путь к файлу для события: {file_path}||||||||||||||||||||||||||||||||||||||||||||||||||||||||') # Проверяем, существует ли файл if os.path.isfile(file_path): print(f'Файл: {new_str} имеется') else: print(f'Компонент: {new_str}') subprocess.run(add_scroll_text('event', new_parts)) # Обновляем parts для последующего использования parts[2] = new_str message_client = ' '.join(parts) print(f'Обновленное сообщение клиента: {message_client}') if len(parts) == 10 and parts[2].startswith('pogoda_'): # Проверяем, существует ли файл, и удаляем его, если он есть if os.path.isfile(weather_video_path): os.remove(weather_video_path) print(f'Видео с погодой по пути {weather_video_path} стёрто для перезаписи') # if os.path.isfile(os.path.join('/home/blade/Videos/VDNH/createdVideos/pogoda.mp4')): # os.remove(os.path.join('/home/blade/Videos/VDNH/createdVideos/pogoda.mp4')) new_parts = parts[2][len('pogoda_'):] print(f'Компонент: {new_parts}') subprocess.run(add_scroll_text('weather forecast', new_parts)) parts[2] = 'pogoda' print(f'parts[2] {parts[2]}') message_client = ' '.join(parts) print(f'message_client {message_client}') if len(parts) == 10 and parts[2].startswith('traffic_'): new_parts = parts[2][len('triffic_'):] print(f'Компонент: {new_parts}') # text_name = new_parts[:30] # Формируем путь к файлу внутри папки createdVideos video_path = os.path.join(base_path, 'data', 'Videos', 'VDNH', 'createdVideos', f'traffic_{new_parts}.mp4') if os.path.isfile(video_path): print(f'Файл: traffic_{new_parts} имеется') else: print(f'Компонент: {new_parts}') # Запускаем функцию добавления текста на видео subprocess.run(add_scroll_text('triffic widget', new_parts)) parts[2] = f'traffic_{new_parts}' print(f'parts[2] {parts[2]}') message_client = ' '.join(parts) print(f'message_client {message_client}') # Добавление в очеред последнего запроса namefile = parts[2] if not filepath_settinh_client or (filepath_settinh_client[-1] != namefile): filepath_settinh_client.append(namefile) print(f'В массив добавлено видео: {filepath_settinh_client}.') else: print(f'Файл {namefile} уже есть в очереди.') print(f'В массив добавлено видео: {filepath_settinh_client}.') def get_filepath_settinh_client(): global filepath_setting_request return filepath_settinh_client def get_filepath(): global filepath return filepath def set_filepath(filepath_init): global filepath filepath = filepath_init def request_above(flag_join_thread): # Обработка запроса с приоритетов выше global setting_client, filepath_settinh_client print('_______________________________________request_above______________________________________________') setting_client = get_setting_client() if len(setting_client) > 1: print('Обработка запроса с приоритетов выше') if flag_join_thread: join_thread() setting_client = get_setting_client() if setting_client: setting_client.pop(0) filepath_settinh_client = get_filepath_settinh_client() if filepath_settinh_client: filepath_settinh_client.pop(0) print(f"Запрос {setting_client} отменён.") onClient('Запрос отменен') return True return False def start_thread(message_client): # Запуск в отдельном потоке check_message_client global setting_client, thread_check_message_client print("Пробуем открыть поток thread_check_message_client") if len(setting_client) > 0: message_client = setting_client[-1] print('Открыли поток thread_check_message_client') thread_check_message_client = Thread(target=check_message_client, args=(message_client,)) thread_check_message_client.start() else: print(f'Поток по «{message_client}» не был открыт') return def join_thread(): # Закрытие потока с check_message_client from video_player import find_file global filepath, setting_client, filepath_settinh_client, thread_check_message_client, namefile print("Пробуем закрыть поток thread_check_message_client") if len(setting_client) > 0: print(f'Запрос клиента во время захода в поток: {setting_client} ||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||') print('Поток завершился, продолжаем выполнение') thread_check_message_client.join() if namefile == '0': filepath = None print(f'Путь «filepath»: {filepath}') return else: file_to_find = f'{namefile}.mp4' # search_directory = "/home/blade/Videos/VDNH" search_directory = os.path.join(base_path, 'data', 'Videos', 'VDNH') filepath = find_file(file_to_find, search_directory) # поиск пути файла для воспроизведения print(f'Путь «filepath»: {filepath}')