Source code for cuvis.Measurement

from typing import Union
from .FileWriteSettings import SaveArgs
import datetime
import os
import numpy as np
from pathlib import Path

from ._cuvis_il import cuvis_il
from .cuvis_aux import SDKException, SessionData, Capabilities, MeasurementFlags, SensorInfo, GPSData
from .cuvis_types import DataFormat, ProcessingMode, ReferenceType
from .cube_utils import ImageData


import cuvis.cuvis_types as internal
base_datetime = datetime.datetime(1970, 1, 1)


[docs] class Measurement(object): capture_time: datetime.datetime # read-only measurement_flags: MeasurementFlags # read-only path: str # read-only comment: str factory_calibration: datetime.datetime # read-only assembly: str # read-only integration_time: int # read-only averages: int # read-only distance: float # read-only serial_number: str # read-only product_name: str # read-only processing_mode: ProcessingMode # read-only name: str session_info: SessionData # read-only frame_id: int # read-only def __init__(self, base: Union[int, str, Path]): self._handle = None self._session = None if isinstance(base, int): self._handle = base elif isinstance(base, str) or isinstance(base, Path): base = Path(base) if not base.exists(): raise FileNotFoundError( 'Could not open Measurement. File does not exists.') _ptr = cuvis_il.new_p_int() if cuvis_il.status_ok != cuvis_il.cuvis_measurement_load(str(base), _ptr): raise SDKException() self._handle = cuvis_il.p_int_value(_ptr) else: raise ValueError( "Could not open Measurement! Unknown Input") self.refresh() pass def _refresh_metadata(self): _metaData = cuvis_il.cuvis_mesu_metadata_allocate() if cuvis_il.status_ok != cuvis_il.cuvis_measurement_get_metadata( self._handle, _metaData): raise SDKException self._capture_time = base_datetime + datetime.timedelta( milliseconds=_metaData.capture_time) self._measurement_flags = MeasurementFlags(_metaData.measurement_flags) self._path = _metaData.path self._comment = _metaData.comment try: self._factory_calibration = base_datetime + datetime.timedelta( milliseconds=_metaData.factory_calibration) except OverflowError: self._factory_calibration = None self._assembly = _metaData.assembly self._averages = _metaData.averages self._distance = _metaData.distance self._integration_time = _metaData.integration_time self._serial_number = _metaData.serial_number self._product_name = _metaData.product_name self._processing_mode = internal.__ProcessingMode__[ _metaData.processing_mode] self._name = _metaData.name self._session_info = SessionData(_metaData.session_info_name, _metaData.session_info_session_no, _metaData.session_info_sequence_no) self._frame_id = _metaData.measurement_frame_id cuvis_il.cuvis_mesu_metadata_free(_metaData)
[docs] def refresh(self) -> None: self.data = {} self._refresh_metadata() pcount = cuvis_il.new_p_int() if cuvis_il.status_ok != cuvis_il.cuvis_measurement_get_data_count( self._handle, pcount): raise SDKException() for ind in range(cuvis_il.p_int_value(pcount)): pType = cuvis_il.new_p_cuvis_data_type_t() key = cuvis_il.cuvis_measurement_get_data_info_swig(self._handle, pType, ind) cdtype = cuvis_il.p_cuvis_data_type_t_value(pType) if cdtype == cuvis_il.data_type_image: data = cuvis_il.cuvis_imbuffer_t() cuvis_il.cuvis_measurement_get_data_image(self._handle, key, data) # t0 = datetime.datetime.now() self.data.update({key: ImageData(img_buf=data, dformat=DataFormat[ data.__getattribute__( "format")])}) # print("image loading time: {}".format( # datetime.datetime.now() - t0)) elif cdtype == cuvis_il.data_type_string: val = cuvis_il.cuvis_measurement_get_data_string_swig( self._handle, key) self.data.update({key: val}) elif cdtype == cuvis_il.data_type_gps: gps = cuvis_il.cuvis_gps_t() cuvis_il.cuvis_measurement_get_data_gps(self._handle, key, gps) self.data.update({key: GPSData._from_internal(gps)}) elif cdtype == cuvis_il.data_type_sensor_info: info = cuvis_il.cuvis_sensor_info_t() cuvis_il.cuvis_measurement_get_data_sensor_info(self._handle, key, info) self.data.update({key: SensorInfo._from_internal(info)}) else: self.data.update({key: "Not Implemented!"})
[docs] def save(self, saveargs: SaveArgs) -> None: ge, sa = saveargs._get_internal() if cuvis_il.status_ok != cuvis_il.cuvis_measurement_save( self._handle, ge.export_dir, sa): raise SDKException() pass
@property def capture_time(self) -> datetime.datetime: return self._capture_time @property def measurement_flags(self) -> MeasurementFlags: return self._measurement_flags @property def path(self) -> str: return self._path @property def comment(self) -> str: return self._comment @comment.setter def comment(self, comment: str) -> None: if cuvis_il.status_ok != cuvis_il.cuvis_measurement_set_comment( self._handle, comment): raise SDKException() self._refresh_metadata() pass @property def factory_calibration(self) -> datetime.datetime: return self._factory_calibration @property def assembly(self) -> str: return self._assembly @property def integration_time(self) -> int: return self._integration_time @property def averages(self) -> int: return self._averages @property def distance(self) -> float: return self._distance @property def serial_number(self) -> str: return self._serial_number @property def product_name(self) -> str: return self._product_name @property def processing_mode(self) -> ProcessingMode: return self._processing_mode @property def name(self) -> str: return self._name @name.setter def name(self, name: str) -> None: if cuvis_il.status_ok != cuvis_il.cuvis_measurement_set_name( self._handle, name): raise SDKException() self._refresh_metadata() pass @property def session_info(self) -> SessionData: return self._session_info @property def frame_id(self) -> int: return self._frame_id @property def cube(self) -> ImageData: """ Retrieves or processes the 'cube' data for this Measurement. This property prioritizes convenience over strict design principles: - Attempts to retrieve the 'cube' from `self.data`. - Lazily initializes a `ProcessingContext` if a session is available but uninitialized. - May trigger expensive processing and modify internal state during property access. While functional, this approach introduces side effects and tight coupling, making it less predictable and not the cleanest solution. Suitable for specific workflows where these trade-offs are acceptable. Raises ------ ValueError If the 'cube' is not available and processing is not possible. Returns ------- ImageData The 'cube' data, either retrieved from `self.data` or generated through processing. """ if 'cube' in self.data: return self.data.get('cube') if self._session is not None: # try fallback if session is known if self._session._pc is None: from .ProcessingContext import ProcessingContext self._session._pc = ProcessingContext(self._session) self._session._pc.apply(self) return self.data.get('cube', None) raise ValueError( "This Measurement does not have a cube saved. Consider reprocessing with a Processing Context.") @property def thumbnail(self): thumb = [val for key, val in self.data.items() if "view" in key] if len(thumb) == 0: print("No thumbnail available. Use cube instead!") return None elif len(thumb) == 1: return thumb[0] elif len(thumb) > 1: shapes = [th.array.shape for th in thumb] return thumb[shapes.index(min(shapes))] @property def capabilities(self) -> Capabilities: _ptr = cuvis_il.new_p_int() if cuvis_il.status_ok != cuvis_il.cuvis_measurement_get_capabilities( self._handle, _ptr): raise SDKException() return Capabilities(cuvis_il.p_int_value(_ptr)) @property def calibration_id(self) -> str: _id = cuvis_il.cuvis_measurement_get_calib_id_swig(self._handle) return _id @property def data_count(self) -> int: out = cuvis_il.new_p_int() cuvis_il.cuvis_measurement_get_data_count(self._handle, out) return cuvis_il.p_int_value(out)
[docs] def clear_cube(self) -> None: if cuvis_il.status_ok != cuvis_il.cuvis_measurement_clear_cube( self._handle): raise SDKException() pass
[docs] def clear_implicit_reference(self, ref_type: ReferenceType) -> None: if cuvis_il.status_ok != \ cuvis_il.cuvis_measurement_clear_implicit_reference( self._handle, internal.__CuvisReferenceType__[ref_type]): raise SDKException()
[docs] def deepcopy(self): _ptr = cuvis_il.new_p_int() if cuvis_il.status_ok != cuvis_il.cuvis_measurement_deep_copy( self._handle, _ptr): raise SDKException() copy = Measurement(cuvis_il.p_int_value(_ptr)) return copy
def __del__(self): _ptr = cuvis_il.new_p_int() self.clear_cube() cuvis_il.p_int_assign(_ptr, self._handle) cuvis_il.cuvis_measurement_free(_ptr) self._handle = cuvis_il.p_int_value(_ptr) pass def __deepcopy__(self, memo): return self.deepcopy() def __copy__(self, memo): '''This functions is not permitted due to the class only keeping a handle, that is managed by the cuvis sdk.''' raise TypeError('Shallow copying is not supported for Measurement')