adsist-cms/python-detector-worker
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

Update markdown
All checks were successful
Build Worker Base and Application Images / check-base-changes (push) Successful in 8s
Details
Build Worker Base and Application Images / build-base (push) Has been skipped
Details
Build Worker Base and Application Images / build-docker (push) Successful in 2m15s
Details
Build Worker Base and Application Images / deploy-stack (push) Successful in 8s
Details

Browse source
This commit is contained in:
ziesorx 2025-08-10 20:51:16 +07:00
parent 1c21f417ce
commit cfc7503a14
3 changed files with 327 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

173
pympta.md
View file

@ -32,14 +32,15 @@ This modular structure allows for creating complex and efficient inference logic
## `pipeline.json` Specification
This file defines the entire pipeline logic. The root object contains a `pipeline` key for the pipeline definition and an optional `redis` key for Redis configuration.
This file defines the entire pipeline logic. The root object contains a `pipeline` key for the pipeline definition, optional `redis` key for Redis configuration, and optional `postgresql` key for database integration.
### Top-Level Object Structure
| Key | Type | Required | Description |
| ---------- | ------ | -------- | ------------------------------------------------------- |
| `pipeline` | Object | Yes | The root node object of the pipeline. |
| `redis` | Object | No | Configuration for connecting to a Redis server. |
| Key | Type | Required | Description |
| ------------ | ------ | -------- | ------------------------------------------------------- |
| `pipeline` | Object | Yes | The root node object of the pipeline. |
| `redis` | Object | No | Configuration for connecting to a Redis server. |
| `postgresql` | Object | No | Configuration for connecting to a PostgreSQL database. |
### Redis Configuration (`redis`)
@ -50,6 +51,16 @@ This file defines the entire pipeline logic. The root object contains a `pipelin
| `password` | String | No | The password for Redis authentication. |
| `db` | Number | No | The Redis database number to use. Defaults to `0`. |
### PostgreSQL Configuration (`postgresql`)
| Key | Type | Required | Description |
| ---------- | ------ | -------- | ------------------------------------------------------- |
| `host` | String | Yes | The hostname or IP address of the PostgreSQL server. |
| `port` | Number | Yes | The port number of the PostgreSQL server. |
| `database` | String | Yes | The database name to connect to. |
| `username` | String | Yes | The username for database authentication. |
| `password` | String | Yes | The password for database authentication. |
### Node Object Structure
| Key | Type | Required | Description |
@ -59,12 +70,17 @@ This file defines the entire pipeline logic. The root object contains a `pipelin
| `minConfidence` | Float | Yes | The minimum confidence score (0.0 to 1.0) required for a detection to be considered valid and potentially trigger a branch. |
| `triggerClasses` | Array<String> | Yes | A list of class names that, when detected by the parent, can trigger this node. For the root node, this lists all classes of interest. |
| `crop` | Boolean | No | If `true`, the image is cropped to the parent's detection bounding box before being passed to this node's model. Defaults to `false`. |
| `cropClass` | String | No | The specific class to use for cropping (e.g., "Frontal" for frontal view cropping). |
| `multiClass` | Boolean | No | If `true`, enables multi-class detection mode where multiple classes can be detected simultaneously. |
| `expectedClasses` | Array<String> | No | When `multiClass` is true, defines which classes are expected. At least one must be detected for processing to continue. |
| `parallel` | Boolean | No | If `true`, this branch will be processed in parallel with other parallel branches. |
| `branches` | Array<Node> | No | A list of child node objects that can be triggered by this node's detections. |
| `actions` | Array<Action> | No | A list of actions to execute upon a successful detection in this node. |
| `parallelActions` | Array<Action> | No | A list of actions to execute after all specified branches have completed. |
### Action Object Structure
Actions allow the pipeline to interact with Redis. They are executed sequentially for a given detection.
Actions allow the pipeline to interact with Redis and PostgreSQL databases. They are executed sequentially for a given detection.
#### Action Context & Dynamic Keys
@ -72,7 +88,12 @@ All actions have access to a dynamic context for formatting keys and messages. T
- All key-value pairs from the detection result (e.g., `class`, `confidence`, `id`).
- `{timestamp_ms}`: The current Unix timestamp in milliseconds.
- `{timestamp}`: Formatted timestamp string (YYYY-MM-DDTHH-MM-SS).
- `{uuid}`: A unique identifier (UUID4) for the detection event.
- `{filename}`: Generated filename with UUID.
- `{camera_id}`: Full camera subscription identifier.
- `{display_id}`: Display identifier extracted from subscription.
- `{session_id}`: Session ID for database operations.
- `{image_key}`: If a `redis_save_image` action has already been executed for this event, this placeholder will be replaced with the key where the image was stored.
#### `redis_save_image`
@ -83,6 +104,9 @@ Saves the current image frame (or cropped sub-image) to a Redis key.
| ---------------- | ------ | -------- | ------------------------------------------------------------------------------------------------------- |
| `type` | String | Yes | Must be `"redis_save_image"`. |
| `key` | String | Yes | The Redis key to save the image to. Can contain any of the dynamic placeholders. |
| `region` | String | No | Specific detected region to crop and save (e.g., "Frontal"). |
| `format` | String | No | Image format: "jpeg" or "png". Defaults to "jpeg". |
| `quality` | Number | No | JPEG quality (1-100). Defaults to 90. |
| `expire_seconds` | Number | No | If provided, sets an expiration time (in seconds) for the Redis key. |
#### `redis_publish`
@ -95,35 +119,98 @@ Publishes a message to a Redis channel.
| `channel` | String | Yes | The Redis channel to publish the message to. |
| `message` | String | Yes | The message to publish. Can contain any of the dynamic placeholders, including `{image_key}`. |
### Example `pipeline.json` with Redis
#### `postgresql_update_combined`
This example demonstrates a pipeline that detects vehicles, saves a uniquely named image of each detection that expires in one hour, and then publishes a notification with the image key.
Updates PostgreSQL database with results from multiple branches after they complete.
| Key | Type | Required | Description |
| ------------------ | ------------- | -------- | ------------------------------------------------------------------------------------------------------- |
| `type` | String | Yes | Must be `"postgresql_update_combined"`. |
| `table` | String | Yes | The database table name (will be prefixed with `gas_station_1.` schema). |
| `key_field` | String | Yes | The field to use as the update key (typically "session_id"). |
| `key_value` | String | Yes | Template for the key value (e.g., "{session_id}"). |
| `waitForBranches` | Array<String> | Yes | List of branch model IDs to wait for completion before executing update. |
| `fields` | Object | Yes | Field mapping object where keys are database columns and values are templates (e.g., "{branch.field}").|
### Complete Example `pipeline.json`
This example demonstrates a comprehensive pipeline for vehicle detection with parallel classification and database integration:
```json
{
"redis": {
"host": "redis.local",
"host": "10.100.1.3",
"port": 6379,
"password": "your-super-secret-password"
"password": "your-redis-password",
"db": 0
},
"postgresql": {
"host": "10.100.1.3",
"port": 5432,
"database": "inference",
"username": "root",
"password": "your-db-password"
},
"pipeline": {
"modelId": "vehicle-detector",
"modelFile": "vehicle_model.pt",
"minConfidence": 0.6,
"triggerClasses": ["car", "truck"],
"modelId": "car_frontal_detection_v1",
"modelFile": "car_frontal_detection_v1.pt",
"crop": false,
"triggerClasses": ["Car", "Frontal"],
"minConfidence": 0.8,
"multiClass": true,
"expectedClasses": ["Car", "Frontal"],
"actions": [
{
"type": "redis_save_image",
"key": "detections:{class}:{timestamp_ms}:{uuid}",
"expire_seconds": 3600
"region": "Frontal",
"key": "inference:{display_id}:{timestamp}:{session_id}:{filename}",
"expire_seconds": 600,
"format": "jpeg",
"quality": 90
},
{
"type": "redis_publish",
"channel": "vehicle_events",
"message": "{\"event\":\"new_detection\",\"class\":\"{class}\",\"confidence\":{confidence},\"image_key\":\"{image_key}\"}"
"channel": "car_detections",
"message": "{\"event\":\"frontal_detected\"}"
}
],
"branches": []
"branches": [
{
"modelId": "car_brand_cls_v1",
"modelFile": "car_brand_cls_v1.pt",
"crop": true,
"cropClass": "Frontal",
"resizeTarget": [224, 224],
"triggerClasses": ["Frontal"],
"minConfidence": 0.85,
"parallel": true,
"branches": []
},
{
"modelId": "car_bodytype_cls_v1",
"modelFile": "car_bodytype_cls_v1.pt",
"crop": true,
"cropClass": "Car",
"resizeTarget": [224, 224],
"triggerClasses": ["Car"],
"minConfidence": 0.85,
"parallel": true,
"branches": []
}
],
"parallelActions": [
{
"type": "postgresql_update_combined",
"table": "car_frontal_info",
"key_field": "session_id",
"key_value": "{session_id}",
"waitForBranches": ["car_brand_cls_v1", "car_bodytype_cls_v1"],
"fields": {
"car_brand": "{car_brand_cls_v1.brand}",
"car_body_type": "{car_bodytype_cls_v1.body_type}"
}
}
]
}
}
```
@ -134,7 +221,7 @@ The `pympta` module exposes two main functions.
### `load_pipeline_from_zip(zip_source: str, target_dir: str) -> dict`
Loads, extracts, and parses an `.mpta` file to build a pipeline tree in memory. It also establishes a Redis connection if configured in `pipeline.json`.
Loads, extracts, and parses an `.mpta` file to build a pipeline tree in memory. It also establishes Redis and PostgreSQL connections if configured in `pipeline.json`.
- **Parameters:**
- `zip_source` (str): The file path to the local `.mpta` zip archive.
@ -142,7 +229,7 @@ Loads, extracts, and parses an `.mpta` file to build a pipeline tree in memory.
- **Returns:**
- A dictionary representing the root node of the pipeline, ready to be used with `run_pipeline`. Returns `None` if loading fails.
### `run_pipeline(frame, node: dict, return_bbox: bool = False)`
### `run_pipeline(frame, node: dict, return_bbox: bool = False, context: dict = None)`
Executes the inference pipeline on a single image frame.
@ -150,12 +237,43 @@ Executes the inference pipeline on a single image frame.
- `frame`: The input image frame (e.g., a NumPy array from OpenCV).
- `node` (dict): The pipeline node to execute (typically the root node returned by `load_pipeline_from_zip`).
- `return_bbox` (bool): If `True`, the function returns a tuple `(detection, bounding_box)`. Otherwise, it returns only the `detection`.
- `context` (dict): Optional context dictionary containing camera_id, display_id, session_id for action formatting.
- **Returns:**
- The final detection result from the last executed node in the chain. A detection is a dictionary like `{'class': 'car', 'confidence': 0.95, 'id': 1}`. If no detection meets the criteria, it returns `None` (or `(None, None)` if `return_bbox` is `True`).
## Database Integration
The pipeline system includes automatic PostgreSQL database management:
### Table Schema (`gas_station_1.car_frontal_info`)
The system automatically creates and manages the following table structure:
```sql
CREATE TABLE IF NOT EXISTS gas_station_1.car_frontal_info (
display_id VARCHAR(255),
captured_timestamp VARCHAR(255),
session_id VARCHAR(255) PRIMARY KEY,
license_character VARCHAR(255) DEFAULT NULL,
license_type VARCHAR(255) DEFAULT 'No model available',
car_brand VARCHAR(255) DEFAULT NULL,
car_model VARCHAR(255) DEFAULT NULL,
car_body_type VARCHAR(255) DEFAULT NULL,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
```
### Workflow
1. **Initial Record Creation**: When both "Car" and "Frontal" are detected, an initial database record is created with a UUID session_id.
2. **Redis Storage**: Vehicle images are stored in Redis with keys containing the session_id.
3. **Parallel Classification**: Brand and body type classification run concurrently.
4. **Database Update**: After all branches complete, the database record is updated with classification results.
## Usage Example
This snippet, inspired by `pipeline_webcam.py`, shows how to use `pympta` to load a pipeline and process an image from a webcam.
This snippet shows how to use `pympta` with the enhanced features:
```python
import cv2
@ -181,9 +299,14 @@ while True:
if not ret:
break
# 4. Run the pipeline on the current frame
# The function will handle the entire logic tree (e.g., find a car, then find its license plate).
detection_result, bounding_box = run_pipeline(frame, model_tree, return_bbox=True)
# 4. Run the pipeline on the current frame with context
context = {
"camera_id": "display-001;cam-001",
"display_id": "display-001",
"session_id": None # Will be generated automatically
}
detection_result, bounding_box = run_pipeline(frame, model_tree, return_bbox=True, context=context)
# 5. Display the results
if detection_result: