(
self,
row_offset: int,
row_count: int,
)
| 3889 | return min(capacity, self.end_pos - self.start_pos - offset) |
| 3890 | |
| 3891 | def media_slice( |
| 3892 | self, |
| 3893 | row_offset: int, |
| 3894 | row_count: int, |
| 3895 | ) -> Tuple[np.ndarray, np.ndarray, List[int]]: |
| 3896 | if self.media is None: |
| 3897 | raise RuntimeError("media segment is missing embeddings or positions") |
| 3898 | row_start = row_offset |
| 3899 | row_end = row_offset + row_count |
| 3900 | embeddings = self.media.embeddings[row_start:row_end] |
| 3901 | if len(self.media.positions) == self.batch_rows: |
| 3902 | positions = self.media.positions[row_start:row_end] |
| 3903 | else: |
| 3904 | positions = ( |
| 3905 | self.media.positions.reshape(4, self.batch_rows)[:, row_start:row_end] |
| 3906 | .reshape(-1) |
| 3907 | ) |
| 3908 | return ( |
| 3909 | embeddings, |
| 3910 | positions, |
| 3911 | self.decoder_position_increments[row_start:row_end], |
| 3912 | ) |
| 3913 | |
| 3914 | |
| 3915 | @dataclass |
no outgoing calls
no test coverage detected