#!/usr/bin/env python # -*- coding: utf-8 -*- # Copyright 2012-2016 Mir Calculate. http://www.calculate-linux.org # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import logging # logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.DEBUG) logger = logging.getLogger(__name__) import os from os import path from PyQt5 import QtCore, QtGui, QtWidgets import time from calculate.consolegui.application.user_update import user_can_run_update from calculate.lib.datavars import DataVars import dbus import dbus.service import dbus.mainloop.glib import pyinotify from calculate.lib.utils.tools import debug from calculate.consolegui.application.dbus_service import ( DBUS_NAME, DBUS_METHOD_APP, DBUS_MAINAPP, DBUS_APP_UPDATER, DBUS_NAME_UPDATER) from calculate.update.update_info import UpdateInfo import sys from calculate.lib.cl_lang import setLocalTranslate, getLazyLocalTranslate, _ setLocalTranslate('cl_consolegui3', sys.modules[__name__]) __ = getLazyLocalTranslate(_) GUI_UPDATE_APP = DBUS_METHOD_APP % "update" MINUTE = 60 def check_livecd_and_autocheck(): """ Проверить необходимость запуска cl-update-checker Не запускать на livecd или если выключена автоматическая проверка обновлений """ dv = DataVars() try: dv.importData() dv.flIniFile() if (dv.Get('os_root_type') == "livecd" or dv.Get('update.cl_update_autocheck_set') == 'off'): return False return True finally: dv.close() class SysTray(QtWidgets.QSystemTrayIcon): ICON = "/usr/share/pixmaps/calculate-console-update.svg" ICON_NO_UPDATE = "/usr/share/pixmaps/calculate-console-update-no.svg" def __init__(self, parent): super().__init__() self.parent = parent self.timeout = 0 self.activated.connect(self.systrayActivate) self.icon_on = QtGui.QIcon() self.icon_on.addPixmap(QtGui.QPixmap(self.ICON)) self.icon_off = QtGui.QIcon() self.icon_off.addPixmap(QtGui.QPixmap(self.ICON_NO_UPDATE)) self.update(has_updates=False) self.setVisible(True) def run_cl_gui_update(self): method = "update" cmd = '/usr/bin/cl-console-gui-%s'%method params = ["--skip-options"] os.system("{cmd} {params} &>/dev/null &".format( cmd=cmd, params=" ".join(params))) def show_update(self): try: iface = self.parent.get_dbus_iface(DBUS_MAINAPP) if iface: iface.update() return True except dbus.DBusException as e: pass try: iface = self.parent.get_dbus_iface(GUI_UPDATE_APP) if iface: iface.show() return True except dbus.DBusException as e: pass return False def systrayActivate(self, reason): if time.time() > self.timeout: self.timeout = time.time() + 5 self.show_update() or self.run_cl_gui_update() self.parent.step() else: self.show_update() def update(self, has_updates=False): if has_updates: self.setVisible(True) self.setIcon(self.icon_on) self.setToolTip(_("Updates are available")) else: self.setIcon(self.icon_off) self.setToolTip(_("No updates")) self.setVisible(False) class DBusChecker(dbus.service.Object): def __init__(self, name, session, parent): self.parent = parent # export this object to dbus dbus.service.Object.__init__(self, name, session) self.parent.systray.setVisible(True) @dbus.service.method(DBUS_NAME_UPDATER, in_signature='', out_signature='') def hide_systray(self): self.parent.systray.setVisible(False) @dbus.service.method(DBUS_NAME_UPDATER, in_signature='', out_signature='') def show_systray(self): self.parent.systray.setVisible(True) @dbus.service.method(DBUS_NAME_UPDATER, in_signature='', out_signature='b') def ping(self): return True @dbus.service.method(DBUS_NAME_UPDATER, in_signature='', out_signature='') def quit(self): QtWidgets.qApp.quit() @dbus.service.method(DBUS_NAME_UPDATER, in_signature='', out_signature='') def check(self): self.parent.step() class CheckThread(QtWidgets.QMainWindow, UpdateInfo): interval = 5 * MINUTE def __init__(self, bus): super().__init__() UpdateInfo.__init__(self) self.bus = bus self.wm = pyinotify.WatchManager() self.notifier = pyinotify.Notifier(self.wm, timeout=1) self.wm.add_watch(path.dirname(UpdateInfo.update_file), pyinotify.IN_DELETE | pyinotify.IN_CREATE, self.event_step) self.already_timer = QtCore.QTimer(self) self.already_timer.timeout.connect(self.notify_step) self.already_timer.start(1) self.check_timer = QtCore.QTimer(self) self.check_timer.timeout.connect(self.step) self.check_timer.start(self.interval * 1) self.systray = SysTray(self) self.gui_runned = False self.systray.update(has_updates=False) self.step() def get_dbus_iface(self, app_name): try: remote_object = self.bus.get_object(DBUS_NAME, app_name) return dbus.Interface(remote_object, DBUS_NAME) except dbus.DBusException: return None def ping_app(self, app_name): try: obj = self.get_dbus_iface(app_name) if obj: obj.ping() return True except dbus.DBusException: pass return False def notify_step(self): if self.notifier.check_events(timeout=1): self.notifier.read_events() self.notifier.process_events() def event_step(self, event): if event.name == path.basename(UpdateInfo.update_file): self.step() def is_console_gui_run(self): return self.ping_app(DBUS_MAINAPP) def is_gui_update_run(self): return self.ping_app(GUI_UPDATE_APP) def step(self): if self.need_update(): self.systray.update(has_updates=True) return self.systray.update(has_updates=False) if __name__ == '__main__': import sys pid = os.fork() if pid != 0: sys.exit(0) if not user_can_run_update(True): sys.stderr.write(_("User can not to perform the system update") + "\n") sys.exit(1) if not check_livecd_and_autocheck(): sys.stderr.write( _("No need to run a live cd or disable autoupdates") + "\n") sys.exit(1) app = QtWidgets.QApplication(sys.argv) for i in [0.5, 1, 2, 5]: if QtWidgets.QSystemTrayIcon.isSystemTrayAvailable(): break time.sleep(i) # Enable glib main loop support dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) # Get the session bus try: bus = dbus.SessionBus() except dbus.exceptions.DBusException as e: sys.exit(1) try: remote_object = bus.get_object(DBUS_NAME_UPDATER, DBUS_APP_UPDATER) g = dbus.Interface(remote_object, DBUS_NAME_UPDATER) sys.exit(1) except Exception as e: pass QtWidgets.QApplication.setQuitOnLastWindowClosed(False) QtWidgets.QApplication.setApplicationName(_("Calculate Update Checker")) # Export the service name = dbus.service.BusName(DBUS_NAME_UPDATER, bus) # Export the object ct = CheckThread(bus) DBusChecker(bus, DBUS_APP_UPDATER, ct) sys.exit(app.exec_())