Spaces:
Running
Running
Download src/sonicagent_rt/protocol/sequence.py from ruslanmv/SonicAgent: direct link, hf CLI and curl.
- Browser
- Download file 749 Bytes
-
https://huggingface.co/spaces/ruslanmv/SonicAgent/resolve/main/src/sonicagent_rt/protocol/sequence.py
- Command line
-
hf download hf://spaces/ruslanmv/SonicAgent/src/sonicagent_rt/protocol/sequence.py
-
curl -L -o sequence.py https://huggingface.co/spaces/ruslanmv/SonicAgent/resolve/main/src/sonicagent_rt/protocol/sequence.py
749 Bytes
| from dataclasses import dataclass | |
| class SequenceError(ValueError): | |
| pass | |
| class SequenceValidator: | |
| expected: int = 0 | |
| allow_first: bool = True | |
| def accept(self, sequence: int) -> None: | |
| if sequence < 0 or sequence > 0xFFFFFFFF: | |
| raise SequenceError("sequence out of range") | |
| if self.allow_first and self.expected == 0: | |
| self.expected = (sequence + 1) & 0xFFFFFFFF | |
| self.allow_first = False | |
| return | |
| if sequence != self.expected: | |
| kind = "replay" if sequence < self.expected else "out-of-order" | |
| raise SequenceError(f"{kind} sequence: expected {self.expected}, got {sequence}") | |
| self.expected = (self.expected + 1) & 0xFFFFFFFF | |