peterlu14
111 lines
5.3 KiB
Python
111 lines
5.3 KiB
Python
import csv
|
|
import time
|
|
import json
|
|
from analysis_mode import VTmode_run, slop_run ,CVmode_run, movstd_run
|
|
from database_api import get_raw_id_list, get_raw_data, get_raw_data_with_time
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
from db.base import Base, Session, engine
|
|
from db.subject_data import SubjectData
|
|
|
|
async def async_callback(topic, payload, conn, client):
|
|
"""
|
|
payload format
|
|
{
|
|
"pattern": {
|
|
"id": number,
|
|
"name": string,
|
|
"parameter": object,
|
|
},
|
|
"data": {
|
|
"id": list[int],
|
|
"channel: list[int]
|
|
},
|
|
"others": {
|
|
...
|
|
},
|
|
}
|
|
"""
|
|
print(f"Received message on {topic}: {payload}")
|
|
start_time = time.time()
|
|
input_data = json.loads(payload)
|
|
meta = input_data['data']
|
|
analysis_pattern = input_data['pattern']
|
|
others = input_data.get('others')
|
|
|
|
# with open('csv_file/output.csv', 'w', newline='') as csvfile:
|
|
# writer = csv.writer(csvfile)
|
|
# write_header_done = False
|
|
meta_data = {}
|
|
for meta_id in meta['id']:
|
|
meta_data[meta_id] = {}
|
|
# TODO write file header
|
|
# create meta_data
|
|
if 'Time' in meta['channel']:
|
|
for channel in meta['channel']:
|
|
if channel != 'Time':
|
|
meta_data[meta_id]['Time'] = []
|
|
meta_data[meta_id][channel] = []
|
|
raw_id_list = await get_raw_id_list(conn, meta_id, channel)
|
|
for raw_id in raw_id_list:
|
|
time_data, raw_data = await get_raw_data_with_time(conn, raw_id, channel)
|
|
meta_data[meta_id]['Time'].extend(time_data)
|
|
meta_data[meta_id][channel].extend(raw_data)
|
|
else:
|
|
for channel in meta['channel']:
|
|
meta_data[meta_id][channel] = []
|
|
raw_id_list = await get_raw_id_list(conn, meta_id, channel)
|
|
for raw_id in raw_id_list:
|
|
raw_data = await get_raw_data(conn, raw_id, channel)
|
|
meta_data[meta_id][channel].extend(raw_data)
|
|
# results = await asyncio.gather(*(get_raw_data(conn, raw_id, channel) for raw_id in raw_id_list))
|
|
|
|
|
|
with ProcessPoolExecutor() as executor:
|
|
channelX = meta['channel'][0]
|
|
channelY = meta['channel'][1]
|
|
|
|
if analysis_pattern['id'] == 0:
|
|
for meta_id in meta['id']:
|
|
data = [meta_id, meta_data[meta_id][channelX], meta_data[meta_id][channelY]]
|
|
# mqtt publish data
|
|
await client.publish("data_analysis", data)
|
|
elif analysis_pattern['id'] == 1:
|
|
for meta_id, result in zip(meta['id'], executor.map(VTmode_run, [[meta_data[meta_id][channelX],meta_data[meta_id][channelY],analysis_pattern['parameter']['percentage']] for meta_id in meta['id']])):
|
|
print('result', meta_id, result)
|
|
|
|
# do analysis function
|
|
# result, csv_data = VTmode_run(x_data, y_data, analysis_pattern['parameter']['percentage'])
|
|
data = [meta_id, meta_data[meta_id][channelX], meta_data[meta_id][channelY], *result[0]]
|
|
|
|
# mqtt publish data
|
|
await client.publish("data_analysis", data)
|
|
|
|
# # create dict writer
|
|
# fieldnames = list(csv_data.keys())
|
|
# dict_writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
|
|
|
|
# # write data header
|
|
# if not write_header_done:
|
|
# dict_writer.writeheader()
|
|
# write_header_done = True
|
|
|
|
# # write data
|
|
# dict_writer.writerow(csv_data)
|
|
elif analysis_pattern['id'] == 4:
|
|
with Session() as session:
|
|
for meta_id in meta['id']:
|
|
print('data' , len(meta_data[meta_id][channelX]), len(meta_data[meta_id][channelY]))
|
|
result = movstd_run([meta_data[meta_id][channelX], meta_data[meta_id][channelY]])
|
|
print('movstd_run', result)
|
|
print(others)
|
|
subject_data = SubjectData(
|
|
subject_id= others["subject"]["id"],
|
|
project = others['project'],
|
|
meta = others['project_meta'],
|
|
mode= others['device'],
|
|
data= result[0]
|
|
)
|
|
session.add(subject_data)
|
|
session.commit()
|
|
|
|
print("執行時間：%f 秒" % (time.time() - start_time)) |