LSB 输入流，按顺序从图像各像素通道的最低有效位中还原数据。
| 9 | |
| 10 | |
| 11 | class LSBInputStream: |
| 12 | """LSB 输入流，按顺序从图像各像素通道的最低有效位中还原数据。""" |
| 13 | |
| 14 | def __init__(self, image: Image.Image, config: LSBConfig): |
| 15 | self.config = config |
| 16 | self.pixels = np.array(image) |
| 17 | self.height, self.width, self.channels = self.pixels.shape |
| 18 | self._bit_offset = 0 |
| 19 | |
| 20 | # 先读固定部分以获取 filename_len，再读变长文件名 |
| 21 | fixed_size = (len(LSBDataHeader.DATA_STAMP) + len(LSBDataHeader.HEADER_VERSION) |
| 22 | + LSBDataHeader.FIXED_HEADER_LENGTH + LSBDataHeader.CRYPT_ALGO_LENGTH) |
| 23 | fixed_bytes = self._read_bytes(fixed_size) |
| 24 | |
| 25 | fixed_header_offset = len(LSBDataHeader.DATA_STAMP) + len(LSBDataHeader.HEADER_VERSION) |
| 26 | filename_len = fixed_bytes[fixed_header_offset + 5] |
| 27 | |
| 28 | filename_bytes = self._read_bytes(filename_len) if filename_len > 0 else b'' |
| 29 | try: |
| 30 | self.header = LSBDataHeader.from_bytes(fixed_bytes + filename_bytes, config) |
| 31 | except Exception as e: |
| 32 | raise ValueError(f"无法解析数据头: {e}") |
| 33 | |
| 34 | self.channel_bits_used = self.header.get_channel_bits_used() |
| 35 | |
| 36 | def _read_bytes(self, n_bytes: int) -> bytes: |
| 37 | bits_per_ch = getattr(self, 'channel_bits_used', None) or self.config.get_max_bits_used_per_channel() |
| 38 | n_bits = n_bytes * 8 |
| 39 | |
| 40 | if bits_per_ch == 1: |
| 41 | flat = self.pixels.reshape(-1) |
| 42 | s, e = self._bit_offset, self._bit_offset + n_bits |
| 43 | bits = (flat[s:e] & 1).astype(np.uint8) |
| 44 | self._bit_offset += n_bits |
| 45 | return np.packbits(bits).tobytes() |
| 46 | else: |
| 47 | result = bytearray() |
| 48 | for _ in range(n_bytes): |
| 49 | byte_val = 0 |
| 50 | for _ in range(8): |
| 51 | byte_val = (byte_val << 1) | self._read_one_bit() |
| 52 | result.append(byte_val) |
| 53 | return bytes(result) |
| 54 | |
| 55 | def _read_one_bit(self) -> int: |
| 56 | bits_per_ch = getattr(self, 'channel_bits_used', None) or self.config.get_max_bits_used_per_channel() |
| 57 | ch_pos = self._bit_offset // bits_per_ch |
| 58 | bit_slot = self._bit_offset % bits_per_ch |
| 59 | pixel_idx = ch_pos // self.channels |
| 60 | channel = ch_pos % self.channels |
| 61 | row = pixel_idx // self.width |
| 62 | col = pixel_idx % self.width |
| 63 | bit = (int(self.pixels[row, col, channel]) >> bit_slot) & 1 |
| 64 | self._bit_offset += 1 |
| 65 | return bit |
| 66 | |
| 67 | def read(self, size: int = -1) -> bytes: |
| 68 | n = self.header.get_data_length() if size < 0 else min(size, self.header.get_data_length()) |
no outgoing calls
no test coverage detected