awendt/pyrelacs
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[dataio/nix] adding mutex, writing data_array, and the last chunk

Browse Source
This commit is contained in:
wendtalexander 2024-10-22 10:10:09 +02:00
parent 12e82dceee
commit 4f7ebbe8c3
1 changed files with 37 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
pyrelacs/dataio/nix_writer.py
View File

@ -1,5 +1,6 @@
import time import time
from IPython import embed from IPython import embed
from PyQt6.QtCore import QMutex
import nixio import nixio
from pyrelacs.dataio.circbuffer import CircBuffer from pyrelacs.dataio.circbuffer import CircBuffer
@ -12,29 +13,53 @@ class NixWriter:
def __init__(self, buffer: CircBuffer) -> None: def __init__(self, buffer: CircBuffer) -> None:
self.buffer = buffer self.buffer = buffer
def write_nix(self, *args, **kwargs): def write_nix(
self._write_header() self,
items = 0 data_array: nixio.DataArray,
chunk = 1000 mutex: QMutex,
channel: int = 0,
chunk_size=1000,
*args,
**kwargs,
):
index = 0
log.debug("Starting the writing") log.debug("Starting the writing")
self.write = True self.write = True
while self.write: while self.write:
total_count = self.buffer.totalcount() total_count = self.buffer.totalcount(channel=channel)
if total_count - items >= chunk: if total_count - index >= chunk_size:
# log.debug(items) mutex.lock()
log.debug(index)
try: try:
data, _ = self.buffer.read(items, extend=chunk) _, data = self.buffer.read(
self.data_array.append(data) index, extend=chunk_size, channel=channel
)
if index == 0:
data_array.write_direct(data)
else:
data_array.append(data)
index += chunk_size
except IndexError as e: except IndexError as e:
time.sleep(0.001) time.sleep(0.001)
log.debug(f"{e}") log.debug(f"{e}")
items += chunk mutex.unlock()
else: else:
time.sleep(0.001) time.sleep(0.001)
continue continue
total_count = self.buffer.totalcount(channel=channel)
try:
mutex.lock()
_, data = self.buffer.read(
index, extend=total_count - index, channel=channel
)
data_array.append(data)
mutex.unlock()
index += total_count - index
except IndexError as e:
log.error(f"Could not read the last samples, {e}")
log.debug("Stoppint the writing") log.debug("Stoppint the writing")
log.debug(f"Samples written {items}") log.debug(f"Samples written {index}")
self.nix_file.close()
def _write_header(self): def _write_header(self):
self.nix_file = nixio.File.open(path="data.nix", mode=nixio.FileMode.Overwrite) self.nix_file = nixio.File.open(path="data.nix", mode=nixio.FileMode.Overwrite)