DhO3T dZddlZddlZddlZddlZddlZddlZddlmZmZm Z e jdZ e jdZ ejdddZd ejd e j fd Zd e j$d e j fd Zd e j(e j$d e j(e j fdZejdfde j,dejd efdZe j2de j4e j6ge j8e fe j8e fd e j8e fdZe j2de j4e j6ge j<e fe j<e fd e j<e fdZde j4e j<e e j6ge j<e fe j8e e j6ge j8e ffd e j4e j<e e j8e ffdZejdej>ddfde j$d e j$de j4e j<e e j6ge j<e ffde de j(e j$d e j<e f dZ!ejdejDddfde j$d e j$de j4e j8e e j6de j8e ffde de j(e j$d e jFe dff dZ$ddejJfde jFe dfde j(e j$de j(e j$d e j(e j6e jFe dfe j(e j$e j(e j$e&ge jNfd!e jPe jRe jNfd e jFe dff d"Z*ddejJfde j(e j$de j(e j$d e j(e j6e jFe jNdfe j(e j$e j(e j$e&ge jNfd!e jPe jRe jNfd e j6e j6e e jFe dffge j6e e jFe dffff d#Z+y)$a# This module provides utility functions for handling time-related operations. Functions: - timedelta_to_seconds: Convert a timedelta to seconds with microseconds as fraction. - delta_to_seconds: Convert a timedelta or numeric interval to seconds. - delta_to_seconds_or_none: Convert a timedelta to seconds or return None. - format_time: Format a timestamp (timedelta, datetime, or seconds) to a string. - timeout_generator: Generate items from an iterable until a timeout is reached. - aio_timeout_generator: Asynchronously generate items from an iterable until a timeout is reached. - aio_generator_timeout_detector: Detect if an async generator has not yielded an element for a set amount of time. - aio_generator_timeout_detector_decorator: Decorator for aio_generator_timeout_detector. N)aio exceptionstypes_T_Pi)yearmonthdaydeltareturnc|jr|jdz}nd}||jz }||jdzdzdzz }|S)aConvert a timedelta to seconds with the microseconds as fraction. Note that this method has become largely obsolete with the `timedelta.total_seconds()` method introduced in Python 2.7. >>> from datetime import timedelta >>> '%d' % timedelta_to_seconds(timedelta(days=1)) '86400' >>> '%d' % timedelta_to_seconds(timedelta(seconds=1)) '1' >>> '%.6f' % timedelta_to_seconds(timedelta(seconds=1, microseconds=1)) '1.000001' >>> '%.6f' % timedelta_to_seconds(timedelta(microseconds=1)) '0.000001' gư>r<) microsecondssecondsdays)r totals s/var/www/fastuser/data/www/generator.snapmosaic.io/flask_app/venv/lib/python3.12/site-packages/python_utils/time.pytimedelta_to_secondsr(sQ" ""T) U]]E UZZ"_r !B &&E Lintervalct|tjr t|St|tt fr|St dt|d|)am Convert a timedelta to seconds. >>> delta_to_seconds(datetime.timedelta(seconds=1)) 1 >>> delta_to_seconds(datetime.timedelta(seconds=1, microseconds=1)) 1.000001 >>> delta_to_seconds(1) 1 >>> delta_to_seconds('whatever') # doctest: +ELLIPSIS Traceback (most recent call last): ... TypeError: Unknown type ... Unknown type : ) isinstancedatetime timedeltarintfloat TypeErrortypers rdelta_to_secondsr$BsO(H../#H-- HsEl +-X'7r(FGGrc|yt|S)z.Convert a timedelta to seconds or return None.N)r$r#s rdelta_to_seconds_or_noner&Ys))rr timestamp precisioncf|j}t|tr t|}t|ttfr t j |}t|t j r7|j}|||zz }tt j |St|t j rbt|dr|j}nt|tz }|||zz } t j j|}t|St|t j r t|S|yt#dt%|d|#t$rd}YwxYw#ttf$r't j j}Yt|SwxYw)a/Formats timedelta/datetime/seconds. >>> format_time('1') '0:00:01' >>> format_time(1.234) '0:00:01' >>> format_time(1) '0:00:01' >>> format_time(datetime.datetime(2000, 1, 2, 3, 4, 5, 6)) '2000-01-02 03:04:05' >>> format_time(datetime.date(2000, 1, 2)) '2000-01-02' >>> format_time(datetime.timedelta(seconds=3661)) '1:01:01' >>> format_time(None) '--:--:--' >>> format_time(format_time) # doctest: +ELLIPSIS Traceback (most recent call last): ... TypeError: Unknown type ... r'Nr(z--:--:--rr) total_secondsrstrr rrr OverflowErrorhasattrr(repoch fromtimestamp ValueErrorOSErrormaxdater!r")r(r)precision_secondsrdts r format_timer7cs4"//1)S!)$ )c5\*  **9=I)X//0))+W'8898%%g677 Ix00 1 9k *))+G*9u+<=GW'889 '""009B2w Ix}} -9~  -Y'89-HII; I *G$ '""&&B2w 's$E(<E:( E76E7:)F0/F0iterablecyNr8s r _to_iterabler=s "rcyr:r;r<s rr=r=s rc*t|r|S|Sr:)callabler<s rr=r=szrg?timeoutinterval_multipliermaximum_intervalc#(Kt|}t|}t|}t|tjz}|D]H} | tj|k\rytj |||z}|s=t ||}Jyw)a Generator that walks through the given iterable (a counter by default) until the float_timeout is reached with a configurable float_interval between items. This can be used to limit the time spent on a slow operation. This can be useful for testing slow APIs so you get a small sample of the data in a reasonable amount of time. >>> for i in timeout_generator(0.1, 0.06): ... # Put your slow code here ... print(i) 0 1 2 >>> timeout = datetime.timedelta(seconds=0.1) >>> interval = datetime.timedelta(seconds=0.06) >>> for i in timeout_generator(timeout, interval, itertools.count()): ... print(i) 0 1 2 >>> for i in timeout_generator(1, interval=0.1, iterable='ab'): ... print(i) a b >>> timeout = datetime.timedelta(seconds=0.1) >>> interval = datetime.timedelta(seconds=0.06) >>> for i in timeout_generator(timeout, interval, interval_multiplier=2): ... print(i) 0 1 2 N)r$r&r=time perf_countersleepmin rArr8rBrCfloat_intervalfloat_maximum_interval iterable_enditems rtimeout_generatorrOsX-X6N4L5X&I 7 #d&7&7&9 9C I    # %  >"-- ! 1GHN Is BBB.cNKt|}t|}t|}t|tjz}|23d{} | tj|k\ryt j |d{||z}|sLt||}Y7T76yw)as Async generator that walks through the given async iterable (a counter by default) until the float_timeout is reached with a configurable float_interval between items. The interval_exponent automatically increases the float_timeout with each run. Note that if the float_interval is less than 1, 1/interval_exponent will be used so the float_interval is always growing. To double the float_interval with each run, specify 2. Doctests and asyncio are not friends, so no examples. But this function is effectively the same as the `timeout_generator` but it uses `async for` instead. N)r$r&r=rErFasynciorGrHrIs raio_timeout_generatorrRs.-X6N4L5X&I 7 #d&7&7&9 9C I Id    # % mmN+++-- ! 1GHN I ,  sBAB%B# B B#7B%B! B% B%B#!B%#B% generator total_timeout on_timeouton_timeout_kwargsc K|d}n tjt|z}tj|} |r,tj|k\rt j d|r0t j|j|d{n|jd{}7$7 #t j $r!}||||||fi|d{7Yd}~yd}~wt$rYywxYww)a This function is used to detect if an asyncio generator has not yielded an element for a set amount of time. The `on_timeout` argument is called with the `generator`, `timeout`, `total_timeout`, `exception` and the extra `**kwargs` to this function as arguments. If `on_timeout` is not specified, the exception is reraised. If `on_timeout` is `None`, the exception is silently ignored and the generator will finish as normal. NzTotal timeout reached) rErFr$ python_utilsr&rQ TimeoutErrorwait_for __anext__StopAsyncIteration)rSrArTrUrVtotal_timeout_end timeout_s exceptions raio_generator_timeout_detectorr`/s8  --/2B 3  55g>I   T%6%6%8._timeout_detector_decorator..wrappers2 24*6*  $  r) functoolswrapsrrcrdrAsyncGeneratorr)rSrfrUrVrArTs` r_timeout_detector_decoratorzMaio_generator_timeout_detector_decorator.._timeout_detector_decorators]  # 77 ii  ! !"d( +   $ r)rCallablerrir)rArTrUrVrjs```` r(aio_generator_timeout_detector_decoratorrlos_*>>"e&:&:2t8&D"DE E00T:: ;( '&r),__doc__rQrrg itertoolsrErXrrrTypeVarr ParamSpecrr/rNumberr delta_typer$Optionalr&timestamp_typer,r7overloadUnionrk AsyncIterabler=Iterablecountr rOacountrirRreraise BaseExceptionAnyMappingTextr`rlr;rrrs(* //U]]4U__T t1!4 2 2u||4Hu//HELLH.*nnU--.* ^^ELL!*%7H$6$6q$A?J##?J!!?J ?JD"kk r5..r223 B !"   ""kk r5>>"--.r0BB ^^B  kk r r5>>"--. B r5..r223 5  [[#U%8%8%<<=  "4!3!3A!> !$9==I   =I=Ikk rENN2u~~b/A+ABB=I  =InnU%5%56=I ^^B=ID"4!3!3A!> !$9=(I   (I(Ikk BU5H5H5L0L!MM(I  (InnU%5%56(I "d(#(IZ156: =##BH-= ^^E,, -=>>%"2"23= $$RX.u//0u//0   II     =uzz599'<==  "d(#!=B156: )' ^^E,, -)'>>%"2"23)' $$UYY_5u//0u//0   II    )'uzz599'<=)' ^^ ^^B,,RX6678 NN2u++BH5568)'r