import os
import sys
import time
sys.path.append(os.path.dirname(__file__))
import datetime
import getpass
import logging
from typing import List, Callable
from qtpy.QtCore import Signal, QTimer, Qt, QTranslator, QLocale, QSize
from qtpy.QtGui import QCloseEvent, QTextCursor, QResizeEvent, QFontDatabase, QMoveEvent, QFont, QIcon, QPixmap
from qtpy.QtWidgets import QApplication, QTextEdit, QMessageBox, QToolBar, QSplashScreen, QStatusBar
import pmgwidgets
pmgwidgets.in_unit_test = lambda: False
from pmgwidgets import PMGToolBar, ActionWithMessage, PMDockObject, create_icon
from pyminer2.extensions.extensions_manager.manager import extensions_manager
from pyminer2.features import base
from pyminer2.features.io.settings import Settings
from pyminer2.features.io.settings import load_theme
from pyminer2.ui.base.widgets.controlpanel import PMPageExt
from pyminer2.ui.pmwidgets import BaseMainWindow
from pyminer2.globals import get_main_window, get_root_dir, get_application, pythonVersion, openURL
from pyminer2 import globals
t0 = time.time()
from pmlocalserver import server
# TODO:实例化server需要消耗
log_folder = os.path.join(get_root_dir(), 'log')
if not os.path.exists(log_folder):
os.mkdir(log_folder)
logging.Formatter.default_msec_format = '%s.%03d'
logging_file = os.path.join(log_folder, f'log_{datetime.datetime.now().strftime("%Y-%m-%d")}.log')
logging.basicConfig(
format='[%(asctime)s] %(levelname)-8s %(name)s [%(module)s:%(funcName)s:%(lineno)s] %(message)+8s',
# filename=logging_file,
# filemode='a',
level=logging.DEBUG,
handlers=[logging.FileHandler(logging_file, 'a', encoding='utf8')]
)
logger = logging.getLogger('app2.py')
logger.setLevel(logging.DEBUG)
logger.info('Program starts up')
if "--debug" in sys.argv:
del sys.argv[sys.argv.index("--debug")]
logging.basicConfig(level=logging.DEBUG)
def load_translator(app: QApplication):
"""加载翻译文件
Args:
app: PyQt的Application。
"""
# 注意需要保留trans变量的引用
app.trans = QTranslator()
lang_name = QLocale.system().name()
lang = os.path.join(os.path.dirname(__file__), 'pyminer2', 'languages', '{}'.format(lang_name),
'{}.qm'.format(lang_name))
print('lang:', lang)
if os.path.isfile(lang):
app.trans.load(lang)
app.installTranslator(app.trans)
def load_fonts(app):
"""
注册字体文件
"""
app.font_dir = path = os.path.join(get_root_dir(), 'ui/source/font')
for name in os.listdir(path):
QFontDatabase.addApplicationFont(os.path.join(path, name))
font_db = QFontDatabase()
def updateSplashMsg(ext_load_status: dict):
splash = get_application().splash
percent = '100%' if ext_load_status.get('ext_count') == 0 \
else round(ext_load_status.get('loaded') / ext_load_status.get('ext_count') * 100)
try:
msg = 'Loading:' + ext_load_status.get('ext_name') + '...' + str(percent) + '%'
splash.showMessage(msg, Qt.AlignHCenter | Qt.AlignBottom, Qt.white)
except TypeError:
return
class PMToolBarHome(PMGToolBar):
"""
定义菜单工具栏按钮。
"""
def __init__(self):
super().__init__()
self.add_tool_button(
'button_new_script', self.tr('New Script'),
create_icon(":/color/source/theme/color/icons/script.svg"))
self.add_tool_button(
'button_new',
self.tr('New'),
create_icon(":/color/source/theme/color/icons/new_project.svg"))
self.add_tool_button('button_open', self.tr('Open'), create_icon(
":/color/source/theme/color/icons/open.svg"))
self.addSeparator()
self.add_tool_button(
'button_import_data', self.tr('Get Data'),
create_icon(":/color/source/theme/color/icons/import.svg"))
self.add_tool_button(
'button_import_database', self.tr('Import Database'),
create_icon(":/color/source/theme/color/icons/import_database.svg"))
self.add_buttons(3, ['button_open_variable', 'button_save_workspace', 'button_clear_workspace'],
[self.tr('Load Var'), self.tr('Save Var'), self.tr('Clear Var')],
[":/color/source/theme/color/icons/var_open.svg",
":/color/source/theme/color/icons/save.svg",
":/color/source/theme/color/icons/clear.svg"])
self.addSeparator()
self.add_tool_button(
'button_appstore',
self.tr('Extensions'),
create_icon(':/color/source/theme/color/icons/pypi_color.svg'))
self.add_tool_button('button_help', self.tr('Help'), create_icon(
':/color/source/theme/color/icons/help.svg'))
self.add_tool_button('button_community', self.tr('Community'), create_icon(
':/color/source/theme/color/icons/community.svg'))
self.add_tool_button('view_config', self.tr('Layout'), create_icon(
':/color/source/theme/color/icons/save_layout.svg'))
self.add_tool_button('button_settings', self.tr('Settings'), create_icon(
':/color/source/theme/color/icons/setting.svg'))
def process_visibility_actions(self, e: ActionWithMessage):
"""
处理”视图“菜单点击时触发的事件。
"""
main_window = get_main_window()
dws = main_window.dock_widgets
if e.message == 'load_standard_layout':
main_window.load_predefined_layout('standard')
elif e.message in dws.keys():
dws[e.message].setVisible(e.isChecked())
elif e.message == 'lock_layout':
main_window.set_dock_titlebar_visible(not e.isChecked()) # 如果界面锁定(True)则标题栏不可见(False)所以需要取反。
def bind_events(self):
"""
绑定事件。
"""
self.get_control_widget('button_clear_workspace').clicked.connect(lambda: get_main_window().clear_workspace())
self.get_control_widget('button_settings').clicked.connect(lambda: get_main_window().main_option_display())
self.get_control_widget('button_appstore').clicked.connect(lambda: get_main_window().main_appstore_dispaly())
self.get_control_widget('button_help').clicked.connect(lambda: get_main_window().main_help_display())
self.get_control_widget('button_community').clicked.connect(lambda: get_main_window().main_community_display())
self.append_menu('button_new_script', 'Python',
lambda: get_main_window().main_homesite_display(),
create_icon(':/color/source/theme/color/icons/python.svg'))
self.append_menu('button_new_script', 'Notebook',
lambda: get_main_window().main_jupyter_display(),
create_icon(':/color/source/theme/color/icons/Jupyter.svg'))
self.append_menu('button_new_script', 'R',
lambda: get_main_window().main_homesite_display(),
create_icon(':/color/source/theme/color/icons/R.svg'))
self.append_menu('button_new_script', 'Markdown',
lambda: get_main_window().main_markdown_display(),
create_icon(':/logo/source/theme/color/icons/markdown.svg'))
self.append_menu('button_new_script', 'SQL',
lambda: get_main_window().main_homesite_display(),
create_icon(':/logo/source/theme/color/icons/sql.svg'))
self.append_menu('button_new_script', 'HTML',
lambda: get_main_window().main_homesite_display(),
create_icon(':/logo/source/theme/color/icons/html.svg'))
homeSiteIcon = create_icon(':/color/source/theme/color/icons/home_site.svg')
self.append_menu('button_help', self.tr('Support'),
lambda: get_main_window().main_homesite_display(),
homeSiteIcon)
helpDocIcon = create_icon(':/color/source/theme/color/icons/help_doc.svg')
self.append_menu('button_help', self.tr('Reference'),
lambda: get_main_window().main_help_display(),
helpDocIcon)
updateIcon = create_icon(':/color/source/theme/color/icons/check_update.svg')
self.append_menu('button_help', self.tr('Check for updates'),
lambda: get_main_window().main_check_update_display(),
updateIcon)
feedbackIcon = create_icon(':/color/source/theme/color/icons/feedback.svg')
self.append_menu('button_help', self.tr('Give Feedback'),
lambda: get_main_window().main_feedback_display(),
feedbackIcon)
aboutIcon = create_icon(':/color/source/theme/color/icons/info.svg')
self.append_menu('button_help', self.tr('About'),
lambda: get_main_window().main_about_display(),
aboutIcon)
class LogOutputConsole(QTextEdit, PMDockObject):
pass
class MainWindow(BaseMainWindow):
setupui_tasks: List[Callable] = []
boot_timer: QTimer = None
close_signal = Signal()
window_geometry_changed_signal = Signal()
layouts_ready_signal = Signal()
widgets_ready_signal = Signal()
events_ready_signal = Signal()
settings_changed_signal = Signal()
@classmethod
def __new__(cls, *args):
if not hasattr(cls, 'instance'):
instance = super().__new__(cls)
cls.instance = instance
return cls.instance
def __init__(self, parent=None):
super().__init__(parent)
t00 = time.time()
self.main_option_form = base.OptionForm()
import pyminer2.globals
pyminer2.globals._main_window = self
# 主窗体默认大小
# self.resize(1500, 850)
self.setIconSize(QSize(40, 40))
# 设置状态栏
self.statusBar = QStatusBar()
version = pythonVersion()
self.statusBar.showMessage(version, 0)
self.setStatusBar(self.statusBar)
settings = Settings()
root_dir = os.path.dirname(__file__)
pyminer2.globals._root_dir = root_dir
self.init_toolbar_tab()
self.add_toolbar('toolbar_home', PMToolBarHome(), text=self.tr('Files'))
self.setDockNestingEnabled(True)
self.setWindowTitle('PyMiner')
self.log_output_console = LogOutputConsole(self)
self.add_widget_on_dock(
'log_output_console',
self.log_output_console,
text='日志',
side='right')
# 初始化日志
self.slot_flush_console('info', 'system', self.tr('Welcome to PyMiner'))
self.extensions_manager = extensions_manager
self.extensions_manager.load_from_extension_folder(updateSplashMsg)
self.ext_manager_widget = PMPageExt(self)
dw = self.add_widget_on_dock(
'extension_panel',
self.ext_manager_widget,
text=self.tr('Plugs'),
side='left')
dw.setMaximumWidth(400)
load_theme(settings['theme']) # 组件都加载后再设置主题,否则可能有些组件不生效
self.show()
self.load_layout()
self.switch_toolbar('toolbar_home') # 启动完成时,将工具栏切换到‘主页’
self.on_main_window_shown()
# self.first_form_display()
self.start_pmlocalserver() # 只要在插件加载完成之后启动就行,目前放在最后
t01 = time.time()
logger.debug('Time Elapsed for loading main window contents: %f' % (t01 - t00))
def start_pmlocalserver(self):
"""
启动本地flask服务器pmlocalserver
Returns:None
"""
server.server_thread.start()
def clear_workspace(self):
from pyminer2.extensions.extensionlib.extension_lib import extension_lib
extension_lib.get_interface('ipython_console').run_command('Clearing_Variables_ =\'Clear All\'',
hint_text=self.tr('Start Clear...'), hidden=False)
extension_lib.get_interface('ipython_console').run_command('get_ipython().clear_all()', hint_text=self.tr('Clear all variables'),
hidden=False)
def add_toolbar(self, name: str, toolbar: QToolBar,
text: str = 'untitled toolbar'):
"""
添加一个工具栏。
"""
b = self.top_toolbar_tab.add_button(text)
toolbar.tab_button = b
b.clicked.connect(lambda: self.on_toolbar_switch_button_clicked(name))
if hasattr(self, 'toolbar_path'):
self.insertToolBar(self.toolbar_path, toolbar)
self.insertToolBarBreak(self.toolbar_path)
else:
self.addToolBarBreak(Qt.TopToolBarArea)
self.addToolBar(toolbar)
toolbar.setObjectName(name)
self.toolbars[name] = toolbar
toolbar.setMovable(False)
toolbar.setFloatable(False)
if self._current_toolbar_name != '':
self.refresh_toolbar_appearance()
def moveEvent(self, a0: 'QMoveEvent') -> None:
self.window_geometry_changed_signal.emit()
def resizeEvent(self, a0: QResizeEvent) -> None:
"""
窗口大小调节,或者位置改变的信号。
Window size adjustment, or a signal of a change in position.
"""
self.size_restriction_acquire()
super().resizeEvent(a0)
self.delayed_call(500, self.size_restriction_release)
self.window_geometry_changed_signal.emit()
def on_settings_changed(self):
self.settings_changed_signal.emit()
def delayed_call(self, time_ms: int, callback: Callable) -> None:
"""
封装了QTimer.SingleShot
:param time_ms:
:param callback:
:return:
"""
timer = QTimer()
timer.singleShot(time_ms, callback)
def size_restriction_acquire(self) -> None:
"""
设置插件尺寸的最大值。
控件需要指定get_split_portion_hint才可以。
:return:
"""
for k in self.dock_widgets.keys():
dw = self.dock_widgets[k]
horizontal_portion_hint = dw.widget().get_split_portion_hint()[0]
if horizontal_portion_hint is not None:
dw.setMaximumWidth(int(self.width() * horizontal_portion_hint))
dw.setMinimumWidth(int(self.width() * horizontal_portion_hint))
def size_restriction_release(self):
for w_name in self.dock_widgets.keys():
self.dock_widgets[w_name].setMaximumWidth(100000)
self.dock_widgets[w_name].setMaximumHeight(100000)
self.dock_widgets[w_name].setMinimumHeight(0)
self.dock_widgets[w_name].setMinimumWidth(0)
def on_main_window_shown(self):
"""
在界面显示后触发的事件。
Returns: None
"""
t0 = time.time()
super().on_main_window_shown()
self.layouts_ready_signal.emit()
for task in self.setupui_tasks:
task()
self.widgets_ready_signal.emit()
t1 = time.time()
logging.info('Layout ready time elapsed:%f' % (t1 - t0))
self.set_dock_titlebar_visible(Settings.get_instance()['dock_titlebar_visible'])
self.bind_events()
self.events_ready_signal.emit()
t2 = time.time()
logging.info('Events ready, time elapsed:%f' % (t2 - t1))
def first_form_display(self):
"""
显示"快速操作"窗口
Displays the "Quick Action" window
"""
self.main_first_form = base.FirstForm(parent=self)
self.main_first_form.show()
def main_appstore_dispaly(self):
"""
显示"应用商店"窗口
Displays the "App Store" window
"""
self.appstore = base.MarketPlace()
self.appstore.show()
def main_option_display(self):
"""
显示"选项"窗口
"""
if self.main_option_form is None:
self.main_option_form = base.OptionForm()
self.main_option_form.exec()
def main_help_display(self):
"""
打开帮助页面
"""
openURL("https://gitee.com/py2cn/pyminer/wikis")
def main_check_update_display(self):
"""
打开'检查更新'页面
"""
from pyminer2.features.util.update import check_update
update = check_update()
if update:
ret = QMessageBox.information(self, self.tr('update status'), self.tr('You can download and install the updated version, whether to restart and install the update'), QMessageBox.Yes | QMessageBox.Cancel,
QMessageBox.Yes)
if ret == QMessageBox.Yes:
self.main_install_update()
else:
QMessageBox.information(self, self.tr('update status'), self.tr('Is the latest version'), QMessageBox.Yes, QMessageBox.Yes)
# if reply == QMessageBox.Yes:
# openURL("http://www.pyminer.com")
def main_install_update(self):
closed = self.close()
if closed:
from pyminer2.features.util.platformutil import run_python_file_in_terminal
path = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'update', 'update.py')
run_python_file_in_terminal(path + ' -i')
def main_feedback_display(self):
"""
打开'反馈'页面
"""
reply = QMessageBox.information(self, self.tr('Feedback'), self.tr('You can give feedback through issue on suggestions or problems encountered in use'), QMessageBox.Yes | QMessageBox.No,
QMessageBox.Yes)
if reply == QMessageBox.Yes:
openURL("https://gitee.com/py2cn/pyminer/issues")
def main_homesite_display(self):
"""
打开官方网站页面
"""
openURL("http://www.pyminer.com")
def main_markdown_display(self):
print("TODO 添加markdown编辑器代码")
def main_jupyter_display(self):
print("打开jupyter-notebook")
os.chdir(globals.get_root_dir())
import subprocess
python_path = sys.executable
cmd = python_path + ' -m notebook'
subprocess.Popen(cmd)
def main_community_display(self):
"""
打开帮助页面
"""
openURL("https://www.kuxai.com/")
def main_project_wizard_display(self):
"""
打开新建项目向导
"""
self.project_wizard = base.ProjectWizardForm()
self.project_wizard.show()
def main_about_display(self):
"""
打开关于页面,并将当前操作系统信息写入页面
"""
self.about_me = base.AboutForm()
self.about_me.show()
def closeEvent(self, a0: QCloseEvent) -> None:
"""
主窗体退出时的事件,包括弹框提示等。Mac 上测试点击无法退出,修改为QMessageBox.Warning
"""
reply = QMessageBox(QMessageBox.Warning, self.tr('Close'), self.tr('Are you sure close?'))
reply.addButton(self.tr('OK'), QMessageBox.ActionRole)
reply.addButton(self.tr('Cancel'), QMessageBox.RejectRole)
if reply.exec_() == QMessageBox.RejectRole:
a0.ignore()
return
else:
a0.accept()
self.delete_temporary_dock_windows()
self.save_layout()
Settings.get_instance().save()
self.close_signal.emit()
self.extensions_manager.stop()
for k in self.dock_widgets.keys():
self.dock_widgets[k].widget().closeEvent(a0)
super().closeEvent(a0)
def slot_flush_console(self, level: str, module, content):
"""刷新主窗体执行情况日志
Args:
level: 报错级别,包括 ``info`` , ``warnning`` , ``error`` 。
module: 业务模块名称,例如 数据获取,数据处理,数据探索,统计,模型,可视化,评估
content: 具体显示的内容
"""
create_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 日志记录时间
user = getpass.getuser()
msg = create_time + ' ' + user + ' ' + level.upper() + ' [' + module + ']' + ':' + content
if level == "error":
html = "" + msg + ""
else:
html = "" + msg + ""
console = self.log_output_console # 由于代码重构,这里出现了不同。
console.moveCursor(QTextCursor.End)
console.append(html)
def main():
global t0, logger
# 异常处理设置
# cgitb.enable(format='text')
app = QApplication(sys.argv)
app.setWindowIcon(QIcon(':/logo/source/icons/logo.png')) # 设置应用logo
app.setAttribute(Qt.AA_EnableHighDpiScaling) # 设置应用支持高分屏
# 设置启动画面
splash_image = QPixmap(':/images/source/images/splash.jpg')
splash_image = splash_image.scaled(700, 400)
splash = QSplashScreen(splash_image)
splash.show() # 显示启动界面
app.splash = splash
globals._application = app
# 设置字体
load_fonts(app)
app.default_font = 'Deng'
f = QFont(app.default_font, 10)
app.setFont(f)
# 设置翻译
load_translator(app)
window = MainWindow()
window.showMaximized()
t1 = time.time()
splash.finish(window) # 启动画面结束
window.first_form_display()
logger.info('boot time elapsed:%f s' % (t1 - t0))
logger.info('boot finished at time:' + time.strftime('%H:%M:%S', time.localtime()) + '+%f s' % (
time.time() - int(time.time())))
res = app.exec()
logging.debug("Shutting down, result %d", res)
logging.shutdown()
sys.exit(res)
if __name__ == '__main__':
logger.info('preload_module_time %f' % (time.time() - t0))
main()
一键复制
编辑
Web IDE
原始数据
按行查看
历史