Чтение данных в реальном времени из внешнего файла .txt, который перезаписывает ⇐ Python

[Anonymous]

Я работаю над программой Python, которая выполняет вторичную программу. Эта дополнительная программа выполняет измерения в реальном времени с помощью устройства Bluetooth (WT901BLECL), которые связаны с основными движениями (движение вперед, назад, влево, вправо, остановка), и записывает результаты в файл .txt. Текстовый файл имеет только одну строку и записывает значения от 0 до 4, каждое из которых связано с соответствующей командой перемещения.
При чтении файла в основной программе я не могу проверить, чтение выполняется правильно, поскольку мой терминал показывает только выходные данные выполненного второго файла. Однако я знаю, что чтение выполняется неправильно, поскольку у меня также есть соединение Arduino, которое получает данные, отправленные основной программой Python. Полученные данные соответствуют данным реального времени в файле .txt. Однако при проверке, получает ли программа Arduino какие-либо данные, я ничего не вижу.
Это мой основной код, второй код которого выполняется после нажатия клавиши «w»:
р>

Код: Выделить всё

import serial  # Para comunicación serial
import time  # Para pausas temporales
from pynput import keyboard  # Para escuchar eventos de teclado
import subprocess  # Para ejecutar comandos del sistema
import re
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# Configurar la conexión serial con Arduino
puerto_serial = serial.Serial('COM10', 9600)  # Ajusta a tu puerto COM
time.sleep(2)  # Esperar para que Arduino se inicie

ultimo_numero = ""
ultimo_valor = ""

# Variable para controlar el Listener
finalizar = False  # Para verificar si el programa debe finalizar

# Variable para almacenar la última tecla conocida
ultima_tecla = None

def imprimir_ultimo_numero(data):
global ultimo_numero, ultimo_valor
if data and data != ultimo_numero:   # Verificar si hay datos y si son diferentes al último número
ultimo_numero = obtener_ultimo_numero(data)  # Obtener el último número
print("Último número:", ultimo_numero)  # Imprimir el último número

# Acciones en función del último número obtenido
if ultimo_numero == '1':
print("Flecha hacia arriba")
puerto_serial.write(b'a\n')
if ultimo_valor != '1':
puerto_serial.write(b'a\n')
ultimo_valor = ultimo_numero
elif ultimo_numero == '2':
print("Flecha hacia abajo")
puerto_serial.write(b'b\n')
if ultimo_valor != '2':
puerto_serial.write(b'b\n')
ultimo_valor = ultimo_numero
elif ultimo_numero == '3':
print("Flecha hacia izquierda")
puerto_serial.write(b'l\n')
if ultimo_valor != '3':
puerto_serial.write(b'l\n')
ultimo_valor = ultimo_numero
elif ultimo_numero == '4':
print("Flecha hacia derecha")
puerto_serial.write(b'r\n')
if ultimo_valor != '4':
puerto_serial.write(b'r\n')
ultimo_valor = ultimo_numero
elif ultimo_numero == '0':
print("PARAR")
puerto_serial.write(b'p\n')
if ultimo_valor != '0':
puerto_serial.write(b'p\n')
ultimo_valor = ultimo_numero
else:
print("Posición no reconocida")
puerto_serial.write(b'p\n')
ultimo_valor = ultimo_numero

def obtener_ultimo_numero(data):
# Eliminar los caracteres no numéricos a la derecha
data = ''.join(filter(str.isdigit, data))
# Si la cadena resulta vacía, devolvemos '0'
if not data:
return '0'
# Obtener el último número de la cadena
return data[-1]

class MyHandler(FileSystemEventHandler):
def on_modified(self, event):
if event.src_path.endswith("x_value.txt"):
data = leer_archivo(event.src_path)
imprimir_ultimo_numero(data)

def leer_archivo(filename):
with open(filename) as fh:
line = fh.read()
return line

def on_press(key):
global finalizar
global ultima_tecla

try:
# Verificar si es una tecla imprimible
if hasattr(key, 'char') and key.char is not None:
char = key.char

# Finalizar si se presiona 'q'
if char == 'q':
print("Presionaste 'q', terminando...")
finalizar = True
return False
if  ultima_tecla != 'q':
puerto_serial.write(b'q\n')
ultima_tecla = char

if char == 'a':
puerto_serial.write(b'a\n')
print("Flecha hacia arriba")
if ultima_tecla != 'a':
puerto_serial.write(b'a\n')
ultima_tecla = char

elif char == 'b':
puerto_serial.write(b'b\n')
print("Flecha hacia abajo")
if ultima_tecla != 'b':
puerto_serial.write(b'b\n')
ultima_tecla = char

elif char == 'l':
puerto_serial.write(b'l\n')
print("Flecha hacia la izquierda")
if ultima_tecla != 'l':
puerto_serial.write(b'l\n')
ultima_tecla = char

elif char == 'r':
puerto_serial.write(b'r\n')
print("Flecha hacia la derecha")
if ultima_tecla != 'r':
puerto_serial.write(b'r\n')
ultima_tecla = char

elif char == 'p':
puerto_serial.write(b'p\n')
print("Flecha para parar")
if ultima_tecla != 'p':
puerto_serial.write(b'p\n')
ultima_tecla = char

elif char == 'w':
archivo_a_ejecutar = r'C:/Users/Lara\Desktop/TFG/Funciona_control_witmotion.py'
subprocess.run(['python',  archivo_a_ejecutar])

ultimo_valor = '0'

event_handler = MyHandler()
observer = Observer()
observer.schedule(event_handler, path='.', recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()

except Exception as e:
print(f"Error al procesar tecla: {e}")

with keyboard.Listener(on_press=on_press) as listener:
listener.join()  # Mantener el Listener hasta que se presione 'q'

if puerto_serial.is_open:
puerto_serial.close()
print("Puerto serial cerrado.")

Мой второй код, в котором я идентифицирую свое соединение Bluetooth, беру эталонные значения и выполняю сравнение в реальном времени между сохраненными данными и полученными значениями в реальном времени, приведен ниже:
(извините, мой код длинный, потому что я также определил, как извлекать данные с моего устройства Bluetooth)

Код: Выделить всё

import asyncio
import threading
import time
import struct
import bleak
import keyboard
import time
import sys
import serial

# we put this because sometimes the keys seemed to be pressed multiple times
last_key_press_time = 0

# Variable to control the Listener in order to check if data comparison should run continuously
comparing_data = False

# Variable to control the listener in order to check if the program should finish
finalizar = False

ultima_tecla = None

async def scan():
global devices
found = []
print("Searching for Bluetooth devices...")
try:
devices = await bleak.BleakScanner.discover()
print("Search ended")

for d in devices:
if d.name and "WT"  in d.name:
found.append(d)
print(f"{d.address}: {d.name}")
if not devices:
print("No devices found!")

return found
except Exception as ex:
print("Bluetooth search failed to start")
print(ex)

def is_within_margin(value, expected, margin):
if value is None or expected is None:
return False  # Si alguno de los valores es None, no podemos comparar
return abs(value - expected) > 8
return tempBytes

def unlock(self):
cmd = self.get_writeBytes(0x69, 0xb588)
self.sendData(cmd)

@staticmethod
def getSignInt16(num):
if num >= pow(2, 15):
num -= pow(2, 16)
return num

def save(self):
cmd = self.get_writeBytes(0x00, 0x0000)
self.sendData(cmd)

def onDataReceived(self, sender, data):
tempdata = bytes.fromhex(data.hex())
for var in tempdata:
self.TempBytes.append(var)
if len(self.TempBytes) == 1 and self.TempBytes[0] != 0x55:
del self.TempBytes[0]
continue
if len(self.TempBytes) == 2 and (self.TempBytes[1] != 0x61 and self.TempBytes[1] != 0x71):
del self.TempBytes[0]
continue
if len(self.TempBytes) == 20:
self.processData(self.TempBytes)
self.TempBytes.clear()

def processData(self, Bytes):
if Bytes[1] == 0x61:
AngX = self.getSignInt16(Bytes[15] = debounce_time:
last_key_press_time = current_time
ultima_tecla = '1'
self.save_data('1', "datos_witmotion_referencia.txt")

elif keyboard.is_pressed("2"):
current_time = time.time()
if current_time - last_key_press_time >= debounce_time:
last_key_press_time = current_time
ultima_tecla = '2'
self.save_data('2', "datos_witmotion_referencia.txt")

elif keyboard.is_pressed("3"):
current_time = time.time()
if current_time - last_key_press_time >= debounce_time:
last_key_press_time = current_time
ultima_tecla = '3'
self.save_data('3', "datos_witmotion_referencia.txt")

elif keyboard.is_pressed("4"):
current_time = time.time()
if current_time - last_key_press_time >= debounce_time:
last_key_press_time = current_time
ultima_tecla = '4'
self.save_data('4', "datos_witmotion_referencia.txt")

elif keyboard.is_pressed("s"):
current_time = time.time()
if current_time - last_key_press_time >= debounce_time:
last_key_press_time = current_time
comparing_data = True
print("Iniciando comparación de datos...")
self.start_continuous_comparison(referencias_archivo, self.compare_with_reference_data)

if keyboard.is_pressed("q"):
current_time = time.time()
if current_time - last_key_press_time >= debounce_time:
last_key_press_time = current_time
print("Finalizando programa...")
finalizar = True
self.stop_keyboard_listener = True
self.closeDevice()

async def compare_with_reference_data(self, nombre_archivo):
print("Performing comparison with reference data...")
# Leer los datos de referencia del archivo
reference_data = {}
with open(nombre_archivo, 'r') as archivo:
for linea in archivo:
partes = linea.strip().split()
if len(partes) == 4:
key = partes[0]
AngX = float(partes[1]) if partes[1] != 'None' else None
AngY = float(partes[2]) if partes[2] != 'None' else None
AngZ = float(partes[3]) if partes[3] != 'None' else None
reference_data[key] = {'AngX': AngX, 'AngY': AngY, 'AngZ': AngZ}

# Get current measurements
current_data = {
'AngX': self.deviceData.get('AngX'),
'AngY': self.deviceData.get('AngY'),
'AngZ': self.deviceData.get('AngZ')
}

margin = 15
x = self.deviceData.get("x", None)

# Write x value to a file
with open('x_value.txt', 'w') as f:
for position, saved_data in reference_data.items():
similarity = all(
is_within_margin(current_data[key], saved_data[key], margin)
for key in current_data
)
if similarity:
print(f"Current measurements are similar to {position} position data.")
print(current_data)
x = position
f.write(str(x))
return  # Detener la comparación después de encontrar una coincidencia
else:
x = 0
f.write(str(x))
print("Current measurements do not match any saved position within the margin of error.")

def start_continuous_comparison(self, file_name, comparison_method):
asyncio.run(self.continuous_comparison(file_name, comparison_method))

async def continuous_comparison(self, file_name, comparison_method):
while comparing_data:   # while TRUE compare data
await comparison_method(file_name)
await asyncio.sleep(2)  # espera 2s entre comparaciones

def stop_keyboard_listener(self):
self.stop_keyboard_listener = True

async def main():
# Define a placeholder callback method
async def callback_method(device):
pass

# Create an instance of the DeviceModel
device = DeviceModel("My Device", "F9:E7:DA:06:2C:4C", callback_method)

# Open the device
await device.openDevice()

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

Я пробовал следующее.

• Программа, которая запускает файл, генерирующий числа в уникальной строке, и принимает последнее значение (та же идея, что и в моей программе)

Код: Выделить всё

import time
import subprocess
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import serial

puerto_serial = serial.Serial('COM10', 9600)  # Ajusta a tu puerto COM

# Variable para almacenar el último número del archivo
ultimo_numero = ""
ultimo_valor = ""

def imprimir_ultimo_numero(data):
global ultimo_numero, ultimo_valor
if data and data != ultimo_numero:   # Verificar si hay datos y si son diferentes al último número
ultimo_numero = obtener_ultimo_numero(data)  # Obtener el último número
print("Último número:", ultimo_numero)  # Imprimir el último número

# Acciones en función del último número obtenido
if ultimo_numero == '1':
print("Flecha hacia arriba")
puerto_serial.write(b'a\n')
if ultimo_valor != '1':
puerto_serial.write(b'a\n')
ultimo_valor = ultimo_numero
elif ultimo_numero == '2':
print("Flecha hacia abajo")
puerto_serial.write(b'b\n')
if ultimo_valor != '2':
puerto_serial.write(b'b\n')
ultimo_valor = ultimo_numero
elif ultimo_numero == '3':
print("Flecha hacia izquierda")
puerto_serial.write(b'l\n')
if ultimo_valor != '3':
puerto_serial.write(b'l\n')
ultimo_valor = ultimo_numero
elif ultimo_numero == '4':
print("Flecha hacia derecha")
puerto_serial.write(b'r\n')
if ultimo_valor != '4':
puerto_serial.write(b'r\n')
ultimo_valor = ultimo_numero
elif ultimo_numero == '0':
print("PARAR")
puerto_serial.write(b'p\n')
if ultimo_valor != '0':
puerto_serial.write(b'p\n')
ultimo_valor = ultimo_numero
else:
print("Posición no reconocida")
puerto_serial.write(b'p\n')
ultimo_valor = ultimo_numero

def obtener_ultimo_numero(data):
# Eliminar los caracteres no numéricos a la derecha
data = ''.join(filter(str.isdigit, data))
# Si la cadena resulta vacía, devolvemos '0'
if not data:
return '0'
# Obtener el último número de la cadena
return data[-1]

class MyHandler(FileSystemEventHandler):
def on_modified(self, event):
if event.src_path.endswith("x_value.txt"):
data = leer_archivo(event.src_path)
imprimir_ultimo_numero(data)

def leer_archivo(filename):
with open(filename) as fh:
line = fh.read()
return line

archivo_a_ejecutar = r'C:/Users/Lara\Desktop/TFG/comprobar_generar_lectura.py'
subprocess.Popen(['python', archivo_a_ejecutar])

# Suponiendo que tienes un puerto serial configurado como "puerto_serial"
ultimo_valor = '0'

event_handler = MyHandler()
observer = Observer()
observer.schedule(event_handler, path='.', recursive=False)
observer.start()

try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()

# Cerrar el puerto serial al finalizar
if puerto_serial.is_open:
puerto_serial.close()
print("Puerto serial cerrado.")

• Файл, генерирующий значения

Код: Выделить всё

import random
import cv2
import time

def generar_numeros():
numeros = []
for _ in range(random.randint(1, 4)):
numero = random.randint(0, 4)
numeros.append(numero)
if len(numeros) > 1:
numeros = [0] * (len(numeros) - 1) + [numero]
linea = ''.join(map(str, numeros))
with open('x_value.txt', 'w') as archivo:
archivo.write(linea)
time.sleep(3)

if __name__ == "__main__":
start_time = time.time()
while True:
time.sleep(3)

generar_numeros()
key = cv2.waitKey(1)
if key == ord('q') or time.time() - start_time > 40:  # Salir del bucle después de 10 segundos o si se presiona 'q'
break

И в этом случае я могу отправлять данные на Arduino

Подробнее здесь: https://stackoverflow.com/questions/784 ... overwrites