Блог

У меня есть данные временных рядов в реальном времени с двух разных входных данных датчика. Я хотел бы упорядочить данные в соответствии с меткой времени.

Проблема в том, что входные данные датчика принимаются непоследовательно, например

  sensor      time  data1     data2     data3     data 4  
['sensor_1', '0', '0.1234', '0.3251', '1.4235', '0.4234']
['sensor_1', '1', '0.1432', '0.3452', '1.4245', '0.1434']
['sensor_2', '0', '0.6543', '0.3231', '1.1235', '0.3434']
['sensor_1', '2', '0.1654', '0.3243', '1.5445', '0.1234']
['sensor_2', '1', '0.1544', '0.3345', '1.5425', '0.6534']
['sensor_2', '2', '0.5432', '0.4551', '1.8755', '0.4245']
  

Как я могу присоединиться к ним (пока я получаю данные), чтобы данные датчика с той же меткой времени были объединены, например:

  time  data1_1   data1_2   data1_3   data1_4   data2_1   data2_2   data2_3   data2_4    
['0', '0.1234', '0.3251', '1.4235', '0.4234', '0.6543', '0.3231', '1.1235', '0.3434']
['1', '0.1432', '0.3452', '1.4245', '0.1434', '0.1544', '0.3345', '1.5425', '0.6534']
['2', '0.1654', '0.3243', '1.5445', '0.1234', '0.5432', '0.4551', '1.8755', '0.4245']
  

Я получаю данные через подпроцесс (по одной строке за раз):

 process = subprocess.Popen(['node', 'server.js'], stdout=subprocess.PIPE)
while True:
     Byte_data = process.stdout.readline()
     string_data = byte_data.rstrip().decode("utf-8") # From byte to string
     list_data = string_data.split(",") # Comma separate strings

     # Sort data here...
  

Вам нужно будет оглядываться назад и сопоставлять временные метки по мере их поступления, что-то вроде этого.

• Из формулировки вашего вопроса кажется, что временные метки совпадают. Если они не упорядочены, а только упорядочены, все становится немного сложнее.
• Если некоторые кадры отсутствуют, они будут задерживаться на неопределенный срок в lookback dicts. Если эта функция выполняется в течение длительных периодов времени, вы можете регулярно очищать старые кадры.

 def time(row):
    return row[1]

def synchronized(rows):
    lookback = {'sensor_1': {}, 'sensor_2': {}}
    for row in rows:
        # add row to lookback
        sensor = row[0]
        lookback[sensor][time(row)] = row

        # check for matching timestamps
        rms = []
        for t, r in sorted(lookback['sensor_1'].items(), key=time):
            if t in lookback['sensor_2']:
                yield r, lookback['sensor_2']
                rms.append(t)

        for rm in rms:
            for tmp in lookback.values():
                del tmp[rm]

def parse(rows):
    for row in rows:
        yield tuple(row.decode().rstrip().split(','))

import subprocess
process = subprocess.Popen(['node', 'server.js'], stdout=subprocess.PIPE)
for row1, row2 in synchronized(parse(process.stdout)):
    print(row1, row2)

  

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

 sorted_data = []

process = subprocess.Popen(['node', 'server.js'], stdout=subprocess.PIPE)
while True:
    Byte_data = process.stdout.readline()
    string_data = byte_data.rstrip().decode("utf-8") # From byte to string
    list_data = string_data.split(",") # Comma separate strings

    try:
        sorted_data[int(list_data[1])]  = list_data[2:]
    except IndexError:
        sorted_data.append(list_data[1:])

    # prints sorted data in real time
    print(str(sorted_data).replace("],", "],n"))

# prints out final list with sorted data
print(str(sorted_data).replace("],", "],n"))