From 150954681139b5e8d92fa55ae51ad50c0148cf77 Mon Sep 17 00:00:00 2001 From: Ross Beyer Date: Tue, 20 Feb 2024 20:30:44 -0800 Subject: [PATCH] feat(pid.py and image_records.py): Changed compression handling and added icer_byte_quota and icer_minloss. --- CHANGELOG.rst | 6 ++ src/vipersci/pds/pid.py | 48 ++++++++-- src/vipersci/vis/db/image_records.py | 136 ++++++++++++++------------- tests/test_create_image.py | 76 ++++++++++++++- tests/test_create_pano_product.py | 15 +-- tests/test_create_raw.py | 5 +- tests/test_image_records.py | 64 ++++++++++++- tests/test_pid.py | 65 ++++++++++++- 8 files changed, 327 insertions(+), 88 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index e11ea4d..279f8cf 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -37,6 +37,12 @@ Changed - create_mmgis_pano.py - create() now takes a thumbsize int or tuple of ints that will control the creation and size of an output thumbnail JPG file, with naming convention set by Yamcs/OpenMCT. +- image_records.py - Added icer_byte_quota and icer_minloss parameters to ImageRecord, + as well as a variety of improved handling related to this change and in pid.py. +- pid.py - Changed handling of the compression value to match the kinds of data we'll + get from telemetry, so that the letters are assigned to cover an interval of possible + compression ratios, and to reflect the new default byte quota value for driving + images. 0.7.0 (2023-02-05) ------------------ diff --git a/src/vipersci/pds/pid.py b/src/vipersci/pds/pid.py index 4dcc809..0121ec2 100755 --- a/src/vipersci/pds/pid.py +++ b/src/vipersci/pds/pid.py @@ -1,7 +1,7 @@ #!/usr/bin/env python """This module contains classes for VIPER Product IDs.""" -# Copyright 2022-2023, United States Government as represented by the +# Copyright 2022-2024, United States Government as represented by the # Administrator of the National Aeronautics and Space Administration. # All rights reserved. # @@ -73,7 +73,7 @@ a=1, # 1:1 Lossless compression b=5, # 5:1 compression c=16, # 16:1 compression - d=64, # 64:1 compression + d=24, # 24:1 compression s="SLoG", # SLoG compression z=None, # Uncompressed ) @@ -319,15 +319,8 @@ def __init__(self, *args): instrument = self.instrument_name(instrument) - if compression in vis_compression: - pass - elif compression in vis_compression.values(): - compression = get_key(compression, vis_compression) - else: - raise ValueError(f"{args[3]} is not one of {vis_compression.keys()}") - super().__init__(date, time, instrument) - self.compression = compression + self.compression = self.compression_letter(compression) def __str__(self): return "-".join((super().__str__(), self.compression)) @@ -380,6 +373,41 @@ def compression_class(self): else: return "Lossy" + @staticmethod + def compression_letter(compression): + """Returns the letter code from the pid.vis_compression dictionary that matches + the value provided via *compression*. + """ + if compression in vis_compression: + return compression + elif compression in vis_compression.values(): + return get_key(compression, vis_compression) + elif isinstance(compression, (int, float)): + compression_ratios = [] + for v in vis_compression.values(): + if isinstance(v, (int, float)): + compression_ratios.append(v) + + if len(compression_ratios) == 0: + raise ValueError( + "There are no numeric values in vis_compression " + f"({vis_compression})." + ) + + for r in sorted(compression_ratios, reverse=True): + if compression >= r: + return get_key(r, vis_compression) + else: + raise ValueError( + f"The numeric value of {compression} is not greater than one " + f"of {compression_ratios}" + ) + else: + raise ValueError( + f"Could not determine one of {vis_compression.keys()} from " + f"compression ({compression})." + ) + @staticmethod def best_compression(identifiers: Iterable): """ diff --git a/src/vipersci/vis/db/image_records.py b/src/vipersci/vis/db/image_records.py index c768c73..9527d3a 100644 --- a/src/vipersci/vis/db/image_records.py +++ b/src/vipersci/vis/db/image_records.py @@ -3,7 +3,7 @@ """Defines the VIS image_records table using the SQLAlchemy ORM.""" -# Copyright 2022-2023, United States Government as represented by the +# Copyright 2022-2024, United States Government as represented by the # Administrator of the National Aeronautics and Space Administration. # All rights reserved. # @@ -44,7 +44,7 @@ from sqlalchemy.orm import mapped_column, relationship, synonym, validates from vipersci.pds import Purpose -from vipersci.pds.pid import VISID, vis_instruments, vis_compression +from vipersci.pds.pid import VISID, vis_instruments from vipersci.pds.xml import find_text, ns from vipersci.pds.datetime import fromisozformat, isozformat from vipersci.vis.header import pga_gain as header_pga_gain @@ -67,15 +67,6 @@ class ImageType(enum.Flag): def _missing_(cls, value): return None - def compression_ratio(self): - # This mapping reflects the current settings in RFSW - if self == ImageType.LOSSLESS_ICER_IMAGE or self == ImageType.SLOG_ICER_IMAGE: - return 1 - if self == ImageType.LOSSY_ICER_IMAGE: - return 16 - else: - return None - class ProcessingStage(enum.Flag): # PROCESS_RESERVED = 1 @@ -155,6 +146,14 @@ class ImageRecord(Base): doc="The absolute path (POSIX style) that contains the Array_2D_Image " "that this metadata refers to.", ) + icer_byte_quota = mapped_column( + Integer, + doc="The byteQuota value during onboard ICER compression. In the returned " + "Yamcs info, the value is in kilobytes, but this value is in bytes.", + ) + icer_minloss = mapped_column( + Integer, doc="The minLoss value during onboard ICER compression." + ) image_id = mapped_column( Integer, nullable=False, @@ -366,6 +365,15 @@ def __init__(self, **kwargs): else: pid = False + # Adjust the byteQuota value + if "byteQuota" in kwargs and "icer_byte_quota" not in kwargs: + kwargs["icer_byte_quota"] = int(kwargs["byteQuota"]) * 1000 + del kwargs["byteQuota"] + + if "minLoss" in kwargs and "icer_minloss" not in kwargs: + kwargs["icer_minloss"] = int(kwargs["minLoss"]) + del kwargs["minLoss"] + rpargs = dict() otherargs = dict() for k, v in kwargs.items(): @@ -438,6 +446,7 @@ def __init__(self, **kwargs): # Ensure product_id consistency if pid: + # Check datetimes if "lobt" in kwargs: if pid.datetime() != lobt_dt: raise ValueError( @@ -451,6 +460,7 @@ def __init__(self, **kwargs): f"provided start_time ({kwargs['start_time']}) disagree." ) + # Check instrument if ( self.instrument_name is not None and vis_instruments[pid.instrument] != self.instrument_name @@ -461,69 +471,59 @@ def __init__(self, **kwargs): f"({self.instrument_name}) disagree." ) + # Check compression letter if self.output_image_mask is None: - if self.processing_info is not None: - ps = ProcessingStage(self.processing_info) - if ProcessingStage.SLOG in ps and pid.compression != "s": + if self.yamcs_name is not None: + if "slog" in self.yamcs_name and pid.compression != "s": raise ValueError( f"The product_id compression code ({pid.compression}) is " - "not s, but processing_info indicates it should be " - f"({self.processing_info}). " - ) - elif ProcessingStage.SLOG not in ps and pid.compression == "s": - raise ValueError( - "The product_id compression code is s, but " - "processing_info indicates it shouldn't be " - f"({self.processing_info}). " + "not s, but yamcs_name indicates it should be " + f"({self.yamcs_name}). " ) else: t = ImageType(self.output_image_mask) if ImageType.SLOG_ICER_IMAGE == t and pid.compression == "s": pass - elif t.compression_ratio() != vis_compression[pid.compression]: + elif ( + VISID.compression_letter(compression_ratio(self.icer_byte_quota)) + != pid.compression + ): raise ValueError( f"The product_id compression code ({pid.compression}) and " - f"the compression ratio ({t.compression_ratio()}) based on " - f"the output_image_mask ({self.output_image_mask}) disagree." + "the compression ratio " + f"({compression_ratio(self.icer_byte_quota)}) based on " + f"the icer_byte_quota ({self.icer_byte_quota}) disagree." ) - elif ( - self.start_time is not None - and self.instrument_name is not None - and self.output_image_mask is not None - ): - try: - t = ImageType(self.output_image_mask) - c = None - if ( - self.processing_info == ProcessingStage.SLOG - or t == ImageType.SLOG_ICER_IMAGE - ): - c = "s" - else: - c = t.compression_ratio() - pid = VISID( - self.start_time.date(), - self.start_time.time(), - self.instrument_name, - c, + elif self.start_time is not None and self.instrument_name is not None: + c = None + if self.output_image_mask is not None: + try: + if ImageType(self.output_image_mask) == ImageType.SLOG_ICER_IMAGE: + c = "s" + except ValueError: + # output_image_mask has bad value + pass + + if c is None and self.yamcs_name is not None and "slog" in self.yamcs_name: + c = "s" + + if c is None and self.icer_byte_quota is not None: + c = compression_ratio(self.icer_byte_quota) + + if c is None: + raise ValueError( + "Could not determine the compression information " + f"from output_image_mask ({self.output_image_mask}), " + f"processing_info ({self.processing_info}), or " + f"icer_byte_quota ({self.icer_byte_quota})." ) - except ValueError as err: - # output_image_mask has bad value, last try - if self.yamcs_name is None: - raise err - else: - if "slog" in self.yamcs_name: - pid = VISID( - self.start_time.date(), - self.start_time.time(), - self.instrument_name, - "s", - ) - else: - raise ValueError( - "Could not determine the compression information " - f"from output_image_mask ({self.output_image_mask})." - ) + + pid = VISID( + self.start_time.date(), + self.start_time.time(), + self.instrument_name, + c, + ) else: got = dict() for k in ( @@ -531,6 +531,8 @@ def __init__(self, **kwargs): "start_time", "instrument_name", "output_image_mask", + "processing_info", + "icer_byte_quota", ): v = getattr(self, k) if v is not None: @@ -538,7 +540,8 @@ def __init__(self, **kwargs): raise ValueError( "Either product_id must be given, or each of start_time, " - f"instrument_name, and output_image_mask. Got: {got}" + f"instrument_name, and output_image_mask plus some other things." + f"Got: {got}" ) self._pid = str(pid) @@ -737,3 +740,10 @@ def update(self, other): setattr(self, k, v) else: self.labelmeta[k] = v + + +def compression_ratio(byte_quota): + """Returns the result of dividing the number of bytes in a grayscale image + (2048 * 2048 * 2 == 8,388,608) by the byte_quota of the returned image. + """ + return (2048 * 2048 * 2) / byte_quota diff --git a/tests/test_create_image.py b/tests/test_create_image.py index 86dfc9b..7386202 100644 --- a/tests/test_create_image.py +++ b/tests/test_create_image.py @@ -11,18 +11,19 @@ from datetime import datetime, timezone from pathlib import Path import unittest -from unittest.mock import create_autospec, patch +from unittest.mock import create_autospec, Mock, mock_open, patch import numpy as np import numpy.testing as npt from PIL import Image +from sqlalchemy.orm import Session from vipersci.pds import pid as pds from vipersci.vis.db.image_records import ImageRecord from vipersci.vis import create_image as ci -class TestParser(unittest.TestCase): +class TestCLI(unittest.TestCase): def test_arg_parser(self): p = ci.arg_parser() self.assertIsInstance(p, ArgumentParser) @@ -31,6 +32,61 @@ def test_arg_parser(self): self.assertIn("output_dir", d) self.assertIn("input", d) + def test_main(self): + pa_ret_val = ci.arg_parser().parse_args( + [ + "input.json", + ] + ) + with patch("vipersci.vis.create_image.arg_parser") as parser, patch( + "vipersci.vis.create_image.create" + ) as m_create, patch( + "vipersci.vis.create_image.open", mock_open(read_data='{"json": "dummy"}') + ): + parser.return_value.parse_args.return_value = pa_ret_val + ci.main() + m_create.assert_called_once() + + def test_main_db(self): + pa2_ret_val = ci.arg_parser().parse_args( + [ + "--dburl", + "db://foo:username@host/db", + "--tiff", + "dummy.tif", + "dummy.json", + ] + ) + session_engine_mock = create_autospec(Session) + session_mock = create_autospec(Session) + session_engine_mock.__enter__ = Mock(return_value=session_mock) + with patch("vipersci.vis.create_image.arg_parser") as parser, patch( + "vipersci.vis.create_image.create" + ) as m_create, patch("vipersci.vis.create_image.create_engine"), patch( + "vipersci.vis.create_image.Session", return_value=session_engine_mock + ), patch( + "vipersci.vis.create_image.open", mock_open(read_data='{"json": "dummy"}') + ): + parser.return_value.parse_args.return_value = pa2_ret_val + ci.main() + m_create.assert_called_once() + session_engine_mock.__enter__.assert_called_once() + session_mock.commit.assert_called_once() + + def test_main_image(self): + + pa_ret_val = ci.arg_parser().parse_args(["--image", "dummy.png", "input.json"]) + with patch("vipersci.vis.create_image.arg_parser") as parser, patch( + "vipersci.vis.create_image.create" + ) as m_create, patch( + "vipersci.vis.create_image.open", mock_open(read_data='{"json": "dummy"}') + ), patch( + "vipersci.vis.create_image.imread" + ): + parser.return_value.parse_args.return_value = pa_ret_val + ci.main() + m_create.assert_called_once() + class TestBitDepth(unittest.TestCase): def test_check_bit_depth(self): @@ -48,12 +104,28 @@ def test_check_bit_depth(self): self.assertRaises(ValueError, ci.check_bit_depth, pids, 16) +class TestCreate(unittest.TestCase): + def test_create(self): + d = {"meta": "data"} + with patch("vipersci.vis.create_image.make_image_record") as m_mir: + ci.create(d, json=False) + m_mir.assert_called_once_with(d, None, Path.cwd(), "tif") + + with patch("vipersci.vis.create_image.make_image_record") as m_mir, patch( + "vipersci.vis.create_image.write_json" + ) as m_wj: + session_mock = create_autospec(Session) + ci.create(d, session=session_mock) + m_wj.assert_called_once() + + class TestMakeImage(unittest.TestCase): def setUp(self) -> None: self.startUTC = datetime(2022, 1, 27, 0, 0, 0, tzinfo=timezone.utc) self.d = { "adcGain": 0, "autoExposure": 0, + "byteQuota": 1677, "cameraId": 0, "captureId": 1, "exposureTime": 111, diff --git a/tests/test_create_pano_product.py b/tests/test_create_pano_product.py index 30f4038..0b3e0cb 100644 --- a/tests/test_create_pano_product.py +++ b/tests/test_create_pano_product.py @@ -70,6 +70,7 @@ def setUp(self) -> None: ir1 = ImageRecord( adc_gain=0, auto_exposure=0, + byteQuota=341, cameraId=0, capture_id=1, exposure_duration=511, @@ -77,7 +78,7 @@ def setUp(self) -> None: file_creation_datetime="2023-11-27T22:18:11.879458Z", file_data_type="UnsignedLSB2", file_md5_checksum="cd30229a7803ec35fbf21a3da254ad10", - file_path="231109-170000-ncl-c.tif", + file_path="231109-170000-ncl-d.tif", imageDepth=2, image_id=0, immediateDownloadInfo=24, @@ -90,7 +91,7 @@ def setUp(self) -> None: padding=0, pga_gain=1.0, processing_info=10, - product_id="231109-170000-ncl-c", + product_id="231109-170000-ncl-d", samples=2048, software_name="vipersci", software_program_name="vipersci.vis.create_image", @@ -106,6 +107,7 @@ def setUp(self) -> None: ir2 = ImageRecord( adc_gain=0, auto_exposure=0, + byteQuota=341, cameraId=0, capture_id=1, exposure_duration=511, @@ -113,7 +115,7 @@ def setUp(self) -> None: file_creation_datetime="2023-11-27T22:18:12.948617Z", file_data_type="UnsignedLSB2", file_md5_checksum="781d510b1f7f7a4048f7a9eea596b9e6", - file_path="231109-170100-ncl-c.tif", + file_path="231109-170100-ncl-d.tif", imageDepth=2, image_id=0, immediateDownloadInfo=24, @@ -126,7 +128,7 @@ def setUp(self) -> None: padding=0, pga_gain=1.0, processing_info=10, - product_id="231109-170100-ncl-c", + product_id="231109-170100-ncl-d", samples=2048, software_name="vipersci", software_program_name="vipersci.vis.create_image", @@ -142,6 +144,7 @@ def setUp(self) -> None: ir3 = ImageRecord( adc_gain=0, auto_exposure=0, + byteQuota=341, cameraId=0, capture_id=1, exposure_duration=511, @@ -149,7 +152,7 @@ def setUp(self) -> None: file_creation_datetime="2023-11-27T22:18:14.068725Z", file_data_type="UnsignedLSB2", file_md5_checksum="ee7a86d7ed7a436d23dd6b87ae6c3058", - file_path="231109-170200-ncl-c.tif", + file_path="231109-170200-ncl-d.tif", imageDepth=2, image_id=0, immediateDownloadInfo=24, @@ -162,7 +165,7 @@ def setUp(self) -> None: padding=0, pga_gain=1.0, processing_info=10, - product_id="231109-170200-ncl-c", + product_id="231109-170200-ncl-d", samples=2048, software_name="vipersci", software_program_name="vipersci.vis.create_image", diff --git a/tests/test_create_raw.py b/tests/test_create_raw.py index 7e3e40b..14e11c4 100644 --- a/tests/test_create_raw.py +++ b/tests/test_create_raw.py @@ -47,6 +47,7 @@ def setUp(self) -> None: self.ir = ImageRecord( adc_gain=0, auto_exposure=0, + byteQuota=341, cameraId=0, capture_id=1, exposure_duration=111, @@ -54,7 +55,7 @@ def setUp(self) -> None: file_creation_datetime="2023-08-01T00:51:51.148919Z", file_data_type="UnsignedLSB2", file_md5_checksum="b8f1a035e39c223e2b7e236846102c29", - file_path="231125-140416-ncl-c.tif", + file_path="231125-140416-ncl-d.tif", imageDepth=2, image_id=0, immediateDownloadInfo=10, @@ -68,7 +69,7 @@ def setUp(self) -> None: padding=0, pga_gain=1.0, processing_info=10, - product_id="231125-140416-ncl-c", + product_id="231125-140416-ncl-d", samples=2048, software_name="vipersci", software_program_name="vipersci.vis.create_image", diff --git a/tests/test_image_records.py b/tests/test_image_records.py index 1e2c2ce..e576536 100644 --- a/tests/test_image_records.py +++ b/tests/test_image_records.py @@ -50,9 +50,6 @@ def test_single_flags(self): def test_not_member(self): self.assertRaises(ValueError, trp.ImageType, 1000) - def test_ratio(self): - self.assertEqual(trp.ImageType(1).compression_ratio(), 1) - class TestProcessingStage(unittest.TestCase): def test_init(self): @@ -85,17 +82,19 @@ def setUp(self): hazlight_center_starboard_on=False, hazlight_fore_port_on=False, hazlight_fore_starboard_on=False, + icer_byte_quota=493448, image_id=0, instrument_name="NavCam Left", instrument_temperature=128, lines=2048, lobt=self.startUTC.timestamp(), md5_checksum="dummychecksum", + minLoss=0, mission_phase="Test", navlight_left_on=False, navlight_right_on=False, offset=16324, - onboard_compression_ratio=5, + # onboard_compression_ratio=5, onboard_compression_type="ICER", output_image_mask=8, output_image_type="?", @@ -119,6 +118,12 @@ def test_init(self): rpl = trp.ImageRecord(**d) self.assertEqual("220127-000000-ncl-c", str(rpl.product_id)) + d = self.d.copy() + d["capture_id"] = 65537 + ir_ci = trp.ImageRecord(**d) + self.assertEqual(1, ir_ci.image_request_id) + + def test_init_slog(self): d_slog = { "adcGain": 0, "autoExposure": 0, @@ -147,6 +152,20 @@ def test_init(self): ir_slog = trp.ImageRecord(**d_slog) self.assertEqual("NavCam Left", ir_slog.instrument_name) + d2_slog = d_slog.copy() + del d2_slog["outputImageMask"] + trp.ImageRecord(**d2_slog) + + d3_slog = d_slog.copy() + del d3_slog["product_id"] + del d3_slog["outputImageMask"] + trp.ImageRecord(**d3_slog) + + err_slog = d_slog.copy() + err_slog["product_id"] = "231026-200000-ncl-z" # bad compression letter + err_slog["outputImageMask"] = None + self.assertRaises(ValueError, trp.ImageRecord, **err_slog) + # for k in dir(rp): # if k.startswith(("_", "validate_")): # continue @@ -175,10 +194,31 @@ def test_init_errors(self): d["onboard_compression_ratio"] = 999 self.assertRaises(ValueError, trp.ImageRecord, **d) + d = self.d.copy() + del d["lobt"] + d["start_time"] = self.startUTC # correct + d["product_id"] = "220127-000001-ncl-c" # pid.datetime incorrect + self.assertRaises(ValueError, trp.ImageRecord, **d) + d = self.d.copy() d["cameraId"] = 1 self.assertWarns(UserWarning, trp.ImageRecord, **d) + d = self.d.copy() + d["product_id"] = "220127-000001-ncl-s" # pid.compression incorrect + del d["output_image_mask"] + del d["processing_info"] + self.assertRaises(ValueError, trp.ImageRecord, **d) + + d = self.d.copy() + del d["output_image_mask"] + del d["icer_byte_quota"] + self.assertRaises(ValueError, trp.ImageRecord, **d) + + d = self.d.copy() + d["processing_info"] = 99 + self.assertWarns(UserWarning, trp.ImageRecord, **d) + # Commented out while this exception has been converted to a warning until we # sort out the Yamcs parameter. # def test_mcam_id(self): @@ -201,6 +241,8 @@ def test_lt(self): ) self.assertTrue(ir1 < ir2) + self.assertEqual(NotImplemented, ir1.__lt__("not an ImageRecord")) + def test_update(self): rp = trp.ImageRecord(**self.d) k = "foo" @@ -483,3 +525,17 @@ def test_fromxml(self): """ # noqa: E501 rp = trp.ImageRecord.from_xml(t.encode()) self.assertEqual("231125-143859-ncl-d", rp.product_id) + + t_not_viper_vis = t.replace( + "urn:nasa:pds:viper_vis:raw:231125-143859-ncl-d", + "urn:nasa:pds:NOT_viper_vis:raw:231125-143859-ncl-d", + ) + self.assertRaises( + ValueError, trp.ImageRecord.from_xml, t_not_viper_vis.encode() + ) + + t_not_raw = t.replace( + "urn:nasa:pds:viper_vis:raw:231125-143859-ncl-d", + "urn:nasa:pds:viper_vis:NOT_raw:231125-143859-ncl-d", + ) + self.assertRaises(ValueError, trp.ImageRecord.from_xml, t_not_raw.encode()) diff --git a/tests/test_pid.py b/tests/test_pid.py index 82cb561..a132781 100644 --- a/tests/test_pid.py +++ b/tests/test_pid.py @@ -27,12 +27,20 @@ # import shutil import datetime import unittest +from unittest.mock import patch # from pathlib import Path from vipersci.pds import pid +class TestGetKey(unittest.TestCase): + def test_get_key(self): + d = {"a": "alpha", "b": "beta", "g": "gamma"} + self.assertEqual(pid.get_key("beta", d), "b") + self.assertRaises(KeyError, pid.get_key, "omega", d) + + class TestVIPERID(unittest.TestCase): def test_init_tuple(self): tuples = ( @@ -80,6 +88,8 @@ def test_init_bad_tuples(self): ("not a date", datetime.time(1, 1, 1), "ncl"), (datetime.date(2024, 1, 1), "not a time", "ncl"), (datetime.datetime(2022, 1, 17, 1, 1, 1), "not an instrument"), + (999, datetime.time(1, 1, 1), "ncl"), + (datetime.date(2024, 1, 1), 999, "ncl"), ) for t in tuples: with self.subTest(t): @@ -122,6 +132,10 @@ def test_str_(self): truth = "220117-010101-aim" self.assertEqual(truth, pid.VIPERID(test).__str__()) + def test_eq(self): + p1 = pid.VIPERID("231120-010101-acl") + self.assertFalse(p1 == "foo") + def test_lt(self): p1 = pid.VIPERID("231120-010101-acl") p2 = pid.VIPERID("231121-010101-acl") @@ -130,6 +144,10 @@ def test_lt(self): self.assertTrue(p1 < p2) self.assertTrue(p2 < p3) self.assertTrue(p3 < p4) + self.assertEqual(NotImplemented, p1.__lt__("foo")) + + def test_format_time(self): + self.assertRaises(ValueError, pid.VISID.format_time, datetime.time(1, 1, 1, 1)) def test_datetime(self): p = pid.VIPERID("231121-010101-acl") @@ -138,6 +156,12 @@ def test_datetime(self): datetime.datetime(2023, 11, 21, 1, 1, 1, tzinfo=datetime.timezone.utc), ) + p2 = pid.VIPERID("240219-010101001-aim") + self.assertEqual( + p2.datetime(), + datetime.datetime(2024, 2, 19, 1, 1, 1, 1000, tzinfo=datetime.timezone.utc), + ) + class TestVISID(unittest.TestCase): def test_init_tuple(self): @@ -190,12 +214,41 @@ def test_init_dict(self): vid = pid.VISID(d) self.assertEqual("220127-000000-ncl-b", str(vid)) + vid2 = pid.VISID( + dict( + start_time=datetime.datetime( + 2024, 2, 19, 1, 1, 1, tzinfo=datetime.timezone.utc + ), + instrument_name="NavCam Left", + onboard_compression_ratio=5, + ) + ) + self.assertEqual("240219-010101-ncl-b", str(vid2)) + + self.assertRaises(ValueError, pid.VISID, dict(a="NavCam Left", b=5)) + def test_init_bad_tuples(self): - tuples = ((datetime.date(2024, 1, 1), datetime.time(1, 1, 1), "ncl", "w"),) + tuples = ( + (datetime.date(2024, 1, 1), datetime.time(1, 1, 1), "ncl", "w"), + (datetime.date(2024, 1, 1), datetime.time(1, 1, 1), "ncl", 0.5), + ) for t in tuples: with self.subTest(test=t): self.assertRaises(ValueError, pid.VISID, *t) + with patch( + "vipersci.pds.pid.vis_compression", return_value=dict(s="SLoG", z=None) + ): + self.assertRaises( + ValueError, + pid.VISID, + dict( + lobt=1643241600, + instrument_name="NavCam Left", + onboard_compression_ratio=5, + ), + ) + def test_init_bad_strings(self): strings = ( "220117-010101-ncl", @@ -245,6 +298,10 @@ def test_repr(self): cid = pid.VISID(s) self.assertEqual("VISID('220117-010101-ncl-a')", repr(cid)) + def test_eq(self): + vid0 = pid.VISID("220117-010101-ncl-z") + self.assertFalse(vid0 == "foo") + def test_lt(self): vid0 = pid.VISID("220117-010101-ncl-z") vid1 = pid.VISID("220117-010101-ncl-a") @@ -256,6 +313,8 @@ def test_lt(self): self.assertTrue(vid2 < vid3) self.assertTrue(vid3 < vid4) + self.assertEqual(NotImplemented, vid0.__lt__("foo")) + vids = [vid3, vid4, vid1, vid2, vid0] self.assertEqual(sorted(vids), [vid0, vid1, vid2, vid3, vid4]) @@ -275,6 +334,10 @@ def test_best_compression(self): result = pid.VISID.best_compression(test) self.assertEqual(result, truth) + def test_compression_class(self): + vid = pid.VISID("240219-010101-ncl-a") + self.assertEqual(vid.compression_class(), "Lossless") + class TestPanoID(unittest.TestCase): def test_init_tuple(self):