j ^ddlZddlZddlmZmZddlmZdZGddejZ y)N)ListCallable)messages2c *eZdZefdeej gdfdededefdZ ddZ de fd Z de fd Z ejdeej fd Zejd ej ddfd Zdeej ge fddfdZdefdZy) BaseBatcherflush_callbackNmax_batch_sizeflush_interval_secondsbatch_memory_limit_mbc|||_||_g|_||_||_t j |_yN)_flush_interval_seconds_flush_callback_accumulated_messages_max_batch_size_batch_memory_limit_mbtime _last_time_flush_callback_called)selfr r r r s /Users/manta/Documents/Projects/TheRoad-I1/backend/.venv/lib/python3.12/site-packages/opik/message_processing/batching/base_batcher.py__init__zBaseBatcher.__init__ s;/E$GUAC"$2+@#7;yy{-returnct|jdkDr/|j}g|_|D]}|j|t j|_yNr)lenr)_create_batches_from_accumulated_messagesrrr)rbatch_messages batch_messages rflushzBaseBatcher.flushsQ t)) *Q .!KKMN)+D &!/ $$]3"004 -rcbtj|jz }||jk\Sr)rrr)relapseds ris_ready_to_flushzBaseBatcher.is_ready_to_flush$s)))+ E EE$6666rc2t|jdk(Srrrrs ris_emptyzBaseBatcher.is_empty(s4--.!33rcyrr(s rrz5BaseBatcher._create_batches_from_accumulated_messages+s&)rmessagec|jj|t|j|jk\r|j yyr)rappendrrr")rr,s raddzBaseBatcher.add0s> ""))'2 t)) *d.B.B B JJL Cr filter_funccTttfd|j|_y)z Remove messages from _accumulated_messages that match the provided filter function. Args: filter_func: A function that takes a BaseMessage and returns True if the message should be removed c| Srr+)xr0s rz7BaseBatcher._remove_matching_messages..@s Q/rN)listfilterr)rr0s `r_remove_matching_messagesz%BaseBatcher._remove_matching_messages6s#&* /1K1K L& "rc,t|jS)ap Gets the total number of accumulated messages in the current instance. The method calculates and retrieves the count of accumulated messages maintained within an internal structure. Thread safety is guaranteed through the usage of a lock mechanism. Returns: int: The total number of accumulated messages. r'r(s rsizezBaseBatcher.sizeCs4--..r)rN)__name__ __module__ __qualname__BASE_BATCH_MEMORY_LIMIT_MBrr BaseMessageintfloatrr"boolr%r)abcabstractmethodrrr/r7r9r+rrr r s &@ C ("6"6!7!=> C C!& C # C<7474$4 ) h"" #)) 8//D  #X%9%9$:D$@A     /c /rr ) rrBtypingrrrr=ABCr r+rrrGs+ ! C/#''C/r