← Back
Editing: observer.cpython-312.pyc
� �ic[ � �� � d dl Z d dlmZ d dlmZ d dlmZ d dlZ ej dd� d dl m Z e j e � Z g d�Z G d � d e� Z G d � d e� Z G d � de� Zy)� N)�partial)�DBUS_FLAG_NONE)�Signal�Gioz2.0)r )�DBusObserverError� DBusObserver�GLibMonitoringc � � e Zd ZdZy)r z'Exception class for the DBus observers.N)�__name__� __module__� __qualname__�__doc__� � �;/usr/lib/python3.12/site-packages/dasbus/client/observer.pyr r ( s � �1�r r c �X � e Zd ZdZeeddfd�� Zed� � Zed� � Zed� � Z y)r z4The low-level DBus monitoring library based on GLib.Nc �� � d}d}|rt | j |df�� }|rt | j |df�� }t j |||||� }t | j ||� S )z,Watch a service name on the DBus connection.Nr )� user_data)r �_name_appeared_callback�_name_vanished_callbackr � bus_watch_name_on_connection� _unwatch_name) �cls� connection�name�flags� name_appeared� name_vanished�name_appeared_closure�name_vanished_closure�registration_ids r � watch_namezGLibMonitoring.watch_name0 s� � � !%� � $� � �$+��+�+�(�"�-�%� !� �$+��+�+�(�"�-�%� !� �:�:� � � � !� !� �� � � � � � � � r c � � |\ }} ||g|��� y)zCallback for watch_name..Nr )r r r � name_ownerr �callback� callback_argss r r z&GLibMonitoring._name_appeared_callbackQ s � � #,� ��-� ��,�m�,r c � � |\ }} ||� y)zCallback for watch_name.Nr )r r r r r% r& s r r z&GLibMonitoring._name_vanished_callbackZ s � � #,� ��-� �-� r c �. � t j |� y)z5Stops watching a service name on the DBus connection.N)r �bus_unwatch_name)r r r! s r r z GLibMonitoring._unwatch_namec s � � � � �_�-r ) r r r r � classmethodr r" r r r r r r r r - s\ � �>��0>�!%�T� � � �@ �-� �-� �!� �!� �.� �.r r c � � e Zd ZdZefd�Zed� � Zed� � Zed� � Z ed� � Z d� Z d� Z d � Z d � Zd � Zd � Zd � Zd� Zd� Zd� Zy)r a4 Base class for DBus observers. This class is recommended to use only to watch the availability of a service on DBus. It doesn't provide any support for accessing objects provided by the service. Usage: .. code-block:: python # Create the observer and connect to its signals. observer = DBusObserver(SystemBus, "org.freedesktop.NetworkManager") def callback1(observer): print("Service is available!") def callback2(observer): print("Service is unavailable!") observer.service_available.connect(callback1) observer.service_unavailable.connect(callback2) # Connect to the service once it is available. observer.connect_once_available() # Disconnect the observer. observer.disconnect() c � � || _ || _ d| _ t � | _ t � | _ || _ g | _ y)z�Creates a DBus service observer. :param message_bus: a message bus :param service_name: a DBus name of a service FN)� _message_bus� _service_name�_is_service_availabler �_service_available�_service_unavailable� _monitoring�_subscriptions)�self� message_bus� service_name� monitorings r �__init__zDBusObserver.__init__� s@ � � (��� )���%*��"�"(�(�� �$*�H��!�%��� ��r c � � | j S )zReturns a DBus name.�r. �r4 s r r6 zDBusObserver.service_name� s � � �!�!�!r c � � | j S )zThe proxy can be accessed.)r/ r; s r �is_service_availablez!DBusObserver.is_service_available� s � � �)�)�)r c � � | j S )z�Signal that emits when the service is available. Signal emits this class as an argument. You have to call the watch method to activate the signals. )r0 r; s r �service_availablez DBusObserver.service_available� s � � �&�&�&r c � � | j S )z�Signal that emits when the service is unavailable. Signal emits this class as an argument. You have to call the watch method to activate the signals. )r1 r; s r �service_unavailablez DBusObserver.service_unavailable� s � � �(�(�(r c �$ � | j � y)z�Connect to the service once it is available. The observer is not connected to the service until it emits the service_available signal. N)�_watchr; s r �connect_once_availablez#DBusObserver.connect_once_available� s � � � � � r c �^ � | j � | j r| j � yy)z�Disconnect from the service. Disconnect from the service if it is connected and stop watching its availability. N)�_unwatchr= �_disable_servicer; s r � disconnectzDBusObserver.disconnect� s' � � � � �� � $� $� � !� !� #� %r c �� � | j j | j j | j t | j | j � }| j j |� y)z Watch the service name on DBus.N) r2 r" r- r r6 r � _service_name_appeared_callback� _service_name_vanished_callbackr3 �append)r4 � subscriptions r rC zDBusObserver._watch� s[ � ��'�'�2�2� � � � (� (� � � � � � 0� 0� � 0� 0� � � ���"�"�<�0r c �z � | j r/| j j � } |� | j r�.yy)z'Stop to watch the service name on DBus.N)r3 �pop)r4 r% s r rF zDBusObserver._unwatch� s/ � ��!�!��*�*�.�.�0�H� �J� �!�!r c �H � d| _ | j j | � y)zEnable the service.TN)r/ r0 �emitr; s r �_enable_servicez DBusObserver._enable_service� s � �%)��"� � � �$�$�T�*r c �H � d| _ | j j | � y)zDisable the service.FN)r/ r1 rQ r; s r rG z DBusObserver._disable_service� s � �%*��"� �!�!�&�&�t�,r c �> � | j s| j � yy�z Callback for the watch method.N)r= rR �r4 �argss r rJ z,DBusObserver._service_name_appeared_callback� s � ��(�(� � � � "� )r c �> � | j r| j � yyrU )r= rG rV s r rK z,DBusObserver._service_name_vanished_callback� s � � � $� $� � !� !� #� %r c � � | j S )z(Returns a string version of this object.r: r; s r �__str__zDBusObserver.__str__� s � ��!�!�!r c �b � dj | j j | j � S )z Returns a string representation.z{}({}))�format� __class__r r. r; s r �__repr__zDBusObserver.__repr__� s) � ���� �N�N� #� #� � � � � r N)r r r r r r8 �propertyr6 r= r? rA rD rH rC rF rR rG rJ rK rZ r^ r r r r r i s� � � �< >L� !� �"� �"� �*� �*� �'� �'� �)� �)�� $� 1�� +� -� #� $� "� r r )�logging� functoolsr �dasbus.constantsr � dasbus.signalr �gi�require_version� gi.repositoryr � getLoggerr �log�__all__� Exceptionr �objectr r r r r �<module>rl sk ��* � � +� � � �� � �5�%� � ��g����!�� �� � � � 9.�V� 9.�xM �6� M r
Save File
Cancel