mirror of
https://github.com/echemdata/galvani.git
synced 2025-12-15 01:25:36 +00:00
Merge branch 'JhonFlash3008-loop-from-file'
Closes https://codeberg.org/echemdata/galvani/pulls/124 See also:8a9d4752228a9d475222
This commit is contained in:
@@ -10,6 +10,7 @@ __all__ = ["MPTfileCSV", "MPTfile"]
|
|||||||
import re
|
import re
|
||||||
import csv
|
import csv
|
||||||
from os import SEEK_SET
|
from os import SEEK_SET
|
||||||
|
import os.path
|
||||||
import time
|
import time
|
||||||
from datetime import date, datetime, timedelta
|
from datetime import date, datetime, timedelta
|
||||||
from collections import defaultdict, OrderedDict
|
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)
|
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"
|
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
|
self.loop_index = None
|
||||||
if isinstance(file_or_path, str):
|
if isinstance(file_or_path, str):
|
||||||
mpr_file = open(file_or_path, "rb")
|
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:
|
else:
|
||||||
mpr_file = file_or_path
|
mpr_file = file_or_path
|
||||||
magic = mpr_file.read(len(MPR_MAGIC))
|
magic = mpr_file.read(len(MPR_MAGIC))
|
||||||
@@ -684,6 +766,11 @@ class MPRfile:
|
|||||||
raise ValueError(
|
raise ValueError(
|
||||||
"Unrecognised version for data module: %d" % data_module["version"]
|
"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:
|
if maybe_log_module:
|
||||||
(log_module,) = maybe_log_module
|
(log_module,) = maybe_log_module
|
||||||
@@ -727,6 +814,10 @@ class MPRfile:
|
|||||||
+ " End date: %s\n" % self.enddate
|
+ " End date: %s\n" % self.enddate
|
||||||
+ " Timestamp: %s\n" % self.timestamp
|
+ " 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):
|
def get_flag(self, flagname):
|
||||||
if flagname in self.flags_dict:
|
if flagname in self.flags_dict:
|
||||||
|
|||||||
@@ -358,3 +358,22 @@ def test_MPR_matches_MPT_v1150(testdata_dir, basename_v1150):
|
|||||||
mpr = MPRfile(binpath)
|
mpr = MPRfile(binpath)
|
||||||
mpt, comments = MPTfile(txtpath, encoding="latin1")
|
mpt, comments = MPTfile(txtpath, encoding="latin1")
|
||||||
assert_MPR_matches_MPT_v2(mpr, mpt, comments)
|
assert_MPR_matches_MPT_v2(mpr, mpt, comments)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skip(reason="Test data file is missing")
|
||||||
|
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"))
|
||||||
|
assert mpr.loop_index is not None, "No loop_index found"
|
||||||
|
assert len(mpr.loop_index) == 4, "loop_index is not the right size"
|
||||||
|
assert_array_equal(mpr.loop_index, [0, 4, 8, 11], "loop_index values are wrong")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skip(reason="Test data file is missing")
|
||||||
|
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"))
|
||||||
|
assert hasattr(mpr, "timestamp"), "No timestamp found"
|
||||||
|
assert mpr.timestamp.timestamp() == pytest.approx(1707299985.908), "timestamp value is wrong"
|
||||||
|
|||||||
Reference in New Issue
Block a user