| 179 | |
| 180 | |
| 181 | class FifoBuff: |
| 182 | def __init__(self, buff_size, threshold_size): |
| 183 | self.data = bytearray(buff_size) |
| 184 | self.buff_size = buff_size |
| 185 | self.threshold_size = threshold_size |
| 186 | self.threshold = 0 |
| 187 | self.idx_put = 0 |
| 188 | self.idx_get = 0 |
| 189 | self.count = 0 |
| 190 | |
| 191 | def put(self, byte): |
| 192 | logger.info("Put byte in FIFO") |
| 193 | |
| 194 | # Check if buffer is already full |
| 195 | if self.count == self.buff_size: |
| 196 | logger.info("FIFO is full") |
| 197 | return None |
| 198 | else: |
| 199 | self.data[self.idx_put] = byte |
| 200 | logger.debug(f"Byte {byte} inserted at position {self.idx_put}") |
| 201 | self.idx_put = (self.idx_put + 1) % self.buff_size |
| 202 | self.count += 1 |
| 203 | if (self.count >= self.threshold_size) and (self.threshold_size != 0): |
| 204 | self.threshold = 1 |
| 205 | return byte |
| 206 | |
| 207 | def get(self): |
| 208 | logger.info("Get byte from FIFO") |
| 209 | |
| 210 | # Check if buffer is already empty |
| 211 | if self.count == 0: |
| 212 | logger.info("FIFO is empty") |
| 213 | return None |
| 214 | else: |
| 215 | byte = self.data[self.idx_get] |
| 216 | logger.debug(f"Byte {byte} extracted from position {self.idx_get}") |
| 217 | self.idx_get = (self.idx_get + 1) % self.buff_size |
| 218 | self.count -= 1 |
| 219 | if self.count < self.threshold_size: |
| 220 | self.threshold = 0 |
| 221 | return byte |
| 222 | |
| 223 | |
| 224 | # User registers |
no outgoing calls
no test coverage detected