Assign sequence numbers to segments with source change detection. Args: source_id: Unique identifier for segment source Returns: int: Next available sequence number None: If segment already downloaded
(self, source_id: str)
| 223 | return False |
| 224 | |
| 225 | def get_next_sequence(self, source_id: str) -> Optional[int]: |
| 226 | """ |
| 227 | Assign sequence numbers to segments with source change detection. |
| 228 | |
| 229 | Args: |
| 230 | source_id: Unique identifier for segment source |
| 231 | |
| 232 | Returns: |
| 233 | int: Next available sequence number |
| 234 | None: If segment already downloaded |
| 235 | |
| 236 | Side effects: |
| 237 | - Updates buffered sequences set |
| 238 | - Tracks source changes for discontinuity |
| 239 | - Maintains sequence numbering |
| 240 | """ |
| 241 | if source_id in self.downloaded_sources: |
| 242 | return None |
| 243 | |
| 244 | seq = self.next_sequence |
| 245 | while (seq in self.buffered_sequences): |
| 246 | seq += 1 |
| 247 | |
| 248 | # Track source changes for discontinuity markers |
| 249 | source_prefix = source_id.split('_')[0] |
| 250 | if not self.switching_stream and self.current_source and self.current_source != source_prefix: |
| 251 | self.source_changes.add(seq) |
| 252 | logging.debug(f"Source change detected at sequence {seq}") |
| 253 | self.current_source = source_prefix |
| 254 | |
| 255 | # Update tracking |
| 256 | self.downloaded_sources[source_id] = seq |
| 257 | self.buffered_sequences.add(seq) |
| 258 | self.next_sequence = seq + 1 |
| 259 | self.highest_sequence = max(self.highest_sequence, seq) |
| 260 | |
| 261 | return seq |
| 262 | |
| 263 | def _fetch_loop(self): |
| 264 | """Background thread for continuous stream fetching""" |