Loading src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py +23 −1 Original line number Diff line number Diff line from uuid import UUID from typing import Literal from uuid import UUID, uuid4 from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import CreateSession from open_exposure_gateway.domain.quality_on_demand import ( EventQosStatusChangedData, NetworkCapabilityDeactivateTarget, NetworkCapabilityParameters, NetworkCapabilityPayload, NetworkCapabilityTarget, NetworkCapabilityTargetApplicationServer, NetworkCapabilityTargetDevice, QosStatusChangedCloudEvent, SRMNetworkCapabilityActivateCommand, SRMNetworkCapabilityDeactivateCommand, ) Loading Loading @@ -69,3 +72,22 @@ def build_deactivate_command( grace_period_seconds=grace_period_seconds, ), ) def build_qos_status_changed_event( session_id: UUID, qos_status: Literal["AVAILABLE", "UNAVAILABLE"], status_info: Literal["DURATION_EXPIRED", "NETWORK_TERMINATED", "DELETE_REQUESTED"] | None, source: str, occurred_at: str, ) -> QosStatusChangedCloudEvent: return QosStatusChangedCloudEvent( id=uuid4(), source=source, time=occurred_at, data=EventQosStatusChangedData( sessionId=session_id, qosStatus=qos_status, statusInfo=status_info, ), ) Loading
src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py +23 −1 Original line number Diff line number Diff line from uuid import UUID from typing import Literal from uuid import UUID, uuid4 from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import CreateSession from open_exposure_gateway.domain.quality_on_demand import ( EventQosStatusChangedData, NetworkCapabilityDeactivateTarget, NetworkCapabilityParameters, NetworkCapabilityPayload, NetworkCapabilityTarget, NetworkCapabilityTargetApplicationServer, NetworkCapabilityTargetDevice, QosStatusChangedCloudEvent, SRMNetworkCapabilityActivateCommand, SRMNetworkCapabilityDeactivateCommand, ) Loading Loading @@ -69,3 +72,22 @@ def build_deactivate_command( grace_period_seconds=grace_period_seconds, ), ) def build_qos_status_changed_event( session_id: UUID, qos_status: Literal["AVAILABLE", "UNAVAILABLE"], status_info: Literal["DURATION_EXPIRED", "NETWORK_TERMINATED", "DELETE_REQUESTED"] | None, source: str, occurred_at: str, ) -> QosStatusChangedCloudEvent: return QosStatusChangedCloudEvent( id=uuid4(), source=source, time=occurred_at, data=EventQosStatusChangedData( sessionId=session_id, qosStatus=qos_status, statusInfo=status_info, ), )