MatthiasQuintern/cpdctrl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

save csv

Browse Source
This commit is contained in:
CPD 2025-02-04 16:55:38 +01:00
parent 76e0a4e64a
commit cbd3aa0c7c
1 changed files with 37 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
cpdctrl/utility/data.py
View File

@ -10,16 +10,17 @@ FLUSH_TYPE = "pickle-ndarray"
METADATA_FILENAME = "_measurement_metadata.pkl"
class DataCollector:
columns = ["idx", "t [s]", "V [V]", "LED [%]"]
def __init__(self,
data_path: str,
data_name: str="CPData",
header: dict[str, str]={},
metadata: dict[str, str]={},
dirname: str|None=None,
dir_exists_is_ok=False,
):
self.data = []
self.name = data_name
self.header = header
self.metadata = metadata
self.path = os.path.abspath(os.path.expanduser(data_path))
if dirname is None:
self.dirname = sanitize_filename(datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%d_%H-%M") + "_" + self.name)
@ -33,14 +34,22 @@ class DataCollector:
else:
os.makedirs(self.dirpath)
self.flushed = False
def _get_filename(self):
return sanitize_filename(get_next_filename(self.name, self.dirpath, digits=5))
def _write_metadata(self):
def write_metadata(self):
f"""
Write the metadata to the disk as '{METADATA_FILENAME}'
Returns
-------
None.
"""
filepath = os.path.join(self.dirpath, METADATA_FILENAME)
with open(filepath, "wb") as file:
pickle.dump(self.header, file)
pickle.dump(self.metadata, file)
def flush(self, verbose:bool=False):
"""
@ -66,7 +75,7 @@ class DataCollector:
filename = self._get_filename() + ".csv"
filepath = os.path.join(self.dirpath, filename)
if verbose: print(f"Flushing data to {filepath}")
self.to_dataframe().to_csv(filepath, sep=",", index=False, header=True)
self.to_dataframe().to_csv(filepath, sep=",", index=False, metadata=True)
elif FLUSH_TYPE == "pickle-ndarray":
filename = self._get_filename() + ".ndarray.pkl"
filepath = os.path.join(self.dirpath, filename)
@ -85,15 +94,34 @@ class DataCollector:
self.data.append((i, t, v, l))
def to_dataframe(self):
df = pd.DataFrame(self.data, columns=["idx", "t [s]", "V [V]", "LED [%]"])
df.meta = str(self.header)
df = pd.DataFrame(self.data, columns=DataCollector.columns)
df.meta = str(self.metadata)
return df
def save_csv(self):
self.to_dataframe().to_csv(os.path.join(self.path, self.name + ".csv"), index=False, header=True)
def to_csv(self, sep=","):
# self.to_dataframe().to_csv(os.path.join(self.path, self.name + ".csv"), index=False, metadata=True)
data, metadata = self.get_data()
return DataCollector.get_csv(data, self.metadata, sep=sep)
def save_csv(self, sep=",", verbose=False):
filepath = os.path.join(self.path, self.name + ".csv")
if verbose: print(f"Writing data to {filepath}")
with open(filepath, "w") as file:
file.write(self.to_csv(sep=sep))
def get_data(self):
return DataCollector.load_data(self.dirpath)
@staticmethod
def get_csv(data, metadata, sep=","):
csv = ""
for k, v in metadata.items():
csv += f"# {k}: {v}\n"
csv += "".join(f"{colname}{sep}" for colname in DataCollector.columns).strip(sep) + "\n"
for i in range(data.shape[0]):
csv += f"{i}{sep}{data[i,1]}{sep}{data[i,2]}{sep}{data[i,3]}\n"
return csv.strip("\n")
@staticmethod
def load_data(dirpath:str, verbose:bool=False) -> np.ndarray:
"""