(self, buffer, value)
| 816 | self.itemsize, self.typecode, _, self.type_id = _np_dtypes_dict[self.dtype] |
| 817 | |
| 818 | def write(self, buffer, value): |
| 819 | assert value.itemsize == self.itemsize |
| 820 | view = memoryview(value) |
| 821 | try: |
| 822 | assert view.format == self.typecode |
| 823 | except AssertionError as e: |
| 824 | raise e |
| 825 | assert view.itemsize == self.itemsize |
| 826 | nbytes = len(value) * self.itemsize |
| 827 | buffer.write_var_uint32(nbytes) |
| 828 | if self.dtype == np.dtype("bool") or not view.c_contiguous: |
| 829 | if not is_little_endian and self.itemsize > 1: |
| 830 | # Swap bytes on big-endian machines for multi-byte types |
| 831 | buffer.write_bytes(value.astype(value.dtype.newbyteorder("<")).tobytes()) |
| 832 | else: |
| 833 | buffer.write_bytes(value.tobytes()) |
| 834 | elif is_little_endian or self.itemsize == 1: |
| 835 | buffer.write_buffer(value) |
| 836 | else: |
| 837 | # Swap bytes on big-endian machines for multi-byte types |
| 838 | buffer.write_bytes(value.astype(value.dtype.newbyteorder("<")).tobytes()) |
| 839 | |
| 840 | def read(self, buffer): |
| 841 | data = buffer.read_bytes_and_size() |
nothing calls this directly
no test coverage detected