diff --git a/python/dlisio/__init__.py b/python/dlisio/__init__.py index 004ae04fe..903316cc7 100644 --- a/python/dlisio/__init__.py +++ b/python/dlisio/__init__.py @@ -81,6 +81,36 @@ def frames(self): """ return self._objects.frames + @property + def tools(self): + """ Read all Tool metadata objects + + Returns + ------- + tools: generator of Tool objects + """ + return self._objects.tools + + @property + def parameters(self): + """ Read all Parameter metadata objects + + Returns + ------- + parameters: generator of Parameter objects + """ + return self._objects.parameters + + @property + def calibrations(self): + """ Read all Calibration objects + + Returns + ------- + calibrations: generator of Calibration objects + """ + return self._objects.calibrations + @property def unknowns(self): return self._objects.unknowns diff --git a/python/dlisio/objects.py b/python/dlisio/objects.py index 6b719ea7f..787c23537 100644 --- a/python/dlisio/objects.py +++ b/python/dlisio/objects.py @@ -25,8 +25,11 @@ def __init__(self, objects): for os in objects: for obj in os.objects: - if os.type == "FRAME" : obj = Frame(obj) - elif os.type == "CHANNEL" : obj = Channel(obj) + if os.type == "FRAME" : obj = Frame(obj) + elif os.type == "CHANNEL" : obj = Channel(obj) + elif os.type == "TOOL" : obj = Tool(obj) + elif os.type == "PARAMETER" : obj = Parameter(obj) + elif os.type == "CALIBRATION" : obj = Calibration(obj) else: obj = Unknown(obj) self.objects.append(obj) @@ -47,14 +50,23 @@ def link(self, obj): if o.name == obj.source.name: obj.source = o if obj.type == "frame": - obj.channels = [r for r in self.channels if r.name in obj.channels] + obj.channels = [o for o in self.channels if obj.haschannel(o.name)] + + if obj.type == "tool": + obj.channels = [o for o in self.channels if obj.haschannel(o.name)] + obj.parameters = [o for o in self.parameters if obj.hasparameter(o.name)] + + if obj.type == "calibration": + obj.uncal_ch = [o for o in self.channels if obj.hasuncalibrated_channel(o.name)] + obj.cal_ch = [o for o in self.channels if obj.hascalibrated_channel(o.name)] + obj.parameters = [o for o in self.parameters if obj.hasparameter(o.name)] def getobject(self, name, type): """ return object corresponding to the unique identifier given by name + type Parameters ---------- - name : tuple or dlisio.core.obname + name : tuple(str, int, int) or dlisio.core.obname type : str Returns @@ -99,6 +111,21 @@ def frames(self): """Frame objects""" return (o for o in self.objects if o.type == "frame") + @property + def tools(self): + """Tool objects""" + return (o for o in self.objects if o.type == "tool") + + @property + def parameters(self): + """Parameter objects""" + return (o for o in self.objects if o.type == "parameter") + + @property + def calibrations(self): + """Calibration objects""" + return (o for o in self.objects if o.type == "calibration") + @property def unknowns(self): """Frame objects""" @@ -136,6 +163,53 @@ def __str__(self): s += "\t{}: {}\n".format(key, value) return s + @staticmethod + def contains(base, name): + """ Check if base cotains obj + + Parameters: + ---------- + base : list of dlis.core.obname or list of any object derived from + basic_object, e.g. Channel, Frame + + obj : dlis.core.obname, tuple (str, int, int) + + Returns + ------- + isin : bool + True if obj or (name, type) is in base, else False + + Examples + -------- + + Check if "frame" contain channel: + + >>> ans = contains(frame.channels, obj=channel.name) + + Check if "frame" contains a channel with name: + >>> name = ("TDEP", 2, 0) + >>> ans = contains(frame.channels, name) + + find all frames that have "channel": + + >>> fr = [o for o in frames if contains(o.channels, obj=channel.name)] + """ + child = None + parents = None + + if isinstance(name, core.obname): + child = (name.id, name.origin, name.copynumber) + else: + child = name + try: + parents = [(o.id, o.origin, o.copynumber) for o in base] + except AttributeError: + parents = [(o.name.id, o.name.origin, o.name.copynumber) for o in base] + + if any(child == p for p in parents): return True + + return False + class Channel(basic_object): """ The Channel object reflects the logical record type CHANNEL (listed in @@ -170,7 +244,7 @@ def hassource(self, obj): Parameters ---------- - obj : dlis.core.obname or any object class derived from basic_object + obj : dlis.core.obname or tuple(str, int, int) Returns ------- @@ -178,15 +252,7 @@ def hassource(self, obj): True if obj is the source of channel, else False """ - if self.source is None: return False - - if isinstance(obj, core.obname): child = obj - else : child = obj.name - - if isinstance(self.source, core.obname): parent = self.source - else : parent = self.source.name - - return parent == child + return self.contains(self.source, obj) class Frame(basic_object): """ @@ -223,7 +289,7 @@ def haschannel(self, channel): Parameters ---------- - channel : dlis.core.obname or Channel object + channel : dlis.core.obname or tuple(str, int, int) Returns ------- @@ -231,17 +297,170 @@ def haschannel(self, channel): True if Frame has the channel obj, else False """ - if len(self.channels) == 0: return False + return self.contains(self.channels, channel) + +class Tool(basic_object): + """ + The tool object reflects the logical record type TOOL (listed in Appendix + A.2 - Logical Record Types, described in Chapter 5.8.4 - Static and Frame + Data, TOOL objects) + """ + def __init__(self, obj): + super().__init__(obj, "tool") + self.description = None + self.trademark_name = None + self.generic_name = None + self.status = None + self.parts = [] + self.channels = [] + self.parameters = [] - if isinstance(channel, core.obname): child = channel - else : child = channel.name + for attr in obj.values(): + if attr.value is None: continue + if attr.label == "DESCRIPTION" : self.description = attr.value[0] + if attr.label == "TRADEMARK-NAME" : self.trademark_name = attr.value[0] + if attr.label == "GENERIC-NAME" : self.generic_name = attr.value[0] + if attr.label == "STATUS" : self.status = attr.value[0] + if attr.label == "PARTS" : self.parts = attr.value + if attr.label == "CHANNELS" : self.channels = attr.value + if attr.label == "PARAMETERS" : self.parameters = attr.value - for ch in self.channels: - if isinstance(ch, core.obname): - if ch == child: return True - if isinstance(ch, Channel): - if ch.name == child: return True - return False + def haschannel(self, channel): + """ + Return True if channels is in tool.channels, + else return False + + Parameters + ---------- + channel : dlis.core.obname or tuple(str, int, int) + + Returns + ------- + haschannel : bool + True if Tool has the channel obj, else False + + """ + return self.contains(self.channels, channel) + + def hasparameter(self, param): + """ + Return True if param is in tool.parameters, + else return False + + Parameters + ---------- + param : dlis.core.obname or tuple(str, int, int) + + Returns + ------- + hasparam : bool + True if Tool has the parameter obj, else False + + """ + return self.contains(self.parameters, param) + + +class Parameter(basic_object): + """ + The Parameter object reflects the logical record type PARAMETER (listed in + Appendix A.2 - Logical Record Types, described in Chapter 5.8.2 - Static + and Frame Data, PARAMETER objects) + """ + def __init__(self, obj): + super().__init__(obj, "parameter") + self.long_name = None + self.dimension = None + self.axis = None + self.zones = None + self.values = None + + for attr in obj.values(): + if attr.value is None: continue + if attr.label == "LONG-NAME" : self.long_name = attr.value[0] + if attr.label == "DIMENSION" : self.dimension = attr.value + if attr.label == "AXIS" : self.axis = attr.value + if attr.label == "ZONES" : self.zones = attr.value + +class Calibration(basic_object): + """ + The Calibration reflects the logical record type CALIBRATION (listed in + Appendix A.2 - Logical Record Types, described in Chapter 5.8.7.3 - Static and + Frame Data, CALIBRATION objects) + + The calibrated_channels and uncalibrated_channels attributes are lists of + refrences to Channel objects. + """ + def __init__(self, obj): + super().__init__(obj, "calibration") + self.method = None + self.calibrated_channel = [] + self.uncalibrated_channel = [] + self.coefficients = [] + self.parameters = [] + + for attr in obj.values(): + if attr.value is None: continue + if attr.label == "METHOD": + self.method = attr.value[0] + if attr.label == "CALIBRATED-CHANNELS": + self.calibrated_channel = attr.value + if attr.label == "UNCALIBRATED-CHANNELS": + self.uncalibrated_channel = attr.value + if attr.label == "COEFFICIENTS": + self.coefficients = attr.value + if attr.label == "PARAMETERS": + self.parameters = attr.value + + def hasuncalibrated_channel(self, channel): + """ + Return True if channels is in Calibration.uncal_ch, + else return False + + Parameters + ---------- + channel : dlis.core.obname or tuple(str, int, int) + + Returns + ------- + hasuncalchannel : bool + True if Calibration has the channel obj in uncal_ch, else False + + """ + return self.contains(self.uncalibrated_channel, channel) + + def hascalibrated_channel(self, channel): + """ + Return True if channels is in Calibration.cal_ch, + else return False + + Parameters + ---------- + channel : dlis.core.obname or tuple(str, int, int) + + Returns + ------- + hasuncalchannel : bool + True if Calibration has the channel obj in self.cal_ch, else False + + """ + return self.contains(self.calibrated_channel, channel) + + def hasparameter(self, param): + """ + Return True if parameter is in calibration.parameter, + else return False + + Parameters + ---------- + param : dlis.core.obname, tuple(str, int, int) + + Returns + ------- + hasparameter : bool + True if Calibration has the param obj, else False + + """ + return self.contains(self.parameters, param) class Unknown(basic_object): """ diff --git a/python/tests/test_core.py b/python/tests/test_core.py index 2b0742e98..dd502ed01 100644 --- a/python/tests/test_core.py +++ b/python/tests/test_core.py @@ -175,14 +175,84 @@ def test_frames(): assert frame.encrypted is None assert frame.description is None - fchannels = [ch for ch in f.channels if frame.haschannel(ch)] + fchannels = [ch for ch in f.channels if frame.haschannel(ch.name)] assert len(fchannels) == len(frame.channels) +def test_tools(): + with dlisio.open('data/206_05a-_3_DWL_DWL_WIRE_258276498.DLIS') as f: + tool = next(f.tools) + assert tool.name.id == "MSCT" + assert tool.name.origin == 2 + assert tool.name.copynumber == 0 + assert tool.type == "tool" + assert tool.description == "Mechanical Sidewall Coring Tool" + assert tool.trademark_name == "MSCT-AA" + assert tool.generic_name == "MSCT" + assert tool.status == 1 + assert len(tool.parameters) == 22 + assert len(tool.channels) == 74 + assert len(tool.parts) == 9 + + channel_matching = [ch for ch in tool.channels if ch.name.id == "UMVL_DL"] + assert len(channel_matching) == 1 + + assert len(list(f.tools)) == 2 + tools = [o for o in f.tools if o.name.id == "MSCT"] + assert len(tools) == 1 + +def test_parameters(): + with dlisio.open('data/206_05a-_3_DWL_DWL_WIRE_258276498.DLIS') as f: + param = next(f.parameters) + assert param.name.id == "FLSHSTRM" + assert param.name.origin == 2 + assert param.name.copynumber == 0 + assert param.type == "parameter" + assert param.long_name == "Flush depth-delayed streams to output at end" + assert param.dimension is None + assert param.axis is None + assert param.zones is None + assert len(list(f.parameters)) == 226 + param = [o for o in f.parameters if o.name.id == "FLSHSTRM"] + assert len(param) == 1 + +def test_calibrations(): + with dlisio.open('data/206_05a-_3_DWL_DWL_WIRE_258276498.DLIS') as f: + calibration = next(f.calibrations) + assert calibration.name.id == "CNU" + assert calibration.name.origin == 2 + assert calibration.name.copynumber == 0 + assert calibration.type == "calibration" + assert len(calibration.parameters) == 0 + assert len(calibration.coefficients) == 2 + assert calibration.method is None + assert len(list(calibration.calibrated_channel)) == 1 + assert len(list(calibration.uncalibrated_channel)) == 1 + + + ref = ("CNU", 2, 0) + cal_ch = [o for o in f.calibrations if o.hascalibrated_channel(ref)] + assert cal_ch[0] == calibration + + ref = calibration.uncal_ch[0] + uncal_ch = [o for o in f.calibrations if o.hasuncalibrated_channel(ref.name)] + assert uncal_ch[0] == calibration + +def test_contains(): + with dlisio.open('data/206_05a-_3_DWL_DWL_WIRE_258276498.DLIS') as f: + frame = f.getobject(("2000T", 2, 0), type="frame") + name = ("TDEP", 2, 4) + channel = f.getobject(name, type="channel") + + result = frame.contains(frame.channels, channel.name) + assert result == True + result = frame.contains(frame.channels, name) + assert result == True + def test_Unknown(): with dlisio.open('data/206_05a-_3_DWL_DWL_WIRE_258276498.DLIS') as f: unknown = next(f.unknowns) assert unknown.type == "unknown" - assert len(list(f.unknowns)) == 770 + assert len(list(f.unknowns)) == 515 def test_object(): with dlisio.open('data/206_05a-_3_DWL_DWL_WIRE_258276498.DLIS') as f: