o
    hR                     @  s^   d Z ddlmZ ddlZddlmZmZ ddlmZ ddl	m
Z
 eeZG dd deZdS )	u   :module: watchdog.observers.inotify_buffer
:synopsis: A wrapper for ``Inotify``.
:author: thomas.amland@gmail.com (Thomas Amland)
:author: contact@tiger-222.fr (Mickaël Schoentgen)
:platforms: linux
    )annotationsN)InotifyInotifyEvent)
BaseThread)DelayedQueuec                      s`   e Zd ZdZdZdddd fddZdddZd ddZd ddZd!ddZ	d ddZ
  ZS )"InotifyBufferzA wrapper for `Inotify` that holds events for `delay` seconds. During
    this time, IN_MOVED_FROM and IN_MOVED_TO events are paired.
    g      ?FN	recursive
event_maskpathbytesr	   boolr
   
int | NonereturnNonec                  s6   t    td | j| _t|||d| _|   d S )N0InotifyEvent | tuple[InotifyEvent, InotifyEvent]r   )super__init__r   delay_queuer   _inotifystart)selfr   r	   r
   	__class__ U/var/www/vscode/kcb/lib/python3.10/site-packages/watchdog/observers/inotify_buffer.pyr      s   
zInotifyBuffer.__init__7InotifyEvent | tuple[InotifyEvent, InotifyEvent] | Nonec                 C  s
   | j  S )zReturns a single event or a tuple of from/to events in case of a
        paired move event. If this buffer has been closed, immediately return
        None.
        )r   getr   r   r   r   
read_event!   s   
zInotifyBuffer.read_eventc                 C  s   | j   | j  d S N)r   closer   r   r   r   r   on_thread_stop(   s   
zInotifyBuffer.on_thread_stopc                 C  s   |    |   d S r!   )stopjoinr   r   r   r   r"   ,   s   zInotifyBuffer.close
event_listlist[InotifyEvent]6list[InotifyEvent | tuple[InotifyEvent, InotifyEvent]]c                   s   g }|D ]I t d  d
 fdd} jrHt|D ]\}}||r*| f||<  nq| j|}|dur=|| f qt d	 |  q|  q|S )zGroup any matching move eventszin-event %seventr   r   r   c                   s   t | t o| jo| j jkS r!   )
isinstancetupleis_moved_fromcookie)r)   inotify_eventr   r   matching_from_event6   s   z8InotifyBuffer._group_events.<locals>.matching_from_eventNz'could not find matching move_from event)r)   r   r   r   )loggerdebugis_moved_to	enumerater   removeappend)r   r&   groupedr0   indexr)   
from_eventr   r.   r   _group_events0   s$   

zInotifyBuffer._group_eventsc                 C  s   d}|   rS|sU| j }| |}|D ]6}t|ts(|jr(|j| jjkr'd}qt|t o0|j	}| j
j||d t|tsJ|jrJ|j| jjkrJd}q|   rW|rdS dS dS dS )zRead event from `inotify` and add them to `queue`. When reading a
        IN_MOVE_TO event, remove the previous added matching IN_MOVE_FROM event
        and add them back to the queue as a tuple.
        FT)r   N)should_keep_runningr   read_eventsr:   r*   r+   
is_ignoredsrc_pathr   r,   r   putis_delete_self)r   deleted_selfinotify_eventsgrouped_eventsr/   r   r   r   r   runK   s&   

zInotifyBuffer.run)r   r   r	   r   r
   r   r   r   )r   r   )r   r   )r&   r'   r   r(   )__name__
__module____qualname____doc__r   r   r    r#   r"   r:   rD   __classcell__r   r   r   r   r      s    



r   )rH   
__future__r   loggingwatchdog.observers.inotify_cr   r   watchdog.utilsr   watchdog.utils.delayed_queuer   	getLoggerrE   r1   r   r   r   r   r   <module>   s    
