Coverage for tests/test_fuzz.py: 100%

1# This file is part of lsst-images. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11from __future__ import annotations 

12 

13import os 

14import tempfile 

15import unittest 

16 

17import astropy.units as u 

18import numpy as np 

19from click.testing import CliRunner 

20 

21from lsst.images import Image, MaskedImage, MaskPlane, MaskSchema 

22from lsst.images.cli import main 

23from lsst.images.cli._fuzz import shuffle_blocks 

24from lsst.images.serialization import read 

25 

26 

27class ShuffleBlocksTestCase(unittest.TestCase): 

28 """Unit tests for the pure pixel-shuffling helper.""" 

29 

30 def test_consistency_even_blocks(self) -> None: 

31 # Each pixel carries the same code in all three planes, so a shared 

32 # per-block permutation must keep the planes aligned afterwards. 

33 ny, nx = 4, 4 

34 codes = np.arange(ny * nx, dtype=np.float64).reshape(ny, nx) 

35 image = codes.copy() 

36 variance = codes.copy() 

37 mask = codes.astype(np.uint8).reshape(ny, nx, 1).copy() 

38 

39 shuffle_blocks(image, mask, variance, (2, 2), np.random.default_rng(7)) 

40 

41 np.testing.assert_array_equal(image, variance) 

42 np.testing.assert_array_equal(image.astype(np.uint8), mask[..., 0]) 

43 # Every 2x2 block keeps its original multiset of values. 

44 for y0 in (0, 2): 

45 for x0 in (0, 2): 

46 np.testing.assert_array_equal( 

47 np.sort(image[y0 : y0 + 2, x0 : x0 + 2], axis=None), 

48 np.sort(codes[y0 : y0 + 2, x0 : x0 + 2], axis=None), 

49 ) 

50 

51 def test_partial_edge_blocks(self) -> None: 

52 # A 2x2 block does not divide a 5x5 image evenly; the edge blocks are 

53 # smaller and must still shuffle without error or value loss. 

54 ny, nx = 5, 5 

55 codes = np.arange(ny * nx, dtype=np.float64).reshape(ny, nx) 

56 image = codes.copy() 

57 variance = codes.copy() 

58 mask = codes.astype(np.uint8).reshape(ny, nx, 1).copy() 

59 

60 shuffle_blocks(image, mask, variance, (2, 2), np.random.default_rng(3)) 

61 

62 np.testing.assert_array_equal(image, variance) 

63 np.testing.assert_array_equal(image.astype(np.uint8), mask[..., 0]) 

64 np.testing.assert_array_equal(np.sort(image, axis=None), np.sort(codes, axis=None)) 

65 

66 

67def _make_masked_image() -> MaskedImage: 

68 """Build a small MaskedImage with noisy pixels and some mask bits set.""" 

69 rng = np.random.default_rng(500) 

70 mi = MaskedImage( 

71 Image(rng.normal(100.0, 8.0, size=(64, 64)), dtype=np.float32, unit=u.nJy), 

72 mask_schema=MaskSchema([MaskPlane("BAD", "Bad pixel.")]), 

73 metadata={"hello": "world"}, 

74 ) 

75 mi.mask.array |= np.multiply.outer(mi.image.array < 100.0, mi.mask.schema.bitmask("BAD")) 

76 mi.variance.array = rng.normal(64.0, 0.5, size=mi.bbox.shape).astype(np.float32) 

77 return mi 

78 

79 

80class FuzzMaskedImageCommandTestCase(unittest.TestCase): 

81 """Tests for the fuzz-masked-image CLI command.""" 

82 

83 def setUp(self) -> None: 

84 tmp = tempfile.TemporaryDirectory() 

85 self.addCleanup(tmp.cleanup) 

86 self.tmp = tmp.name 

87 

88 def test_help(self) -> None: 

89 result = CliRunner().invoke(main, ["fuzz-masked-image", "--help"]) 

90 self.assertEqual(result.exit_code, 0, result.output) 

91 

92 def test_no_files_errors(self) -> None: 

93 result = CliRunner().invoke(main, ["fuzz-masked-image"]) 

94 self.assertNotEqual(result.exit_code, 0) 

95 

96 def test_round_trip_fits(self) -> None: 

97 src = os.path.join(self.tmp, "mi.fits") 

98 mi = _make_masked_image() 

99 mi.write(src) 

100 original_image = mi.image.array.copy() 

101 original_mask = mi.mask.array.copy() 

102 

103 result = CliRunner().invoke(main, ["fuzz-masked-image", src]) 

104 self.assertEqual(result.exit_code, 0, result.output) 

105 

106 out = os.path.join(self.tmp, "mi.fuzzed.fits") 

107 self.assertTrue(os.path.exists(out)) 

108 check = read(out) 

109 finite = np.isfinite(original_image) 

110 changed = float(np.mean(check.image.array[finite] != original_image[finite])) 

111 self.assertGreaterEqual(changed, 0.5) 

112 self.assertFalse(np.array_equal(check.mask.array, original_mask)) 

113 # An untouched part of the object survives the round trip. 

114 self.assertEqual(check.metadata.get("hello"), "world") 

115 

116 def test_skips_existing_without_overwrite(self) -> None: 

117 src = os.path.join(self.tmp, "mi.fits") 

118 _make_masked_image().write(src) 

119 out = os.path.join(self.tmp, "mi.fuzzed.fits") 

120 with open(out, "w") as stream: 

121 stream.write("EXISTING") 

122 result = CliRunner().invoke(main, ["fuzz-masked-image", src]) 

123 self.assertEqual(result.exit_code, 0, result.output) 

124 with open(out) as stream: 

125 self.assertEqual(stream.read(), "EXISTING") 

126 

127 

128if __name__ == "__main__": 

129 unittest.main()