qik/dZddlmZmZmZmZmZmZddlm Z ddl m Z m Z m Z mZmZmZddlmZddlmZddlmZmZddlmZGd d eZGd d eZGd deZedeZGddeeeZy)zjService switcher for switching between different services at runtime, with different switching strategies.)AnyGenericListOptionalTypeTypeVar)logger) ErrorFrameFrameManuallySwitchServiceFrameServiceMetadataFrameServiceSwitcherFrame#ServiceSwitcherRequestMetadataFrame)ParallelPipeline)FunctionFilter)FrameDirectionFrameProcessor) BaseObjectceZdZdZdeeffd ZedeefdZedefdZ de de de efd Z d ede efd Zd ede efd ZxZS)ServiceSwitcherStrategyaBase class for service switching strategies. Note: Strategy classes are instantiated internally by ServiceSwitcher. Developers should pass the strategy class (not an instance) to ServiceSwitcher. Event handlers available: - on_service_switched: Called when the active service changes. Example:: @strategy.event_handler("on_service_switched") async def on_service_switched(strategy, service): ... servicesct|t|dk(r td||_|d|_|j dy)aInitialize the service switcher strategy with a list of services. Note: This is called internally by ServiceSwitcher. Do not instantiate directly. Args: services: List of frame processors to switch between. rz2ServiceSwitcherStrategy needs at least one serviceon_service_switchedN)super__init__len Exception _services_active_service_register_event_handler)selfr __class__s S/opt/pipecat/venv/lib/python3.12/site-packages/pipecat/pipeline/service_switcher.pyrz ServiceSwitcherStrategy.__init__-sJ  x=A PR R!'{ $$%:;returnc|jSz&Return the list of available services.rr!s r#rz ServiceSwitcherStrategy.services@~~r$c|jS)z$Return the currently active service.)rr)s r#active_servicez&ServiceSwitcherStrategy.active_serviceEs###r$frame directionc Kyw)aHandle a frame that controls service switching. The base implementation returns ``None`` for all frames. Subclasses override this to implement specific switching behaviors. Args: frame: The frame to handle. direction: The direction of the frame (upstream or downstream). Returns: The newly active service if a switch occurred, or None otherwise. Nr!r-r.s r# handle_framez$ServiceSwitcherStrategy.handle_frameJs errorc Kyw)aHandle an error from the active service. Called by ``ServiceSwitcher`` when a non-fatal ``ErrorFrame`` is pushed upstream by the currently active service. Subclasses can override this to implement automatic failover. Args: error: The error frame pushed by the active service. Returns: The newly active service if a switch occurred, or None otherwise. Nr0)r!r4s r# handle_errorz$ServiceSwitcherStrategy.handle_error[s r3servicecK||jvrF||_|jt|d{|j d|d{|Sy7!7 w)aSet the active service to the given one, if it is in the list of available services. If it's not in the list, the request is ignored, as it may have been intended for another ServiceSwitcher in the pipeline. Args: service: The service to set as active. Returns: The newly active service, or None if the service was not found. )r7Nr)rr queue_framer_call_event_handler)r!r7s r#_set_active_if_availablez0ServiceSwitcherStrategy._set_active_if_availablejs` dmm ##*D %%&IRY&Z[ [ [**+@'J J JN \ Js!4AAAAAA)__name__ __module__ __qualname____doc__rrrpropertyrr,rrrr2r r6r; __classcell__r"s@r#rrs"<n!5<&$~.$$$)6D . !"  x7O nR`Iar$rc*eZdZdZdededeefdZy)ServiceSwitcherStrategyManualacA strategy for switching between services manually. This strategy allows the user to manually select which service is active. The initial active service is the first one in the list. Example:: stt_switcher = ServiceSwitcher( services=[stt_1, stt_2], strategy_type=ServiceSwitcherStrategyManual ) r-r.r%cvKt|tr#|j|jd{Sy7w)aHandle a frame that controls service switching. Args: frame: The frame to handle. direction: The direction of the frame (upstream or downstream). Returns: The newly active service if a switch occurred, or None otherwise. N) isinstancer r;r7r1s r#r2z*ServiceSwitcherStrategyManual.handle_frames4 e7 866u}}EE EFs /979N) r<r=r>r?rrrrr2r0r$r#rDrD~s) )6D . !r$rDc&eZdZdZdedeefdZy)ServiceSwitcherStrategyFailoveraA strategy that automatically switches to a backup service on failure. When the active service produces a non-fatal error, this strategy switches to the next available service in the list. Recovery and fallback policies are left to application code via the ``on_service_switched`` event. Event handlers available: - on_service_switched: Called when the active service changes. Example:: switcher = ServiceSwitcher( services=[primary_stt, backup_stt], strategy_type=ServiceSwitcherStrategyFailover, ) @switcher.strategy.event_handler("on_service_switched") async def on_switched(strategy, service): # App decides when/how to recover the failed service ... r4r%cKtjd|jjd|jt |j dkrtjdy|j j|j}|dzt |j z}|j|j |d{S7w)aHandle an error from the active service by failing over. Switches to the next service in the list. The failed service remains in the list and can be switched back to manually or via application logic in the ``on_service_switched`` event handler. Args: error: The error frame pushed by the active service. Returns: The newly active service if a switch occurred, or None if no other service is available. zService z reported an error: z'No other service available to switch toN) r warningrnamer4rrindexr;)r!r4 current_idxnext_idxs r#r6z,ServiceSwitcherStrategyFailover.handle_errors $"6"6";";!<>'::224>>(3KLLLLsC C C CN)r<r=r>r?r rrr6r0r$r#rHrHs".M Mx7OMr$rH StrategyType)boundceZdZdZefdeedeeffd Z e defdZ e deefdZ e deededeefd Ze d ededefd Zej&fd ed effd Zd ed effd ZxZS)ServiceSwitcheraParallel pipeline that routes frames to one active service at a time. Wraps each service in a pair of filters that gate frame flow based on which service is currently active. Switching is controlled by `ServiceSwitcherFrame` frames and delegated to a pluggable `ServiceSwitcherStrategy`. Example:: switcher = ServiceSwitcher(services=[stt_1, stt_2]) r strategy_typecj||}t||j||||_||_y)a?Initialize the service switcher with a list of services and a switching strategy. Args: services: List of frame processors to switch between. strategy_type: The strategy class to use for switching between services. Defaults to ``ServiceSwitcherStrategyManual``. N)rr_make_pipeline_definitionsr _strategy)r!rrTrWr"s r#rzServiceSwitcher.__init__s7"(+  $99(INO!"r$r%c|jS)z%Return the active switching strategy.)rWr)s r#strategyzServiceSwitcher.strategyr*r$c|jSr'r(r)s r#rzServiceSwitcher.servicesr*r$rYcbg}|D]'}|jtj||)|SN)appendrS_make_pipeline_definition)rrY pipelinesr7s r#rVz*ServiceSwitcher._make_pipeline_definitionss<  [G   _FFwPXY Z [r$r7cdtdtffd }t|tjddt|tj ddgS)N_r%c*Kjk(Swr\)r,)rar7rYs r#filterz9ServiceSwitcher._make_pipeline_definition..filtersh555 5sT)rcr.filter_system_framesenable_direct_mode)r boolrr DOWNSTREAMUPSTREAM)r7rYrcs`` r#r^z)ServiceSwitcher._make_pipeline_definitions[ 6E 6d 6 (33%)#'    (11%)#'    r$r-r.cKt|tr$|j|jjk(ryt|t r.|j |jjjk7ryt|tr/|js#|jj|d{t|1||d{y77w)a Push a frame out of the service switcher. Suppresses `ServiceSwitcherRequestMetadataFrame` targeting the active service (since it has already been handled) and `ServiceMetadataFrame` from inactive services so only the active service's metadata reaches downstream processors. One case this happens is with `StartFrame` since all the filters let it pass, and `StartFrame` causes the service to generate `ServiceMetadataFrame`. Non-fatal ``ErrorFrame`` instances are forwarded to the strategy via ``handle_error`` so strategies like ``ServiceSwitcherStrategyFailover`` can perform failover. The error frame is still propagated upstream so that application-level error handlers can observe it. N) rFrr7rYr,r service_namerLr fatalr6r push_frame)r!r-r.r"s r#rlzServiceSwitcher.push_frame!s" e@ A}} < << e1 2!!T]]%A%A%F%FF eZ (--,,U3 3 3g  222 42s$B-C0C1CC CCcKt|tr@|jj||d{}|st|||d{yyt|||d{y797!7 w)zProcess a frame, handling frames which affect service switching. Args: frame: The frame to process. direction: The direction of the frame (upstream or downstream). N)rFrrYr2r process_frame)r!r-r.r7r"s r#rnzServiceSwitcher.process_frameAss e1 2 MM66uiHHGg+E9===''y9 9 9I > 9s30A3A-A3 A/A3'A1(A3/A31A3)r<r=r>r?rDrrrrPrr@rYr staticmethodrrrVr^rrgr rlrnrArBs@r#rSrSs -J#~&#L)#",$~.~&2I c  +B   :JXIbIb3e33@::>::r$rSN)r?typingrrrrrrlogurur pipecat.frames.framesr r r r rr"pipecat.pipeline.parallel_pipeliner*pipecat.processors.filters.function_filterr"pipecat.processors.frame_processorrrpipecat.utils.base_objectrrrDrHrPrSr0r$r#rwsq>>@EM0`j`F$;@.M&C.Mb~-DE ~:& (=~:r$