Сервисы поведения

Сервисы поведения - модульные компоненты прикладного уровня, подключаемые к транспортным средствам и RSU. Они обмениваются типизированными оболочками TransportMessage, предоставляют снимки состояния и выполняются в детерминированном порядке на каждом такте.

Этот уровень сервисов также служит точкой подключения фреймворка атак. Этапы атак не изменяют менеджеры напрямую. Они оборачивают привязки возможностей, предоставляемые сервисами поведения. Архитектура со стороны атак описана в Фреймворк атак. Текущий каталог встроенных сервисов приведён в Доступные сервисы.

Что представляет собой сервис

Сервис поведения - любой объект, реализующий протокол BehaviorService из opencda/core/application/behavior/behavior_service_protocol.py.

Каждый сервис должен определять:

• service_type: стабильный строковый идентификатор, используемый в конфигурации и маршрутизации сообщений

• priority: целочисленный порядок выполнения; меньшие значения выполняются раньше

• capability_bindings: предоставляемая карта соответствий возможностей и вызываемых объектов

• on_attach(owner): хук инициализации для конкретного владельца

• process(messages): точка входа обработки сообщений на каждом такте

• get_state(): неизменяемый снимок состояния для проверки во время выполнения

• on_detach(): хук очистки

Во время выполнения владельцем является VehicleManager или RSUManager.

Регистрация и обнаружение

Пакеты встроенных сервисов находятся в:

opencda/core/application/behavior/services/

Загрузчик уровня пакета импортирует все модули встроенных сервисов при запуске, а конкретные сервисы самостоятельно регистрируются через @BehaviorServiceRegistry.register. Чтобы сервис можно было создать, объявите уникальный service_type и импортируйте его модуль.

Менеджеры создают сервисы из YAML сценария:

behavior_services:
  - type: self_informer
  - type: aim_client
    debug: true
  - type: movement_controller

Поле type разрешается через BehaviorServiceRegistry, а экземпляр создаётся с помощью create_service(...).

Жизненный цикл на стороне менеджера

VehicleManager и RSUManager используют одинаковый жизненный цикл:

1. считать behavior_services из конфигурации

2. создать экземпляры сервисов через реестр

3. проверить, что каждый сервис реализует протокол

4. сортировать сервисы по priority

5. вызвать on_attach(...) для каждого сервиса

6. на каждом такте симуляции выполнять update_behavior_services(...)

7. при завершении вызывать on_detach() в обратном порядке

Оба менеджера используют одинаковый поток выполнения на каждом такте:

1. проверять входящие объекты TransportMessage

2. оставлять только сообщения, адресованные текущему узлу, и допустимые широковещательные сообщения

3. группировать сообщения по dst_service_type

4. вызвать process(...) для каждого подключённого сервиса в порядке приоритета

5. немедленно возвращать адресованные себе выходные данные в тот же такт

6. сохранять нелокальные выходные данные в behavior_service_results

7. сохранять последний результат get_state() в behavior_service_states

Этот общий конвейер - основная причина, по которой новые сервисы должны следовать общей структуре реализации. Если каждый сервис использует произвольный поток управления, становится сложнее анализировать приоритеты, маршрутизацию сообщений, сбор состояния и перехват атак.

Общая модель сообщений

Сервисы обмениваются данными через типизированные транспортные оболочки:

@dataclass(frozen=True)
class TransportMessage(Generic[payloadT]):
    src_owner_id: str
    src_service_type: str
    dst_owner_id: str
    dst_service_type: str
    payload: payloadT

Важные константы маршрутизации:

• BROADCAST_OWNER_ID: получатель широковещательной рассылки на уровне узла

• BROADCAST_SERVICE_TYPE: получатель широковещательной рассылки на уровне сервиса

Она предоставляет сервисам единообразный способ:

• публиковать ответы с состоянием для соседних сервисов

• отправлять команды другому локальному сервису на том же узле

• отправлять запросы или ответы сервисам на других узлах

Возможности

Каждый сервис предоставляет карту capability_bindings с ключами из общего словаря возможностей:

• request.observe

• request.submit

• response.observe

• response.submit

• command.submit

• state.observe

Эти привязки служат двум целям:

• они документируют видимые извне точки взаимодействия сервиса

• они предоставляют стабильные хуки, через которые фреймворк атак может наблюдать или изменять поведение сервиса

Если у сервиса нет значимых операций для внешнего подключения, он может предоставлять пустую карту привязок, как сейчас делает movement_controller.

Рекомендуемый шаблон разработки

Новые сервисы следует реализовывать примерно по одному внутреннему шаблону, даже если их прикладная логика различается. Цель заключается не в одинаковом стиле кода как таковом, а в предсказуемой семантике выполнения на всём уровне сервисов.

Рекомендуемая структура пакета:

services/<service_name>/
  __init__.py
  service.py
  messages.py
  types.py
  utils.py

Рекомендуемый каркас класса:

@BehaviorServiceRegistry.register
class ExampleService:
    service_type = "example_service"
    priority = 50

    @property
    def capability_bindings(self) -> CapabilityBindings:
        return {
            Capability.REQUEST_OBSERVE: self._observe_requests,
            Capability.RESPONSE_SUBMIT: self._build_responses,
            Capability.STATE_OBSERVE: self.get_state,
        }

    def __init__(self, priority: int = 50, **config: Any) -> None:
        self.priority = priority
        self._owner_ref = None
        self._runtime_state = None

    def on_attach(self, owner: Any) -> None:
        self._owner_ref = weakref.ref(owner)

    def get_state(self) -> ExampleServiceState:
        return ExampleServiceState(...)

    def process(
        self,
        messages: Sequence[TransportMessage[ExampleRequest]],
    ) -> tuple[TransportMessage[ExampleResponse], ...]:
        observed = self._observe_requests(messages)
        self._update_runtime_state(observed)
        return self._build_responses(observed)

    def on_detach(self) -> None:
        self._owner_ref = None

Этапы, общие для всех сервисов

Новые сервисы должны проходить одинаковые концептуальные этапы обработки на каждом такте, даже если некоторые этапы тривиальны для конкретной реализации:

1. фильтрация входных данных

2. наблюдение или декодирование соответствующих сообщений

3. обновление внутреннего состояния или принятие решения

4. формирование выходных данных или отправка команды

5. предоставление снимка состояния

На практике обычно это означает:

• отделять выбор сообщений от прикладной логики

• обновлять локальные поля времени выполнения до создания исходящих сообщений

• возвращать типизированные объекты TransportMessage вместо изменения общего состояния

• обеспечивать, чтобы get_state() отражал последний значимый снимок сервиса

Общая модель этапов важна по трём причинам:

• она сохраняет предсказуемость выполнения нескольких сервисов при разных приоритетах

• она обеспечивает единообразный перехват атак, поскольку привязки всегда оборачивают известные этапы

• она упрощает реализацию метрик, отладку и добавление новых сервисов

Встроенные сервисы как эталонные реализации

Текущие встроенные сервисы уже следуют этому шаблону с разным уровнем сложности.

Полный каталог встроенных сервисов приведён в Доступные сервисы.

self_informer

• наблюдает текущие позу и скорость владельца

• обновляет кешированные локальные поля

• отправляет широковещательный SelfInformerResponse

• предоставляет компактный снимок состояния владельца

movement_controller

• фильтрует локальные запросы управления

• выбирает последнюю допустимую команду движения

• обновляет локальное целевое состояние

• вызывает owner.control(...)

• не отправляет исходящие сообщения

aim_client

• наблюдает ответы сервера AIM и локальные данные self-informer

• обновляет текущую траекторию управления

• отправляет команды контроллера движения

• отправляет следующий запрос в aim_server

• предоставляет активную траекторию как состояние сервиса

aim_server

• наблюдает входящие запросы AIM

• перенаправляет их в AIMModelManager

• отправляет сообщения ответов AIM

• предоставляет состояние отслеживаемых транспортных средств через get_state()

Примеры конфигурации

Конвейер транспортного средства по умолчанию:

vehicle_base:
  behavior_services:
    - type: self_informer
    - type: movement_controller

Настройка в стиле AIM с сервисами транспортных средств и RSU:

scenario:
  rsu_list:
    - id: 1
      behavior_services:
        - type: aim_server
          priority: 1

  single_cav_list:
    - id: 100
      behavior_services:
        - type: self_informer
        - type: aim_client
          debug: true
        - type: movement_controller

Правила реализации

При добавлении нового сервиса сохраняйте следующие правила, если нет веской архитектурной причины отступить от них:

• использовать один уникальный service_type для каждого сервиса

• задавать priority явно и осмысленно

• предоставлять стабильные capability_bindings для этапов, которые должны быть наблюдаемыми

• обеспечивать детерминированность process(...) для одинаковых входных данных

• сохранять get_state() неизменяемым и нетребовательным к ресурсам при чтении

• предпочитать вспомогательные методы для каждого этапа обработки одному монолитному process(...)

• очистить всё состояние, зависящее от владельца, в on_detach()

Иными словами, сервисы должны различаться логикой, а не структурой жизненного цикла. Эта согласованность обеспечивает компонуемость фреймворка сервисов и синхронизирует фреймворк атак, метрики и среду выполнения сценария.