92 lines
3.5 KiB
Text
92 lines
3.5 KiB
Text
@startuml Crowd Node - Request Processing Flow
|
||
|
||
!define COMPONENT_BG_COLOR #E3F2FD
|
||
!define API_BG_COLOR #FFF3E0
|
||
!define STORAGE_BG_COLOR #F3E5F5
|
||
|
||
title Crowd Node: Процесс обработки запроса на анализ изображения
|
||
|
||
participant "Scheduler" as Scheduler
|
||
participant "Redis Queue" as RedisQueue #FFCCCC
|
||
participant "tasks.py" as TaskListener COMPONENT_BG_COLOR
|
||
participant "analyzer.py" as Analyzer COMPONENT_BG_COLOR
|
||
participant "frames.py" as FramePuller COMPONENT_BG_COLOR
|
||
participant "LivePreview service" as LivePreview API_BG_COLOR
|
||
box #LightGreen
|
||
participant "PersonDetector" as PersonDetector
|
||
participant "PersonDetectionService" as PersonDetectionService
|
||
end box
|
||
participant "Redis Cache" as RedisCache #FFCCCC
|
||
participant "S3 Storage" as S3 STORAGE_BG_COLOR
|
||
participant "Central" as Central API_BG_COLOR
|
||
|
||
== Получение задания ==
|
||
|
||
Scheduler -> RedisQueue: push task\n{cmd: "analyze_crowd",\nparams: {uin, camera_id, zones, ...}}
|
||
activate RedisQueue
|
||
|
||
TaskListener -> RedisQueue: pull_task()
|
||
activate TaskListener
|
||
RedisQueue --> TaskListener: task data
|
||
deactivate RedisQueue
|
||
|
||
TaskListener -> Analyzer: analyze_crowd(uin, camera_id, zones, **options)
|
||
activate Analyzer
|
||
|
||
== Получение кадра ==
|
||
|
||
Analyzer -> FramePuller: pull(uin, camera)
|
||
activate FramePuller
|
||
FramePuller -> LivePreview: GET /internal/preview
|
||
LivePreview --> FramePuller: JPEG image bytes
|
||
FramePuller -> FramePuller: Frame(content)\n- создает PIL Image\n- генерирует image_id
|
||
FramePuller --> Analyzer: Frame object
|
||
deactivate FramePuller
|
||
|
||
group #LightGreen Новый подход
|
||
== Взаимодействие с новым сервисом==
|
||
Analyzer -> PersonDetector: _run_detecotrs(uin, camera, frame, zones)
|
||
activate PersonDetector
|
||
PersonDetector -> PersonDetector: prepare() - Метод для подготовки\nданных для отправки в сервис
|
||
PersonDetector -> PersonDetectionService: POST /picture/analyze\nМетод для анализа
|
||
activate PersonDetectionService
|
||
PersonDetectionService -> PersonDetectionService: Сервис анализирует изображение
|
||
PersonDetectionService --> PersonDetector: {results:\n\t[detection_1, detection_2, ...]\n}
|
||
deactivate PersonDetectionService
|
||
PersonDetector -> PersonDetector: parse_response() - Метод\nдля приведения резултата в формат,\nкоторый был раньше
|
||
end
|
||
PersonDetector --> Analyzer: result dict\n{zone_id: {count, objects}, ...}
|
||
deactivate PersonDetector
|
||
|
||
|
||
== Форматирование результата ==
|
||
deactivate Detector
|
||
|
||
== Сохранение результата ==
|
||
|
||
Analyzer -> S3: storage.upload_fileobj(\n image, bucket, key, ...)
|
||
S3 --> Analyzer: ObjRef
|
||
|
||
Analyzer -> S3: storage.generate_presigned_url(obj_ref)
|
||
S3 --> Analyzer: presigned_url
|
||
|
||
== Отправка результата в Central ==
|
||
|
||
Analyzer -> Central: central.send('new_measurement', {\n timestamp,\n camera_id,\n measurement_id,\n image: zones_url,\n timings,\n errors,\n zones: zones_info\n})
|
||
note right
|
||
Отправка через aio_broker
|
||
в очередь 'overmind:input'
|
||
с командой 'new_measurement'
|
||
end note
|
||
Central --> Analyzer: (async, no wait)
|
||
|
||
Analyzer -> Analyzer: Обновить БД zones_db:\ndetected_at = time.time()
|
||
|
||
Analyzer --> TaskListener: complete
|
||
deactivate Analyzer
|
||
|
||
TaskListener -> TaskListener: Ожидать следующую задачу
|
||
deactivate TaskListener
|
||
|
||
@enduml
|
||
|