10 add downsampling option #33
Conversation
AnthonyCvn
left a comment
AnthonyCvn
left a comment
There was a problem hiding this comment.
Nice progress! I added a few comments for your review, and I'd also like to avoid having the state object updated in a static function (let's update it in the process_message function).
reductstore_agent/config_models.py
Outdated
| static_labels: dict[str, str] = Field(default_factory=dict) | ||
| filename_mode: FilenameMode = FilenameMode.TIMESTAMP | ||
|
|
||
| downsampling_mode: str = Field("none", pattern=r"^(none|max_rate|stride)$") |
There was a problem hiding this comment.
You can use an Enum for that field.
|
@hungngyenn I checked your integration test, the problem is that the publisher is at 10Hz and recorder has a I saw that other tests got broken and needs your attention. Some of which are about formatting and linting. For formatting, you can run: black .And sort the imports with: isort . For linting, you need to check the result of I hope that helps ! |
atimin
left a comment
atimin
left a comment
There was a problem hiding this comment.
Looks good to me, just move the Downloader into a separated file and use Enum for the downsampling_mode field as @AnthonyCvn asked. Good job!
AnthonyCvn
left a comment
AnthonyCvn
left a comment
There was a problem hiding this comment.
Looking good! Added small suggestions and then you can merge.
reductstore_agent/pipeline_state.py
Outdated
| timer: Timer | None = None | ||
| current_size: int = 0 | ||
| is_uploading: bool = False | ||
| downsampler: Optional["Downsampler"] = Field(default=None) |
There was a problem hiding this comment.
| downsampler: Optional["Downsampler"] = Field(default=None) | |
| downsampler: Downsampler | None = None |
reductstore_agent/recorder.py
Outdated
| topics=topics, | ||
| buffer=buffer, | ||
| writer=writer, | ||
| downsampler=downsampler_instance, |
There was a problem hiding this comment.
| downsampler=downsampler_instance, | |
| downsampler=Downsampler(cfg), |
Closes #10
Please check if the PR fulfills these requirements
What kind of change does this PR introduce?
This PR introduces per-pipeline dwnsampling.
What was changed?
Configuration: Updated PipelineConfig to include three new optional parameters: downsampling_mode, stride_n, and max_rate_hz.
Logic: Implemented the down_sampling method to act as a gate function, enforcing message skips based on the configured mode.
stride Mode: Skips messages based on a counter (stride_n), ensuring only the N-th message is recorded.
max_rate Mode: Skips messages based on time (max_rate_hz), ensuring the interval between recorded messages meets a minimum time threshold.
Related issues
(Add links to related issues)
Does this PR introduce a breaking change?
No.
Other information: