Frame13/ayon-core
1
0
Fork 1
mirror of https://github.com/ynput/ayon-core.git synced 2026-01-01 16:34:53 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

moved timers manager module

Browse source
This commit is contained in:
iLLiCiTiT 2021-07-28 14:12:16 +02:00
parent fca039fdfa
commit 6fb0d1fece
5 changed files with 0 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

7
openpype/modules/timers_manager/__init__.py
View file

@ -1,7 +0,0 @@
from .timers_manager import (
TimersManager
)
__all__ = (
"TimersManager",
)

26
openpype/modules/timers_manager/interfaces.py
View file

@ -1,26 +0,0 @@
from abc import abstractmethod
from openpype.modules import OpenPypeInterface
class ITimersManager(OpenPypeInterface):
timer_manager_module = None
@abstractmethod
def stop_timer(self):
pass
@abstractmethod
def start_timer(self, data):
pass
def timer_started(self, data):
if not self.timer_manager_module:
return
self.timer_manager_module.timer_started(self.id, data)
def timer_stopped(self):
if not self.timer_manager_module:
return
self.timer_manager_module.timer_stopped(self.id)

40
openpype/modules/timers_manager/rest_api.py
View file

@ -1,40 +0,0 @@
from aiohttp.web_response import Response
from openpype.api import Logger
log = Logger().get_logger("Event processor")
class TimersManagerModuleRestApi:
"""
REST API endpoint used for calling from hosts when context change
happens in Workfile app.
"""
def __init__(self, user_module, server_manager):
self.module = user_module
self.server_manager = server_manager
self.prefix = "/timers_manager"
self.register()
def register(self):
self.server_manager.add_route(
"POST",
self.prefix + "/start_timer",
self.start_timer
)
async def start_timer(self, request):
data = await request.json()
try:
project_name = data['project_name']
asset_name = data['asset_name']
task_name = data['task_name']
hierarchy = data['hierarchy']
except KeyError:
log.error("Payload must contain fields 'project_name, " +
"'asset_name', 'task_name', 'hierarchy'")
return Response(status=400)
self.module.stop_timers()
self.module.start_timer(project_name, asset_name, task_name, hierarchy)
return Response(status=200)

225
openpype/modules/timers_manager/timers_manager.py
View file

@ -1,225 +0,0 @@
import os
import collections
from openpype.modules import PypeModule
from openpype_interfaces import (
ITimersManager,
ITrayService,
IIdleManager,
IWebServerRoutes
)
from avalon.api import AvalonMongoDB
class TimersManager(PypeModule, ITrayService, IIdleManager, IWebServerRoutes):
""" Handles about Timers.
Should be able to start/stop all timers at once.
If IdleManager is imported then is able to handle about stop timers
when user idles for a long time (set in presets).
"""
name = "timers_manager"
label = "Timers Service"
def initialize(self, modules_settings):
timers_settings = modules_settings[self.name]
self.enabled = timers_settings["enabled"]
auto_stop = timers_settings["auto_stop"]
# When timer will stop if idle manager is running (minutes)
full_time = int(timers_settings["full_time"] * 60)
# How many minutes before the timer is stopped will popup the message
message_time = int(timers_settings["message_time"] * 60)
self.auto_stop = auto_stop
self.time_show_message = full_time - message_time
self.time_stop_timer = full_time
self.is_running = False
self.last_task = None
# Tray attributes
self.signal_handler = None
self.widget_user_idle = None
self.signal_handler = None
self.modules = []
def tray_init(self):
from .widget_user_idle import WidgetUserIdle, SignalHandler
self.widget_user_idle = WidgetUserIdle(self)
self.signal_handler = SignalHandler(self)
def tray_start(self, *_a, **_kw):
return
def tray_exit(self):
"""Nothing special for TimersManager."""
return
def webserver_initialization(self, server_manager):
"""Implementation of IWebServerRoutes interface."""
if self.tray_initialized:
from .rest_api import TimersManagerModuleRestApi
self.rest_api_obj = TimersManagerModuleRestApi(self,
server_manager)
def start_timer(self, project_name, asset_name, task_name, hierarchy):
"""
Start timer for 'project_name', 'asset_name' and 'task_name'
Called from REST api by hosts.
Args:
project_name (string)
asset_name (string)
task_name (string)
hierarchy (string)
"""
dbconn = AvalonMongoDB()
dbconn.install()
dbconn.Session["AVALON_PROJECT"] = project_name
asset_doc = dbconn.find_one({
"type": "asset", "name": asset_name
})
if not asset_doc:
raise ValueError("Uknown asset {}".format(asset_name))
task_type = ''
try:
task_type = asset_doc["data"]["tasks"][task_name]["type"]
except KeyError:
self.log.warning("Couldn't find task_type for {}".
format(task_name))
hierarchy = hierarchy.split("\\")
hierarchy.append(asset_name)
data = {
"project_name": project_name,
"task_name": task_name,
"task_type": task_type,
"hierarchy": hierarchy
}
self.timer_started(None, data)
def timer_started(self, source_id, data):
for module in self.modules:
if module.id != source_id:
module.start_timer(data)
self.last_task = data
self.is_running = True
def timer_stopped(self, source_id):
for module in self.modules:
if module.id != source_id:
module.stop_timer()
def restart_timers(self):
if self.last_task is not None:
self.timer_started(None, self.last_task)
def stop_timers(self):
if self.is_running is False:
return
self.widget_user_idle.bool_not_stopped = False
self.widget_user_idle.refresh_context()
self.is_running = False
for module in self.modules:
module.stop_timer()
def connect_with_modules(self, enabled_modules):
for module in enabled_modules:
if not isinstance(module, ITimersManager):
continue
module.timer_manager_module = self
self.modules.append(module)
def callbacks_by_idle_time(self):
"""Implementation of IIdleManager interface."""
# Time when message is shown
if not self.auto_stop:
return {}
callbacks = collections.defaultdict(list)
callbacks[self.time_show_message].append(lambda: self.time_callback(0))
# Times when idle is between show widget and stop timers
show_to_stop_range = range(
self.time_show_message - 1, self.time_stop_timer
)
for num in show_to_stop_range:
callbacks[num].append(lambda: self.time_callback(1))
# Times when widget is already shown and user restart idle
shown_and_moved_range = range(
self.time_stop_timer - self.time_show_message
)
for num in shown_and_moved_range:
callbacks[num].append(lambda: self.time_callback(1))
# Time when timers are stopped
callbacks[self.time_stop_timer].append(lambda: self.time_callback(2))
return callbacks
def time_callback(self, int_def):
if not self.signal_handler:
return
if int_def == 0:
self.signal_handler.signal_show_message.emit()
elif int_def == 1:
self.signal_handler.signal_change_label.emit()
elif int_def == 2:
self.signal_handler.signal_stop_timers.emit()
def change_label(self):
if self.is_running is False:
return
if (
not self.idle_manager
or self.widget_user_idle.bool_is_showed is False
):
return
if self.idle_manager.idle_time > self.time_show_message:
value = self.time_stop_timer - self.idle_manager.idle_time
else:
value = 1 + (
self.time_stop_timer -
self.time_show_message -
self.idle_manager.idle_time
)
self.widget_user_idle.change_count_widget(value)
def show_message(self):
if self.is_running is False:
return
if self.widget_user_idle.bool_is_showed is False:
self.widget_user_idle.show()
def change_timer_from_host(self, project_name, asset_name, task_name):
"""Prepared method for calling change timers on REST api"""
webserver_url = os.environ.get("OPENPYPE_WEBSERVER_URL")
if not webserver_url:
self.log.warning("Couldn't find webserver url")
return
rest_api_url = "{}/timers_manager/start_timer".format(webserver_url)
try:
import requests
except Exception:
self.log.warning("Couldn't start timer")
return
data = {
"project_name": project_name,
"asset_name": asset_name,
"task_name": task_name
}
requests.post(rest_api_url, json=data)

170
openpype/modules/timers_manager/widget_user_idle.py
View file

@ -1,170 +0,0 @@
from Qt import QtCore, QtGui, QtWidgets
from openpype import resources, style
class WidgetUserIdle(QtWidgets.QWidget):
SIZE_W = 300
SIZE_H = 160
def __init__(self, module):
super(WidgetUserIdle, self).__init__()
self.bool_is_showed = False
self.bool_not_stopped = True
self.module = module
icon = QtGui.QIcon(resources.pype_icon_filepath())
self.setWindowIcon(icon)
self.setWindowFlags(
QtCore.Qt.WindowCloseButtonHint
| QtCore.Qt.WindowMinimizeButtonHint
)
self._translate = QtCore.QCoreApplication.translate
self.font = QtGui.QFont()
self.font.setFamily("DejaVu Sans Condensed")
self.font.setPointSize(9)
self.font.setBold(True)
self.font.setWeight(50)
self.font.setKerning(True)
self.resize(self.SIZE_W, self.SIZE_H)
self.setMinimumSize(QtCore.QSize(self.SIZE_W, self.SIZE_H))
self.setMaximumSize(QtCore.QSize(self.SIZE_W+100, self.SIZE_H+100))
self.setStyleSheet(style.load_stylesheet())
self.setLayout(self._main())
self.refresh_context()
self.setWindowTitle('Pype - Stop timers')
def _main(self):
self.main = QtWidgets.QVBoxLayout()
self.main.setObjectName('main')
self.form = QtWidgets.QFormLayout()
self.form.setContentsMargins(10, 15, 10, 5)
self.form.setObjectName('form')
msg_info = 'You didn\'t work for a long time.'
msg_question = 'Would you like to stop Timers?'
msg_stopped = (
'Your Timers were stopped. Do you want to start them again?'
)
self.lbl_info = QtWidgets.QLabel(msg_info)
self.lbl_info.setFont(self.font)
self.lbl_info.setTextFormat(QtCore.Qt.RichText)
self.lbl_info.setObjectName("lbl_info")
self.lbl_info.setWordWrap(True)
self.lbl_question = QtWidgets.QLabel(msg_question)
self.lbl_question.setFont(self.font)
self.lbl_question.setTextFormat(QtCore.Qt.RichText)
self.lbl_question.setObjectName("lbl_question")
self.lbl_question.setWordWrap(True)
self.lbl_stopped = QtWidgets.QLabel(msg_stopped)
self.lbl_stopped.setFont(self.font)
self.lbl_stopped.setTextFormat(QtCore.Qt.RichText)
self.lbl_stopped.setObjectName("lbl_stopped")
self.lbl_stopped.setWordWrap(True)
self.lbl_rest_time = QtWidgets.QLabel("")
self.lbl_rest_time.setFont(self.font)
self.lbl_rest_time.setTextFormat(QtCore.Qt.RichText)
self.lbl_rest_time.setObjectName("lbl_rest_time")
self.lbl_rest_time.setWordWrap(True)
self.lbl_rest_time.setAlignment(QtCore.Qt.AlignCenter)
self.form.addRow(self.lbl_info)
self.form.addRow(self.lbl_question)
self.form.addRow(self.lbl_stopped)
self.form.addRow(self.lbl_rest_time)
self.group_btn = QtWidgets.QHBoxLayout()
self.group_btn.addStretch(1)
self.group_btn.setObjectName("group_btn")
self.btn_stop = QtWidgets.QPushButton("Stop timer")
self.btn_stop.setToolTip('Stop\'s All timers')
self.btn_stop.clicked.connect(self.stop_timer)
self.btn_continue = QtWidgets.QPushButton("Continue")
self.btn_continue.setToolTip('Timer won\'t stop')
self.btn_continue.clicked.connect(self.continue_timer)
self.btn_close = QtWidgets.QPushButton("Close")
self.btn_close.setToolTip('Close window')
self.btn_close.clicked.connect(self.close_widget)
self.btn_restart = QtWidgets.QPushButton("Start timers")
self.btn_restart.setToolTip('Timer will be started again')
self.btn_restart.clicked.connect(self.restart_timer)
self.group_btn.addWidget(self.btn_continue)
self.group_btn.addWidget(self.btn_stop)
self.group_btn.addWidget(self.btn_restart)
self.group_btn.addWidget(self.btn_close)
self.main.addLayout(self.form)
self.main.addLayout(self.group_btn)
return self.main
def refresh_context(self):
self.lbl_question.setVisible(self.bool_not_stopped)
self.lbl_rest_time.setVisible(self.bool_not_stopped)
self.lbl_stopped.setVisible(not self.bool_not_stopped)
self.btn_continue.setVisible(self.bool_not_stopped)
self.btn_stop.setVisible(self.bool_not_stopped)
self.btn_restart.setVisible(not self.bool_not_stopped)
self.btn_close.setVisible(not self.bool_not_stopped)
def change_count_widget(self, time):
str_time = str(time)
self.lbl_rest_time.setText(str_time)
def stop_timer(self):
self.module.stop_timers()
self.close_widget()
def restart_timer(self):
self.module.restart_timers()
self.close_widget()
def continue_timer(self):
self.close_widget()
def closeEvent(self, event):
event.ignore()
if self.bool_not_stopped is True:
self.continue_timer()
else:
self.close_widget()
def close_widget(self):
self.bool_is_showed = False
self.bool_not_stopped = True
self.refresh_context()
self.hide()
def showEvent(self, event):
self.bool_is_showed = True
class SignalHandler(QtCore.QObject):
signal_show_message = QtCore.Signal()
signal_change_label = QtCore.Signal()
signal_stop_timers = QtCore.Signal()
def __init__(self, module):
super(SignalHandler, self).__init__()
self.module = module
self.signal_show_message.connect(module.show_message)
self.signal_change_label.connect(module.change_label)
self.signal_stop_timers.connect(module.stop_timers)