o tf @sddlmZddlmZddlmZmZmZmZm Z m Z er5ddl Z ddl m Z e dZe dZe dZne dZGd d d eeZGd d d eeeZGd ddeeeZ   dd ddZ   d!d"ddZdS)#) annotations)Timer) TYPE_CHECKINGAnyCallableGenericLiteralTypeVarN) ParamSpec)Z throttlerZ debouncer)trailingleadingPc@speZdZUded<  d#d$d dZd%ddZd%ddZd%ddZd%ddZd%ddZ d&ddZ e d'd d!Z d"S)(_ThrottlerBaser_timerdr funcCallable[P, Any]intervalintpolicyEmissionPolicyreturnNonecCs||_|d|_||_d|_tddd|_|jd|_i|_t |dd|_ t |d d|_ t |d d|_ t |d d|_ t |d i|_dS) NiFrcSsdSNrrr[/var/www/html/software/conda/envs/catlas/lib/python3.10/site-packages/psygnal/_throttler.py"sz)_ThrottlerBase.__init__..r __module____name__ __qualname____doc____annotations__) __wrapped__ _interval_policy _has_pendingrrstart_args_kwargsgetattrrrr r!r"selfrrrrrr__init__s  z_ThrottlerBase.__init__cCs&d|_|j|ji|j|dS)NF)r&r#r(r) _start_timerr,rrr_actually_call0s z_ThrottlerBase._actually_callcCs|jr |dSdSr)r&r0r/rrr_call_if_has_pending5s z#_ThrottlerBase._call_if_has_pendingcCs(|jt|j|j|_|jdSr)rcancelrr$r1r'r/rrrr.9s z_ThrottlerBase._start_timercCsd|_|jdS)zCancel any pending calls.FN)r&rr2r/rrrr2>sz_ThrottlerBase.cancelcCs |dS)z%Force a call if there is one pending.N)r1r/rrrflushCs z_ThrottlerBase.flushargsP.argskwargsP.kwargscOstd)Nz&Subclasses must implement this method.)NotImplementedErrorr,r4r6rrr__call__Gsz_ThrottlerBase.__call__inspect.SignaturecCsddl}||jS)Nr)inspect signaturer#)r,r<rrr __signature__Js z_ThrottlerBase.__signature__Nrr rrrrrrrr)rrr4r5r6r7rr)rr;) rrr r"r-r0r1r.r2r3r:propertyr>rrrrrs       rc<eZdZUdZded<  ddfdd ZdddZZS) ThrottleraClass that prevents calling `func` more than once per `interval`. Parameters ---------- func : Callable[P, Any] a function to wrap interval : int, optional the minimum interval in ms that must pass before the function is called again, by default 100 policy : EmissionPolicy, optional Whether to invoke the function on the "leading" or "trailing" edge of the wait timer, by default "leading" rrrr rrrrrrrrct|||dSrsuperr-r+ __class__rrr-bzThrottler.__init__r4r5r6r7cOsBd|_||_||_|js|jdkr|dS|dSdSzCall underlying function.Tr Nr&r(r)ris_aliver%r0r.r9rrrr:js    zThrottler.__call__r?r@rArrr r!r"r-r: __classcell__rrrHrrDQ rDcrC) DebounceraClass that waits at least `interval` before calling `func`. Parameters ---------- func : Callable[P, Any] a function to wrap interval : int, optional the minimum interval in ms that must pass before the function is called again, by default 100 policy : EmissionPolicy, optional Whether to invoke the function on the "leading" or "trailing" edge of the wait timer, by default "trailing" rrrr rrrrrrrrcrErrFr+rHrrr-rJzDebouncer.__init__r4r5r6r7cOs:d|_||_||_|js|jdkr||dSrKrLr9rrrr:s  zDebouncer.__call__)rr r@rArNrrrHrrQwrPrQrTrCallable[P, Any] | Nonetimeoutrr boolr9Throttler[P] | Callable[[Callable[P, Any]], Throttler[P]]c$dfdd }|dur||S|S) a$Create a throttled function that invokes func at most once per timeout. The throttled function comes with a `cancel` method to cancel delayed func invocations and a `flush` method to immediately invoke them. Options to indicate whether func should be invoked on the leading and/or trailing edge of the wait timeout. The func is invoked with the last arguments provided to the throttled function. Subsequent calls to the throttled function return the result of the last func invocation. This decorator may be used with or without parameters. Parameters ---------- func : Callable A function to throttle timeout : int Timeout in milliseconds to wait before allowing another call, by default 100 leading : bool Whether to invoke the function on the leading edge of the wait timer, by default True Examples -------- ```python from psygnal import Signal, throttled class MyEmitter: changed = Signal(int) def on_change(val: int) # do something possibly expensive ... emitter = MyEmitter() # connect the `on_change` whenever `emitter.changed` is emitted # BUT, no more than once every 50 milliseconds emitter.changed.connect(throttled(on_change, timeout=50)) ``` rrr Throttler[P]crdnd}t||SNr r )rDrrr rSrrdeco  zthrottled..decoN)rrrrWrrrSr r\rr[r throttleds.r_F9Debouncer[P] | Callable[[Callable[P, Any]], Debouncer[P]]crV) aCreate a debounced function that delays invoking `func`. `func` will not be invoked until `timeout` ms have elapsed since the last time the debounced function was invoked. The debounced function comes with a `cancel` method to cancel delayed func invocations and a `flush` method to immediately invoke them. Options indicate whether func should be invoked on the leading and/or trailing edge of the wait timeout. The func is invoked with the *last* arguments provided to the debounced function. Subsequent calls to the debounced function return the result of the last `func` invocation. This decorator may be used with or without parameters. Parameters ---------- func : Callable A function to throttle timeout : int Timeout in milliseconds to wait before allowing another call, by default 100 leading : bool Whether to invoke the function on the leading edge of the wait timer, by default False Examples -------- ```python from psygnal import Signal, debounced class MyEmitter: changed = Signal(int) def on_change(val: int) # do something possibly expensive ... emitter = MyEmitter() # connect the `on_change` whenever `emitter.changed` is emitted # ONLY once at least 50 milliseconds have passed since the last signal emission. emitter.changed.connect(debounced(on_change, timeout=50)) ``` rrr Debouncer[P]crXrY)rQrZr[rrr\r]zdebounced..decoN)rrrrarr^rr[r debounceds1rb)NrT)rrRrSrr rTrrU)NrF)rrRrSrr rTrr`) __future__r threadingrtypingrrrrrr r<Ztyping_extensionsr Kindrr rrDrQr_rbrrrrs*    <&% 6