58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
|
|
from CoreLibrary.Device import Device
|
|
|
|
LOGGER = logging.getLogger("my_logger")
|
|
|
|
|
|
class Measurement:
|
|
"""
|
|
Handle data and write it in a file
|
|
"""
|
|
|
|
def __init__(self, name, path=None, comment=None, writer=None):
|
|
self.name = name
|
|
self.path = os.path.join(path or "./meas_values", self.name)
|
|
os.makedirs(self.path)
|
|
self.save_version_text(comment)
|
|
self.devices = {}
|
|
self.data_files = {"log": self.create_file("log")}
|
|
|
|
def create_file(self, name: str):
|
|
return open(os.path.join(self.path, name), "w")
|
|
|
|
def attach_device(self, device: Device, name: str):
|
|
self.devices[name] = device
|
|
|
|
def write_log(self, text: str):
|
|
self.data_files["log"].write(str(datetime.datetime.utcnow()) + ":\t" + text + "\n")
|
|
|
|
def append_data(self, device_name, data):
|
|
"""
|
|
Save data from a device. Write it into the corresponding file.
|
|
|
|
TODO: file size limit, zip?
|
|
data: list of tuples: (timestamp, value)
|
|
"""
|
|
if device_name not in self.devices:
|
|
return
|
|
|
|
LOGGER.info("append-data: " + device_name)
|
|
if device_name not in self.data_files:
|
|
self.data_files[device_name] = self.create_file(device_name)
|
|
for timestamp, value in zip(data[0], data[1]):
|
|
self.data_files[device_name].write(
|
|
f"{timestamp.isoformat()}, {timestamp.timestamp()}, {value}\n"
|
|
)
|
|
|
|
def save_version_text(self, comment):
|
|
params = {
|
|
"version": "0.1",
|
|
"Start time": datetime.datetime.utcnow().isoformat(),
|
|
}
|
|
with open(os.path.join(self.path, "params.json"), "w") as file:
|
|
json.dump(params, file, indent=4)
|