mirror of
https://github.com/echemdata/galvani.git
synced 2025-12-12 16:45:35 +00:00
Add loop_from_file and timestamp_from_file functions
to extract loop_index and timestamp from the temporary _LOOP.txt and .mpl files during MPRfile initialization Added unit tests but cannot upload test files due to LFS quota exceeded Edited by Chris Kerr to fix flake8 warnings and resolve my comments from the original PR https://github.com/echemdata/galvani/pull/102
This commit is contained in:
@@ -10,6 +10,7 @@ __all__ = ["MPTfileCSV", "MPTfile"]
|
||||
import re
|
||||
import csv
|
||||
from os import SEEK_SET
|
||||
import os.path
|
||||
import time
|
||||
from datetime import date, datetime, timedelta
|
||||
from collections import defaultdict, OrderedDict
|
||||
@@ -541,6 +542,85 @@ def read_VMP_modules(fileobj, read_module_data=True):
|
||||
fileobj.seek(hdr_dict["offset"] + hdr_dict["length"], SEEK_SET)
|
||||
|
||||
|
||||
def loop_from_file(file: str, encoding: str = "latin1"):
|
||||
"""
|
||||
When an experiment is still running and it includes loops,
|
||||
a _LOOP.txt file is temporarily created to progressively store the indexes of new loops.
|
||||
This function reads the file and creates the loop_index array for MPRfile initialization.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
file : str
|
||||
Path of the loop file.
|
||||
encoding : str, optional
|
||||
Encoding of the text file. The default is "latin1".
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If the file does not start with "VMP EXPERIMENT LOOP INDEXES".
|
||||
|
||||
Returns
|
||||
-------
|
||||
loop_index : np.array
|
||||
Indexes of data points that start a new loop.
|
||||
|
||||
"""
|
||||
with open(file, "r", encoding=encoding) as f:
|
||||
line = f.readline().strip()
|
||||
if line != LOOP_MAGIC:
|
||||
raise ValueError("Invalid magic for LOOP.txt file")
|
||||
loop_index = np.array([int(line) for line in f], dtype="u4")
|
||||
|
||||
return loop_index
|
||||
|
||||
|
||||
def timestamp_from_file(file: str, encoding: str = "latin1"):
|
||||
"""
|
||||
When an experiment is still running, a .mpl file is temporarily created to store
|
||||
information that will be added in the log module and will be appended to the data
|
||||
module in the .mpr file at the end of experiment.
|
||||
This function reads the file and extracts the experimental starting date and time
|
||||
as a timestamp for MPRfile initialization.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
file : str
|
||||
Path of the log file.
|
||||
encoding : str, optional
|
||||
Encoding of the text file. The default is "latin1".
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If the file does not start with "EC-Lab LOG FILE" or "BT-Lab LOG FILE".
|
||||
|
||||
Returns
|
||||
-------
|
||||
timestamp
|
||||
Date and time of the start of data acquisition
|
||||
"""
|
||||
with open(file, "r", encoding=encoding) as f:
|
||||
line = f.readline().strip()
|
||||
if line not in LOG_MAGIC:
|
||||
raise ValueError("Invalid magic for .mpl file")
|
||||
log = f.read()
|
||||
start = tuple(
|
||||
map(
|
||||
int,
|
||||
re.findall(
|
||||
r"Acquisition started on : (\d+)\/(\d+)\/(\d+) (\d+):(\d+):(\d+)\.(\d+)",
|
||||
"".join(log),
|
||||
)[0],
|
||||
)
|
||||
)
|
||||
return datetime(
|
||||
int(start[2]), start[0], start[1], start[3], start[4], start[5], start[6] * 1000
|
||||
)
|
||||
|
||||
|
||||
LOG_MAGIC = "EC-Lab LOG FILEBT-Lab LOG FILE"
|
||||
LOOP_MAGIC = "VMP EXPERIMENT LOOP INDEXES"
|
||||
MPR_MAGIC = b"BIO-LOGIC MODULAR FILE\x1a".ljust(48) + b"\x00\x00\x00\x00"
|
||||
|
||||
|
||||
@@ -574,6 +654,8 @@ class MPRfile:
|
||||
self.loop_index = None
|
||||
if isinstance(file_or_path, str):
|
||||
mpr_file = open(file_or_path, "rb")
|
||||
loop_file = file_or_path[:-4] + "_LOOP.txt" # loop file for running experiment
|
||||
log_file = file_or_path[:-1] + "l" # log file for runnning experiment
|
||||
else:
|
||||
mpr_file = file_or_path
|
||||
magic = mpr_file.read(len(MPR_MAGIC))
|
||||
@@ -684,6 +766,11 @@ class MPRfile:
|
||||
raise ValueError(
|
||||
"Unrecognised version for data module: %d" % data_module["version"]
|
||||
)
|
||||
else:
|
||||
if os.path.isfile(loop_file):
|
||||
self.loop_index = loop_from_file(loop_file)
|
||||
if self.loop_index[-1] < n_data_points:
|
||||
self.loop_index = np.append(self.loop_index, n_data_points)
|
||||
|
||||
if maybe_log_module:
|
||||
(log_module,) = maybe_log_module
|
||||
@@ -727,6 +814,10 @@ class MPRfile:
|
||||
+ " End date: %s\n" % self.enddate
|
||||
+ " Timestamp: %s\n" % self.timestamp
|
||||
)
|
||||
else:
|
||||
if os.path.isfile(log_file):
|
||||
self.timestamp = timestamp_from_file(log_file)
|
||||
self.enddate = None
|
||||
|
||||
def get_flag(self, flagname):
|
||||
if flagname in self.flags_dict:
|
||||
|
||||
@@ -358,3 +358,25 @@ def test_MPR_matches_MPT_v1150(testdata_dir, basename_v1150):
|
||||
mpr = MPRfile(binpath)
|
||||
mpt, comments = MPTfile(txtpath, encoding="latin1")
|
||||
assert_MPR_matches_MPT_v2(mpr, mpt, comments)
|
||||
|
||||
|
||||
def test_loop_from_file(testdata_dir):
|
||||
"""Check if the loop_index is correctly extracted from the _LOOP.txt file
|
||||
"""
|
||||
mpr = MPRfile(os.path.join(testdata_dir, "running", "running_OCV.mpr"))
|
||||
if mpr.loop_index is None:
|
||||
raise AssertionError("No loop_index found")
|
||||
elif not len(mpr.loop_index) == 4:
|
||||
raise AssertionError("loop_index is not the right size")
|
||||
elif not (mpr.loop_index == [0, 4, 8, 11]).all():
|
||||
raise AssertionError("loop_index values are wrong")
|
||||
|
||||
|
||||
def test_timestamp_from_file(testdata_dir):
|
||||
"""Check if the loop_index is correctly extracted from the _LOOP.txt file
|
||||
"""
|
||||
mpr = MPRfile(os.path.join(testdata_dir, "running", "running_OCV.mpr"))
|
||||
if not hasattr(mpr, "timestamp"):
|
||||
raise AssertionError("No timestamp found")
|
||||
elif not mpr.timestamp.timestamp() == 1707299985.908:
|
||||
raise AssertionError("timestamp value is wrong")
|
||||
|
||||
Reference in New Issue
Block a user