Open
Conversation
Contributor
Reviewer's GuideThis PR implements history length limitation for FSM snapshots, refactors AsyncStateStore to use asyncio.Queue with overflow handling and subscriber cleanup, migrates QSimService to bounded asyncio queues with drop-on-full logging, and adds comprehensive unit tests covering types, snapshot behaviors, and store functionality. Sequence diagram for subscriber notification and cleanup in AsyncStateStoresequenceDiagram
participant StateStore
participant Subscriber
loop For each subscriber
StateStore->>Subscriber: queue.put_nowait(snap)
alt Queue full
StateStore->>StateStore: Remove subscriber from list
end
end
Sequence diagram for QSimService queue overflow handlingsequenceDiagram
participant QSimService
participant WorldModel
participant SensorDataQueue
participant ActuatorCommandQueue
QSimService->>WorldModel: step(delta_time)
QSimService->>QSimService: generate_sensor_data()
QSimService->>SensorDataQueue: put_nowait(sensor_data)
alt SensorDataQueue full
QSimService->>QSimService: Log warning, drop data
end
QSimService->>ActuatorCommandQueue: put_nowait(command)
alt ActuatorCommandQueue full
QSimService->>QSimService: Log warning, drop command
end
Class diagram for FsmSnapshotDTO and TransitionDTO with history limitationclassDiagram
class FsmState {
<<enum>>
UNSPECIFIED
BOOTING
IDLE
ACTIVE
ERROR_STATE
SHUTDOWN
}
class TransitionStatus {
<<enum>>
UNSPECIFIED
SUCCESS
FAILED
PENDING
}
class TransitionDTO {
+from_state: FsmState
+to_state: FsmState
+trigger_event: str
+status: TransitionStatus
+error_message: str
+ts_mono: float
+ts_wall: float
}
class FsmSnapshotDTO {
+version: int
+state: FsmState
+reason: str
+ts_mono: float
+ts_wall: float
+snapshot_id: str
+prev_state: FsmState
+fsm_instance_id: str
+source_module: str
+attempt_count: int
+history: Tuple[TransitionDTO]
+context_data: Dict[str, str]
+state_metadata: Dict[str, str]
+add_transition(transition: TransitionDTO, max_history: int): FsmSnapshotDTO
}
FsmSnapshotDTO "*" -- "*" TransitionDTO : history
TransitionDTO --> FsmState
TransitionDTO --> TransitionStatus
FsmSnapshotDTO --> FsmState
Class diagram for AsyncStateStore with subscriber queue managementclassDiagram
class AsyncStateStore {
-_lock: asyncio.Lock
-_snap: FsmSnapshotDTO
-_subscribers: List[asyncio.Queue]
-_subscriber_ids: Dict[int, str]
-_metrics: Dict[str, Any]
+get(): FsmSnapshotDTO
+get_with_meta(): tuple[FsmSnapshotDTO, Dict[str, Any]]
+set(new_snap: FsmSnapshotDTO, enforce_version: bool): FsmSnapshotDTO
+subscribe(subscriber_id: str): asyncio.Queue
+unsubscribe(queue: asyncio.Queue)
+initialize_if_empty(): FsmSnapshotDTO
+get_metrics(): Dict[str, Any]
+health_check(): Dict[str, Any]
-_notify_subscribers(snap: FsmSnapshotDTO)
}
AsyncStateStore --> FsmSnapshotDTO
AsyncStateStore --> asyncio.Queue
Class diagram for QSimService with bounded asyncio queuesclassDiagram
class QSimService {
+config: Dict[str, Any]
+world_model: WorldModel
+sensor_data_queue: asyncio.Queue
+actuator_command_queue: asyncio.Queue
+generate_sensor_data(): SensorReading
+receive_actuator_command(command: ActuatorCommand)
+run()
+step()
}
QSimService --> WorldModel
QSimService --> asyncio.Queue
QSimService --> SensorReading
QSimService --> ActuatorCommand
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey there - I've reviewed your changes and they look great!
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `QIKI_DTMP/services/q_sim_service/main.py:51` </location>
<code_context>
+ logger.info(f"QSim received actuator command: {MessageToDict(command)}")
+ self.world_model.update(command) # Update world model based on command
+
+ def run(self):
+ logger.info("QSimService started.")
+ try:
+ while True:
+ self.step() # Call the new step method
+ time.sleep(self.config.get("sim_tick_interval", 1))
+ except KeyboardInterrupt:
+ logger.info("QSimService stopped by user.")
+
</code_context>
<issue_to_address>
Blocking run method uses time.sleep in a synchronous loop.
Refactor run to be async and replace time.sleep with await asyncio.sleep to prevent blocking the event loop and improve compatibility with async code.
Suggested implementation:
```python
async def run(self):
logger.info("QSimService started.")
try:
while True:
self.step() # Call the new step method
await asyncio.sleep(self.config.get("sim_tick_interval", 1))
except KeyboardInterrupt:
logger.info("QSimService stopped by user.")
```
If there are places in your codebase where `run()` is called directly (e.g., `service.run()`), you will need to update those calls to use `await service.run()` or schedule it with `asyncio.create_task(service.run())` depending on your application's async structure.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+51
to
+57
| def run(self): | ||
| logger.info("QSimService started.") | ||
| try: | ||
| while True: | ||
| self.step() # Call the new step method | ||
| time.sleep(self.config.get("sim_tick_interval", 1)) | ||
| except KeyboardInterrupt: |
Contributor
There was a problem hiding this comment.
suggestion: Blocking run method uses time.sleep in a synchronous loop.
Refactor run to be async and replace time.sleep with await asyncio.sleep to prevent blocking the event loop and improve compatibility with async code.
Suggested implementation:
async def run(self):
logger.info("QSimService started.")
try:
while True:
self.step() # Call the new step method
await asyncio.sleep(self.config.get("sim_tick_interval", 1))
except KeyboardInterrupt:
logger.info("QSimService stopped by user.")If there are places in your codebase where run() is called directly (e.g., service.run()), you will need to update those calls to use await service.run() or schedule it with asyncio.create_task(service.run()) depending on your application's async structure.
| def __post_init__(self): | ||
| # Устанавливаем значения по умолчанию для mutable полей | ||
| if self.history is None: | ||
| object.__setattr__(self, 'history', tuple()) |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Replace tuple() with () (tuple-literal)
Suggested change
| object.__setattr__(self, 'history', tuple()) | |
| object.__setattr__(self, 'history', ()) |
Explanation
The most concise and Pythonic way to create an empty tuple is to use the()literal.
This fits in with the way we create tuples with items, saving a bit of
mental energy that might be taken up with thinking about two different ways of
creating tuples:
x = ("first", "second")Doing things this way has the added advantage of being a nice little performance
improvement. Here are the timings before and after the change:
$ python3 -m timeit "tuple()"
10000000 loops, best of 5: 22.6 nsec per loop
$ python3 -m timeit "()"
50000000 loops, best of 5: 5.46 nsec per loop
| new_version = current.version + version_increment | ||
|
|
||
| # Новая история переходов | ||
| new_history = current.history if current.history else tuple() |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Replace if-expression with or (or-if-exp-identity)
Suggested change
| new_history = current.history if current.history else tuple() | |
| new_history = current.history or tuple() |
Explanation
Here we find ourselves setting a value if it evaluates toTrue, and otherwiseusing a default.
The 'After' case is a bit easier to read and avoids the duplication of
input_currency.
It works because the left-hand side is evaluated first. If it evaluates to
true then currency will be set to this and the right-hand side will not be
evaluated. If it evaluates to false the right-hand side will be evaluated and
currency will be set to DEFAULT_CURRENCY.
| new_version = current.version + version_increment | ||
|
|
||
| # Новая история переходов | ||
| new_history = current.history if current.history else tuple() |
Contributor
There was a problem hiding this comment.
suggestion (code-quality): Replace tuple() with () (tuple-literal)
Suggested change
| new_history = current.history if current.history else tuple() | |
| new_history = current.history if current.history else () |
Explanation
The most concise and Pythonic way to create an empty tuple is to use the()literal.
This fits in with the way we create tuples with items, saving a bit of
mental energy that might be taken up with thinking about two different ways of
creating tuples:
x = ("first", "second")Doing things this way has the added advantage of being a nice little performance
improvement. Here are the timings before and after the change:
$ python3 -m timeit "tuple()"
10000000 loops, best of 5: 22.6 nsec per loop
$ python3 -m timeit "()"
50000000 loops, best of 5: 5.46 nsec per loop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
FsmSnapshotDTO.add_transitionasyncio.Queueс обработкой переполненияTesting
pytest QIKI_DTMP/services/q_core_agent/state/tests/test_types.py -qpytest QIKI_DTMP/services/q_core_agent/state/tests/test_store.py -qhttps://chatgpt.com/codex/tasks/task_e_68a27e4b71848331af1d95c7f02e0293
Summary by Sourcery
Implement history length restriction for FSM snapshots, migrate simulation queues in QSimService to asyncio with drop-on-full logic, enhance AsyncStateStore subscriber management and version/metrics enforcement, and introduce comprehensive unit tests for the FSM types and state store.
New Features:
Enhancements:
Tests: