import datetime import json import logging import os from CoreLibrary.Driver import Driver from CoreLibrary.Sensor import Sensor 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.sensors = {} self.drivers = {} self.data_files = {} 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_sensor(self, sensor: Sensor, name: str): self.sensors[name] = sensor def attach_driver(self, driver: Driver, name: str): self.drivers[name] = driver def attach_combined(self, comb, name: str): self.attach_sensor(comb, name) self.attach_driver(comb, name) def write_log(self, text: str): self.data_files["log"].write(str(datetime.datetime.utcnow()) + ":\t" + text + "\n") def append_data(self, sensor_name, data): """ Save data from a sensor. Write it into the corresponding file. TODO: file size limit, zip? data: list of tuples: (timestamp, value) """ if sensor_name not in self.sensors: return LOGGER.info("append-data: " + sensor_name) if sensor_name not in self.data_files: self.data_files[sensor_name] = self.create_file(sensor_name) for timestamp, value in zip(data[0], data[1]): self.data_files[sensor_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)