Some warnings fixed. The rest is markup suggestions from black, of which part is committed
This commit is contained in:
parent
dd3aa5a0d6
commit
832302bba2
@ -1,6 +1,24 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
from contextlib import contextmanager
|
||||
from weakref import WeakValueDictionary
|
||||
import h5py as h5
|
||||
import uuid
|
||||
import pathlib
|
||||
import glob
|
||||
import itertools
|
||||
import numpy as np
|
||||
from enum import Enum, unique
|
||||
from .lasp_config import LASP_NUMPY_FLOAT_TYPE
|
||||
from scipy.io import wavfile
|
||||
import os, time, wave, logging
|
||||
from .lasp_common import SIQtys, Qty, getFreq
|
||||
from .lasp_version import LASP_VERSION_MAJOR, LASP_VERSION_MINOR
|
||||
from .lasp_cpp import Window, DaqChannel, AvPowerSpectra
|
||||
from typing import List
|
||||
from functools import lru_cache
|
||||
from .lasp_config import ones
|
||||
"""!
|
||||
Author: J.A. de Jong - ASCEE
|
||||
|
||||
@ -46,29 +64,11 @@ The video dataset can possibly be not present in the data.
|
||||
|
||||
"""
|
||||
|
||||
__all__ = ["Measurement", "scaleBlockSens", "MeasurementType"]
|
||||
from contextlib import contextmanager
|
||||
from weakref import WeakValueDictionary
|
||||
import h5py as h5
|
||||
import uuid
|
||||
import pathlib
|
||||
import glob
|
||||
import itertools
|
||||
import numpy as np
|
||||
from enum import Enum, unique
|
||||
from .lasp_config import LASP_NUMPY_FLOAT_TYPE
|
||||
from scipy.io import wavfile
|
||||
import os, time, wave, logging
|
||||
from .lasp_common import SIQtys, Qty, getFreq
|
||||
from .lasp_version import LASP_VERSION_MAJOR, LASP_VERSION_MINOR
|
||||
from .lasp_cpp import Window, DaqChannel, AvPowerSpectra
|
||||
from typing import List
|
||||
from functools import lru_cache
|
||||
from .lasp_config import ones
|
||||
|
||||
__all__ = ["Measurement", "scaleBlockSens", "MeasurementType"]
|
||||
# Measurement file extension
|
||||
MEXT = 'h5'
|
||||
DOTMEXT = f'.{MEXT}'
|
||||
MEXT = "h5"
|
||||
DOTMEXT = f".{MEXT}"
|
||||
|
||||
|
||||
@unique
|
||||
@ -272,36 +272,36 @@ class Measurement:
|
||||
self.version_major = f.attrs["LASP_VERSION_MAJOR"]
|
||||
self.version_minor = f.attrs["LASP_VERSION_MINOR"]
|
||||
except KeyError:
|
||||
# No version information stored
|
||||
# No version information stored
|
||||
self.version_major = 0
|
||||
self.version_minor = 1
|
||||
|
||||
try:
|
||||
# Try to catch UUID (Universally Unique IDentifier)
|
||||
self._UUID = f.attrs['UUID']
|
||||
self._UUID = f.attrs["UUID"]
|
||||
# Flag indicating we have to add a new UUID
|
||||
create_new_uuid = False
|
||||
except KeyError:
|
||||
create_new_uuid = True
|
||||
|
||||
try:
|
||||
# UUID of the reference measurement. Should be stored as
|
||||
# UUID of the reference measurement. Should be stored as
|
||||
# a lists of tuples, where each tuple is a combination of (<MeasurementType.value>, <uuid_string>, <last_filename>).
|
||||
|
||||
# The last filename is a filename that *probably* is the reference measurement with
|
||||
# given UUID. If it is not, we will search for it in the same directory as `this` measurement.
|
||||
# If we cannot find it there, we will give up, and remove the field corresponding to this reference measurement type.
|
||||
refMeas_list = f.attrs['refMeas']
|
||||
refMeas_list = f.attrs["refMeas"]
|
||||
|
||||
# Build a tuple string from it
|
||||
self._refMeas = {}
|
||||
for (key, val, name) in refMeas_list:
|
||||
for key, val, name in refMeas_list:
|
||||
self._refMeas[MeasurementType(int(key))] = (val, name)
|
||||
except KeyError:
|
||||
self._refMeas = {}
|
||||
|
||||
try:
|
||||
self._type_int = f.attrs['type_int']
|
||||
self._type_int = f.attrs["type_int"]
|
||||
except KeyError:
|
||||
self._type_int = 0
|
||||
|
||||
@ -368,7 +368,9 @@ class Measurement:
|
||||
self.genNewUUID()
|
||||
else:
|
||||
if self.UUID in Measurement.uuid_s.keys():
|
||||
raise RuntimeError(f"Measurement '{self.name}' is already opened. Cannot open measurement twice. Note: this error can happen when measurements are manually copied.")
|
||||
raise RuntimeError(
|
||||
f"Measurement '{self.name}' is already opened. Cannot open measurement twice. Note: this error can happen when measurements are manually copied."
|
||||
)
|
||||
|
||||
# Store weak reference to 'self' in list of UUID's. They are removed when no file is open anymore
|
||||
Measurement.uuid_s[self._UUID] = self
|
||||
@ -380,11 +382,11 @@ class Measurement:
|
||||
Args:
|
||||
newname: New name, with or without extension
|
||||
"""
|
||||
_ , ext = os.path.splitext(newname)
|
||||
_, ext = os.path.splitext(newname)
|
||||
# Add proper extension if new name is given without extension.
|
||||
if ext != DOTMEXT:
|
||||
newname = newname + DOTMEXT
|
||||
|
||||
|
||||
# Folder, Base filename + extension
|
||||
folder, _ = os.path.split(self.fn)
|
||||
|
||||
@ -397,7 +399,7 @@ class Measurement:
|
||||
"""
|
||||
Create new UUID for measurement and store in file.
|
||||
"""
|
||||
self.setAttribute('UUID', str(uuid.uuid1()))
|
||||
self.setAttribute("UUID", str(uuid.uuid1()))
|
||||
|
||||
@property
|
||||
def UUID(self):
|
||||
@ -405,59 +407,73 @@ class Measurement:
|
||||
Universally unique identifier
|
||||
"""
|
||||
return self._UUID
|
||||
|
||||
|
||||
def getRefMeas(self, mtype: MeasurementType):
|
||||
"""
|
||||
Return corresponding reference measurement, if configured and can be found. If the reference
|
||||
Return corresponding reference measurement, if configured and can be found. If the reference
|
||||
measurement is currently not open, it tries to open it by traversing other measurement
|
||||
files in the current directory. Throws a runtime error in case the reference measurement cannot be found.
|
||||
Throws a ValueError when the reference measurement is not configured.
|
||||
"""
|
||||
|
||||
|
||||
# See if we can find the UUID for the required measurement type
|
||||
try:
|
||||
required_uuid, possible_name = self._refMeas[mtype]
|
||||
except KeyError:
|
||||
raise ValueError(f"No reference measurement configured for '{self.name}'")
|
||||
|
||||
|
||||
m = None
|
||||
# Try to find it in the dictionary of of open measurements
|
||||
if required_uuid in Measurement.uuid_s.keys():
|
||||
m = Measurement.uuid_s[required_uuid]
|
||||
logging.debug(f'Returned reference measurement {m.name} from list of open measurements')
|
||||
|
||||
logging.debug(
|
||||
f"Returned reference measurement {m.name} from list of open measurements"
|
||||
)
|
||||
|
||||
# Not found in list of openend measurements. See if we can open it using its last stored file name we know of
|
||||
if m is None:
|
||||
try:
|
||||
m = Measurement(possible_name)
|
||||
if m.UUID == required_uuid:
|
||||
logging.info(f'Opened reference measurement {m.name} by name')
|
||||
logging.info(f"Opened reference measurement {m.name} by name")
|
||||
except Exception as e:
|
||||
logging.error(f'Could not find reference measurement using file name: {possible_name}')
|
||||
logging.error(
|
||||
f"Could not find reference measurement using file name: {possible_name}"
|
||||
)
|
||||
|
||||
# Last resort, see if we can find the right measurement in the same folder
|
||||
if m is None:
|
||||
try:
|
||||
folder, _ = os.path.split(self.fn)
|
||||
m = Measurement.fromFolderWithUUID(required_uuid, folder, skip=[self.name])
|
||||
logging.info('Found reference measurement in folder with correct UUID. Updating name of reference measurement')
|
||||
m = Measurement.fromFolderWithUUID(
|
||||
required_uuid, folder, skip=[self.name]
|
||||
)
|
||||
logging.info(
|
||||
"Found reference measurement in folder with correct UUID. Updating name of reference measurement"
|
||||
)
|
||||
# Update the measurement file name in the list, such that next time it
|
||||
# can be opened just by its name.
|
||||
self.setRefMeas(m)
|
||||
except:
|
||||
logging.error("Could not find the reference measurement. Is it deleted?")
|
||||
logging.error(
|
||||
"Could not find the reference measurement. Is it deleted?"
|
||||
)
|
||||
|
||||
# Well, we found it. Now make sure the reference measurement actually has the right type (User could have marked it as a NotSpecific for example in the mean time).
|
||||
if m is not None:
|
||||
if m.measurementType() != mtype:
|
||||
m.removeRefMeas(mtype)
|
||||
raise RuntimeError(f"Reference measurement for {self.name} is not a proper reference (anymore).")
|
||||
|
||||
raise RuntimeError(
|
||||
f"Reference measurement for {self.name} is not a proper reference (anymore)."
|
||||
)
|
||||
|
||||
# Whow, we passed all security checks, here we go!
|
||||
return m
|
||||
else:
|
||||
# Nope, not there.
|
||||
raise RuntimeError(f"Could not find the reference measurement for '{self.name}'. Is it deleted?")
|
||||
raise RuntimeError(
|
||||
f"Could not find the reference measurement for '{self.name}'. Is it deleted?"
|
||||
)
|
||||
|
||||
def removeRefMeas(self, mtype: MeasurementType):
|
||||
"""
|
||||
@ -478,9 +494,12 @@ class Measurement:
|
||||
with self.file("r+") as f:
|
||||
# Update attribute in file. Stored as a flat list of string tuples:
|
||||
# [(ref_value1, uuid_1, name_1), (ref_value2, uuid_2, name_2), ...]
|
||||
reflist = list((str(key.value), val1, val2) for key, (val1, val2) in self._refMeas.items())
|
||||
reflist = list(
|
||||
(str(key.value), val1, val2)
|
||||
for key, (val1, val2) in self._refMeas.items()
|
||||
)
|
||||
# print(reflist)
|
||||
f.attrs['refMeas'] = reflist
|
||||
f.attrs["refMeas"] = reflist
|
||||
|
||||
def setRefMeas(self, m: Measurement):
|
||||
"""
|
||||
@ -490,32 +509,36 @@ class Measurement:
|
||||
"""
|
||||
mtype = m.measurementType()
|
||||
if mtype == MeasurementType.NotSpecific:
|
||||
raise ValueError('Measurement to be set as reference is not a reference measurement')
|
||||
|
||||
raise ValueError(
|
||||
"Measurement to be set as reference is not a reference measurement"
|
||||
)
|
||||
|
||||
self._refMeas[mtype] = (m.UUID, m.name)
|
||||
self.__storeReafMeas()
|
||||
|
||||
@staticmethod
|
||||
def fromFolderWithUUID(uuid_str: str, folder: str='', skip=[]):
|
||||
def fromFolderWithUUID(uuid_str: str, folder: str = "", skip=[]):
|
||||
"""
|
||||
Returns Measurement object from a given UUID string. It first tries to find whether there
|
||||
is an uuid in the static list of weak references. If not, it will try to open files in
|
||||
is an uuid in the static list of weak references. If not, it will try to open files in
|
||||
the current file path.
|
||||
"""
|
||||
for fn in glob.glob(str(pathlib.Path(folder)) + f'/*{DOTMEXT}'):
|
||||
for fn in glob.glob(str(pathlib.Path(folder)) + f"/*{DOTMEXT}"):
|
||||
# Do not try to open this file in case it is in the 'skip' list.
|
||||
if len(list(filter(lambda a: a in fn, skip))) > 0:
|
||||
continue
|
||||
|
||||
|
||||
try:
|
||||
m = Measurement(fn)
|
||||
if m.UUID == uuid_str:
|
||||
if m.UUID == uuid_str:
|
||||
# Update 'last_fn' attribute in dict of stored reference measurements
|
||||
return m
|
||||
except Exception as e:
|
||||
logging.error(f'Possible measurement file {fn} returned error {e} when opening.')
|
||||
logging.error(
|
||||
f"Possible measurement file {fn} returned error {e} when opening."
|
||||
)
|
||||
|
||||
raise RuntimeError(f'Measurement with UUID {uuid_str} could not be found.')
|
||||
raise RuntimeError(f"Measurement with UUID {uuid_str} could not be found.")
|
||||
|
||||
def setAttribute(self, attrname: str, value):
|
||||
"""
|
||||
@ -535,17 +558,17 @@ class Measurement:
|
||||
"""
|
||||
Returns True when a measurement is flagged as being of a certaint "MeasurementType"
|
||||
"""
|
||||
if (type_.value & self._type_int):
|
||||
if type_.value & self._type_int:
|
||||
return True
|
||||
elif type_.value == self._type_int == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def setType(self, type_: MeasurementType):
|
||||
"""
|
||||
Set the measurement type to given type
|
||||
"""
|
||||
self.setAttribute('type_int', type_.value)
|
||||
self.setAttribute("type_int", type_.value)
|
||||
|
||||
def measurementType(self):
|
||||
"""
|
||||
@ -590,7 +613,7 @@ class Measurement:
|
||||
@channelConfig.setter
|
||||
def channelConfig(self, chcfg: List[DaqChannel]):
|
||||
"""
|
||||
Set new channel configuration from list of DaqChannel objects.
|
||||
Set new channel configuration from list of DaqChannel objects.
|
||||
|
||||
Use cases:
|
||||
- Update channel types, sensitivities etc.
|
||||
@ -999,9 +1022,9 @@ class Measurement:
|
||||
happen that a Measurement object is created twice for the same backing file, which we do not allow.
|
||||
"""
|
||||
# See if the base part of the filename is referring to a file that is already open
|
||||
with h5.File(fn, 'r') as f:
|
||||
with h5.File(fn, "r") as f:
|
||||
try:
|
||||
theuuid = f.attrs['UUID']
|
||||
theuuid = f.attrs["UUID"]
|
||||
except KeyError:
|
||||
# No UUID stored in measurement. This is an old measurement that did not have UUID's
|
||||
# We create a new UUID here such that the file is opened from the filesystem
|
||||
@ -1010,7 +1033,7 @@ class Measurement:
|
||||
|
||||
if theuuid in Measurement.uuid_s.keys():
|
||||
return Measurement.uuid_s[theuuid]
|
||||
|
||||
|
||||
return Measurement(fn)
|
||||
|
||||
@staticmethod
|
||||
@ -1091,8 +1114,8 @@ class Measurement:
|
||||
sensitivity,
|
||||
mfn,
|
||||
timestamp=None,
|
||||
qtys: List[SIQtys]=None,
|
||||
channelNames: List[str]=None,
|
||||
qtys: List[SIQtys] = None,
|
||||
channelNames: List[str] = None,
|
||||
force=False,
|
||||
) -> Measurement:
|
||||
"""
|
||||
@ -1162,7 +1185,7 @@ class Measurement:
|
||||
hf.attrs["nchannels"] = nchannels
|
||||
|
||||
# Add physical quantity indices
|
||||
hf.attrs['qtys_enum_idx'] = [qty.toInt() for qty in qtys]
|
||||
hf.attrs["qtys_enum_idx"] = [qty.toInt() for qty in qtys]
|
||||
|
||||
# Add channel names in case given
|
||||
if channelNames is not None:
|
||||
@ -1218,4 +1241,3 @@ class Measurement:
|
||||
ad[0] = data
|
||||
|
||||
return Measurement(newfn)
|
||||
|
||||
|
||||
@ -79,7 +79,7 @@ class MeasurementSet(list):
|
||||
mtype = m.measurementType()
|
||||
if mtype == MeasurementType.NotSpecific:
|
||||
continue
|
||||
if not mtype in newest:
|
||||
if mtype not in newest:
|
||||
newest[mtype] = m
|
||||
else:
|
||||
if m.time > newest[mtype].time:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user