[dataio/nix] moving creation of nix to nix writer
This commit is contained in:
parent
3d3d9a0cdd
commit
a53ff62d96
@ -1,4 +1,9 @@
|
|||||||
|
from datetime import datetime
|
||||||
|
import itertools
|
||||||
|
import string
|
||||||
import time
|
import time
|
||||||
|
from dataclasses import asdict
|
||||||
|
|
||||||
from IPython import embed
|
from IPython import embed
|
||||||
from PyQt6.QtCore import QMutex
|
from PyQt6.QtCore import QMutex
|
||||||
import nixio
|
import nixio
|
||||||
@ -10,8 +15,9 @@ log = config_logging()
|
|||||||
|
|
||||||
|
|
||||||
class NixWriter:
|
class NixWriter:
|
||||||
def __init__(self, buffer: CircBuffer) -> None:
|
def __init__(self, buffer: CircBuffer, config) -> None:
|
||||||
self.buffer = buffer
|
self.buffer = buffer
|
||||||
|
self.config = config
|
||||||
|
|
||||||
def write_nix(
|
def write_nix(
|
||||||
self,
|
self,
|
||||||
@ -68,5 +74,33 @@ class NixWriter:
|
|||||||
"Analog1", "ndarray", shape=(1000,), dtype=nixio.DataType.Double
|
"Analog1", "ndarray", shape=(1000,), dtype=nixio.DataType.Double
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def generate_letter_sequence(self):
|
||||||
|
alphabet = string.ascii_lowercase
|
||||||
|
for size in range(2, 3):
|
||||||
|
for combo in itertools.product(alphabet, repeat=size):
|
||||||
|
yield "".join(combo)
|
||||||
|
|
||||||
|
def create_nix_file(self, file_path, metadata):
|
||||||
|
data_time = datetime.now().strftime("%Y-%m-%t_%H-%M-%S")
|
||||||
|
sequence_generator = self.generate_letter_sequence()
|
||||||
|
sequence = next(sequence_generator)
|
||||||
|
|
||||||
|
self.nix_file = nixio.File.open(
|
||||||
|
path=f"{file_path}/{data_time}_{sequence}.nix",
|
||||||
|
mode=nixio.FileMode.Overwrite,
|
||||||
|
)
|
||||||
|
self.block = self.nix_file.create_block("recording", "testfile")
|
||||||
|
self.section = self.nix_file.create_section("metadata", "config.yaml")
|
||||||
|
for key, value in asdict(metadata).items():
|
||||||
|
self.section[key] = value
|
||||||
|
|
||||||
|
self.data_array_analog1 = self.block.create_data_array(
|
||||||
|
"Analog1", "ndarray", shape=(1000,), dtype=nixio.DataType.Double
|
||||||
|
)
|
||||||
|
self.data_array_analog2 = self.block.create_data_array(
|
||||||
|
"Analog2", "ndarray", shape=(1000,), dtype=nixio.DataType.Double
|
||||||
|
)
|
||||||
|
return self.data_array_analog1, self.data_array_analog2
|
||||||
|
|
||||||
def stop_writing(self):
|
def stop_writing(self):
|
||||||
self.write = False
|
self.write = False
|
||||||
|
Loading…
Reference in New Issue
Block a user