Python PyQt5.QtCore 模块，QThread() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用PyQt5.QtCore.QThread()。
def check_for_updates(self, from_menu):
"""
Opens and runs the update checker on a separate thread. Sets self.from_menu so that other dialogs know the
updater has been ran by the user, this will result in different dialog behaviour
"""
self.update_thread = QtCore.QThread()
self.update_checker = UpdateChecker(self.version)
self.update_checker.moveToThread(self.update_thread)
self.update_thread.started.connect(self.update_checker.run)
self.update_checker.update_available_signal.connect(self.display_update)
if from_menu:
self.update_checker.no_update_signal.connect(self.no_update_available_dialog)
self.update_checker.finished.connect(self.update_thread.quit)
self.update_checker.finished.connect(self.update_checker.deleteLater)
self.update_thread.finished.connect(self.update_thread.deleteLater)
self.update_thread.start()
def main():
app = QtWidgets.QApplication(sys.argv)
queue = Queue()
thread = QtCore.QThread()
receiver = MessageReceiver(queue)
window = RedditDownloaderGUI(queue, receiver)
receiver.output_signal.connect(window.update_output)
receiver.moveToThread(thread)
thread.started.connect(receiver.run)
receiver.finished.connect(thread.quit)
receiver.finished.connect(receiver.deleteLater)
thread.finished.connect(thread.deleteLater)
thread.start()
window.show()
sys.exit(app.exec_())
def __init__(self):
if not hasattr(self, 'threading_queue'):
raise AttributeError('main_gui object is missing a threading_queue')
if not hasattr(self, 'threading_signal'):
raise AttributeError('main_gui object is missing a threading_signal')
t = QThread()
# noinspection PyUnresolvedReferences
w = MainGuiWorker(self.threading_signal, self.threading_queue)
w.moveToThread(t)
# noinspection PyUnresolvedReferences
t.started.connect(w.run)
# noinspection PyUnresolvedReferences
self.threading_signal.connect(self._do)
t.start()
self.__threading = {
'thread': t,
'worker': w,
}
self.__pool = ThreadPool(1, 'main_ui', False)
def __init__(self):
"""
Initialise class
"""
QtWidgets.QWidget.__init__(self)
self.setupUi(self)
output.PPTevents.updateSlide = self.powerpointSlides.setCurrentRow
self.contentControls.setCurrentIndex(0)
# TODO not implemented
"""
self.mediaProgressSeek.setMouseTracking(True)
self.mediaProgressSeek.leaveEvent=self.seekMouse
self.mediaProgressSeek.mouseMoveEvent=self.seekMouse
self.mediaProgressSeek.mouseReleaseEvent=lambda e: output.seek(e.x()/self.mediaProgressSeek.width())
"""
# Update video scrubber position
class ProgressBarUpdater(QtCore.QThread):
tick = QtCore.pyqtSignal(int)
def run(self):
while True:
self.sleep(1)
self.tick.emit(oncue.lib.utils.confine(int(output.VLCposition() * 1000), 0, 1000))
self.mediaProgressBarThread = ProgressBarUpdater()
self.mediaProgressBarThread.tick.connect(
lambda value: self.mediaProgressBar.setValue(value) or self.mediaProgressBar.repaint())
def __init__(self, parent=None):
super().__init__(parent)
self.setContextMenuPolicy(QtCore.Qt.CustomContextMenu)
self.customContextMenuRequested.connect(self.context_menu)
self.setItemsExpandable(False)
self.doubleClicked.connect(self.double_click)
self.setIconSize(QtCore.QSize(36, 36))
self.setAlternatingRowColors(True)
self.setIndentation(10)
self._model = None
self._worker_thread = QtCore.QThread(self)
self._worker = plexdesktop.workers.HubWorker()
self._worker.moveToThread(self._worker_thread)
self._worker.result_ready.connect(self._add_container)
self.request_container.connect(self._worker.run)
self._worker_thread.start()
self.delegate = plexdesktop.delegates.ListDelegate(self)
self.setItemDelegate(self.delegate)
self.goto_hub.connect(self.load_hub)
def __init__(self, ip, port, ui_widget=None, hm=None, logr=None, window=None):
global ui, MainWindow, home, logger, html_default_arr, getdb
global BASEDIR, TMPDIR, OSNAME
QtCore.QThread.__init__(self)
self.ip = ip
self.port = int(port)
self.cert_signal.connect(generate_ssl_cert)
self.media_server_start.connect(media_server_started)
self.httpd = None
logger = logr
ui = ui_widget
home = hm
MainWindow = window
arg_dict = ui.get_parameters_value(
r='html_default_arr', b='BASEDIR', t='TMPDIR')
#logger.info(arg_dict)
html_default_arr = arg_dict['html_default_arr']
BASEDIR = arg_dict['BASEDIR']
TMPDIR = arg_dict['TMPDIR']
OSNAME = os.name
if ui.getdb is None:
getdb = ServerLib(ui, home, BASEDIR, TMPDIR, logger)
elif isinstance(ui.getdb, ServerLib):
logger.info('--server--initiated---2477--')
getdb = ui.getdb
def __init__(
self, ui_widget, logr, tmp, name, url=None, direct_url=None,
copy_fanart=None, copy_poster=None, copy_summary=None, use_search=None):
QtCore.QThread.__init__(self)
global ui, logger, TMPDIR, site
ui = ui_widget
logger = logr
TMPDIR = tmp
self.name = name
self.url = url
self.direct_url = direct_url
self.copy_fanart = copy_fanart
self.copy_poster = copy_poster
self.copy_summary = copy_summary
self.use_search = use_search
self.summary_signal.connect(copy_information)
site = ui.get_parameters_value(s='site')['site']
def __init__(self, ui_widget, url, img_list=None, get_text=None):
QtCore.QThread.__init__(self)
global ui
self.url = url
self.interval = 1
self.picn = 'picn'
ui = ui_widget
if img_list:
"""img_list=[img_url, date, ep_url, local_path, site, img_key, dest_dir]"""
self.img_list = img_list.copy()
else:
self.img_list = []
if get_text:
self.get_text = True
else:
self.get_text = False
def __init__(self):
QtCore.QThread.__init__(self)
self.mutex = QtCore.QMutex()
self.pack_opener = None
self._isRunning = False
def __init__(self, file: str, curse: CurseAPI, path: str, fname=False, callback=False):
super().__init__()
self.callback = callback
self.setWindowTitle(translate("downloading.update"))
self.layout = QVBoxLayout(self)
self.progress = QProgressBar()
self.layout.addWidget(self.progress)
self.show()
self.downloader = FileDownloaderThread(file, curse, path, fname)
self.downloader.done.connect(self.download_done)
self.downloader.update.connect(self.progress.setValue)
self.download_thread = QThread()
self.downloader.moveToThread(self.download_thread)
self.download_thread.started.connect(self.downloader.download)
self.download_thread.start()
def __init__(self, file: CurseFile, curse: CurseAPI, instance, initmods):
super().__init__()
self.initmods = initmods
self.setWindowTitle(translate("downloading.mod").format(file.name))
self.layout = QVBoxLayout(self)
self.progress = QProgressBar()
self.layout.addWidget(self.progress)
self.show()
self.downloader = ModDownloaderThread(file, curse, instance)
self.downloader.done.connect(self.download_done)
self.downloader.update.connect(self.progress.setValue)
self.download_thread = QThread()
self.downloader.moveToThread(self.download_thread)
self.download_thread.started.connect(self.downloader.download)
self.download_thread.start()
def __init__(self):
super(QWidget, self).__init__()
layout = QVBoxLayout()
self.edit = QTextEdit()
self.thread = QThread()
self.setupConnections()
self.edit.setWindowTitle("QTextEdit Standard Output Redirection")
layout.addWidget(self.edit)
self.setLayout(layout)
self.show()
def _show_camera(self):
"""
Shows the camera window
"""
print("Showing Camera")
if not isinstance(self._camera_window, CameraWindow):
self._camera_window = CameraWindow()
self._camera_window.before_close_event = self._hide_camera
self._camera_window.show()
if self._camera is None or self._camera_thread is None:
self._camera_thread = QThread()
self._camera = CameraThread()
self._camera.frame_ready.connect(self.process_data)
self._camera.moveToThread(self._camera_thread)
self._camera_thread.started.connect(self._camera.start_processing)
self._camera_thread.start()
def _start_camera(self):
"""
Starts the camera processing
"""
print("Starting Camera")
if self._camera is None or self._camera_thread is None:
self._camera_thread = QThread()
self._camera = CameraThread()
self._camera.moveToThread(self._camera_thread)
self._camera.frame_ready.connect(self.process_data)
self._camera_thread.started.connect(self._camera.init)
self._camera_thread.start()
QtCore.QMetaObject.invokeMethod(
self._camera, 'start_processing', Qt.Qt.QueuedConnection)
self._update_params()
def start_reddit_extractor_thread(self, download_type):
"""Moves the extractor to a different thread and calls the appropriate function for the type of download"""
self.stop_download.connect(self.reddit_extractor.stop_download)
self.thread = QtCore.QThread()
self.reddit_extractor.moveToThread(self.thread)
if download_type == 'USER':
self.thread.started.connect(self.reddit_extractor.validate_users)
elif download_type == 'SUBREDDIT':
self.thread.started.connect(self.reddit_extractor.validate_subreddits)
elif download_type == 'USERS_AND_SUBREDDITS':
self.thread.started.connect(self.reddit_extractor.validate_users_and_subreddits)
elif download_type == 'UNFINISHED':
self.thread.started.connect(self.reddit_extractor.finish_downloads)
self.reddit_extractor.remove_invalid_user.connect(self.remove_invalid_user)
self.reddit_extractor.downloaded_users_signal.connect(self.fill_downloaded_users_list)
self.reddit_extractor.setup_progress_bar.connect(self.setup_progress_bar)
self.reddit_extractor.update_progress_bar_signal.connect(self.update_progress_bar)
self.reddit_extractor.unfinished_downloads_signal.connect(self.set_unfinished_downloads)
self.reddit_extractor.finished.connect(self.thread.quit)
self.reddit_extractor.finished.connect(self.reddit_extractor.deleteLater)
self.thread.finished.connect(self.thread.deleteLater)
self.thread.finished.connect(self.finished_download_gui_shift)
self.thread.start()
def download_user_samples(self):
"""
Creates an instance of the RedditExtractor class and moves it to another thread where it then downloads the
specified number of posts from the found users
"""
if len(self.found_users) > 0 and self.watchlist_download_sample_spinbox_2.value() > 0:
self.found_user_output.append('Starting Download\n')
self.reddit_extractor = RedditExtractor(self.found_users, None, self.queue,
self.watchlist_download_sample_spinbox_2.value(), self.save_path,
None, None, False, False, None, None, None)
self.user_finder_download_thread = QtCore.QThread()
self.reddit_extractor.moveToThread(self.user_finder_download_thread)
self.user_finder_download_thread.started.connect(self.reddit_extractor.validate_users)
self.reddit_extractor.finished.connect(self.user_finder_download_thread.quit)
self.reddit_extractor.finished.connect(self.reddit_extractor.deleteLater)
self.user_finder_download_thread.finished.connect(self.user_finder_download_thread.deleteLater)
self.user_finder_download_thread.finished.connect(self.download_finished)
self.user_finder_download_thread.start()
elif len(self.found_users) > 0 >= self.watchlist_download_sample_spinbox_2.value():
pass
else:
self.found_user_output.append('No users found that meet criteria\n')
self.download_finished()
def startFetchTorrentThread(self, action, arguments = {}):
worker = FetchTorrentWorker(action, arguments)
thread = QThread()
self.threads.append((thread, worker)) # need to store worker too otherwise will be gc'd
worker.moveToThread(thread)
worker.finished.connect(self.onFetchTorrentThreadResponse)
thread.started.connect(worker.run)
thread.start()
def open_connection(self, address, status_port, data_port):
self.statusBar().showMessage("Open session to {}:{}".format(address, status_port), 2000)
socket = self.context.socket(zmq.DEALER)
socket.identity = "Matplotlib_Qt_Client".encode()
socket.connect("tcp://{}:{}".format(address, status_port))
socket.send(b"WHATSUP")
poller = zmq.Poller()
poller.register(socket, zmq.POLLOUT)
time.sleep(0.1)
evts = dict(poller.poll(100))
if socket in evts:
try:
reply, desc = [e.decode() for e in socket.recv_multipart()]
desc = json.loads(desc)
self.statusBar().showMessage("Connection established. Pulling plot information.", 2000)
except:
self.statusBar().showMessage("Could not connect to server.", 2000)
return
else:
self.statusBar().showMessage("Server did not respond.", 2000)
socket.close()
self.construct_plots(desc)
# Actual data listener
if self.listener_thread:
self.Datalistener.running = False
self.listener_thread.quit()
self.listener_thread.wait()
self.listener_thread = QtCore.QThread()
self.Datalistener = DataListener(address, data_port)
self.Datalistener.moveToThread(self.listener_thread)
self.listener_thread.started.connect(self.Datalistener.loop)
self.Datalistener.message.connect(self.data_signal_received)
self.Datalistener.finished.connect(self.stop_listening)
QtCore.QTimer.singleShot(0, self.listener_thread.start)
def __init__(self,url):
QtCore.QThread.__init__(self)
self.url = url
self.interval = 1
def __init__(self,pic,label_num):
QtCore.QThread.__init__(self)
self.picn = pic
self.label_num = label_num
self.interval = 1
def __init__(self,site,name,pgn,p,n):
QtCore.QThread.__init__(self)
self.site = site
self.name = name
self.pgn = pgn
self.pic = p
self.label = n
def playItem(self):
"""
Plays the selected item
"""
data = self.listItemsPrimary.currentItem().data(256)
output.load(data)
if data["type"] == "media":
# Plays video
self.mediaProgressBarThread.start()
self.mediaProgressBarThread.setPriority(QtCore.QThread.TimeCriticalPriority)
self.mediaControls_PLAY.click()
self.contentControls.setCurrentIndex(2)
elif data["type"] == "powerpoint":
# Clear existing content in the slide preview list
self.powerpointSlides.clear()
# Connect to PowerPoint COM
PPTApplication = win32com.client.Dispatch("PowerPoint.Application")
Presentation = PPTApplication.Presentations.Open(data["path"].replace("/", "\\"),
WithWindow=False)
# Create slide previews
temp = tempfile.TemporaryDirectory().name
Presentation.Export(temp, "png")
i = 1
for file in glob.iglob(temp + "\\*.PNG"):
item = QtWidgets.QListWidgetItem()
item.setIcon(QtGui.QIcon(file))
item.setText(str(i))
item.setTextAlignment(QtCore.Qt.AlignCenter)
i += 1
self.powerpointSlides.addItem(item)
self.contentControls.setCurrentIndex(1)
else:
# 'unknown' case - Hide controls
self.contentControls.setCurrentIndex(0)
def __init__(self):
"""Init thread."""
QtCore.QThread.__init__(self)
if FlaskThread._single:
raise FlaskThread._single
FlaskThread._single = self
self.application = flask_app
def start(self, table_name, sql_insert_queue, sql_insert_mutex, sql_insert_condition):
# table_name = "dfdfd"
#
# sql_insert_queue = Queue.Queue()
# sql_insert_mutex = QtCore.QMutex()
# sql_insert_condition = QtCore.QWaitCondition()
self.rw_lock = QtCore.QReadWriteLock()
mftsize = (os.path.getsize(self.options.filename)) / 1024
self.mftsize = mftsize
filename = self.options.filename
self.mft_seqs_flag_list = [0] * self.mftsize # 0 empty, 1 parsering, 2 done.
self.mft = {}
self.thread_no = max(1, QtCore.QThread.idealThreadCount())
self.thread_no = 10 # 1'26''
self.thread_no = 1 # 1'12''
printself.thread_no
self.thread_pool = [MftWorkerThread(self.mft, self.mft_seqs_flag_list, mftsize, filename, self.rw_lock,
sql_insert_queue, sql_insert_mutex, sql_insert_condition,
table_name, self.options,
parent=self.parent) for _ in range(self.thread_no)]
for thread in self.thread_pool:
# thread.add_row_to_model_SIGNAL.connect(self.target_slot)
# thread.update_progress_SIGNAL.connect(self.update_progress_slot)
thread.start()
def initThumbs(self) -> None:
framesize = self.parent.videoService.framesize()
thumbsize = QSize(VideoService.ThumbSize.TIMELINE.value.height() * (framesize.width() / framesize.height()),
VideoService.ThumbSize.TIMELINE.value.height())
positions, frametimes = [], []
thumbs = int(math.ceil((self.rect().width() - (self.offset * 2)) / thumbsize.width()))
for pos in range(thumbs):
val = QStyle.sliderValueFromPosition(self.minimum(), self.maximum(),
(thumbsize.width() * pos) - self.offset,
self.rect().width() - (self.offset * 2))
positions.append(val)
positions[0] = 1000
[frametimes.append(self.parent.delta2QTime(msec).toString(self.parent.timeformat)) for msec in positions]
class ThumbWorker(QObject):
completed = pyqtSignal(list)
def __init__(self, media: str, times: list, size: QSize):
super(ThumbWorker, self).__init__()
self.media = media
self.times = times
self.size = size
@pyqtSlot()
def generate(self):
frames = list()
[frames.append(VideoService.captureFrame(self.media, frame, self.size)) for frame in self.times]
self.completed.emit(frames)
self.thumbsThread = QThread(self)
self.thumbsWorker = ThumbWorker(self.parent.currentMedia, frametimes, thumbsize)
self.thumbsWorker.moveToThread(self.thumbsThread)
self.thumbsThread.started.connect(self.parent.sliderWidget.setLoader)
self.thumbsThread.started.connect(self.thumbsWorker.generate)
self.thumbsThread.finished.connect(self.thumbsThread.deleteLater, Qt.DirectConnection)
self.thumbsWorker.completed.connect(self.buildTimeline)
self.thumbsWorker.completed.connect(self.thumbsWorker.deleteLater, Qt.DirectConnection)
self.thumbsWorker.completed.connect(self.thumbsThread.quit, Qt.DirectConnection)
self.thumbsThread.start()
def __init__(self, mutex, item, destination, parent=None):
super().__init__(parent)
self.id = hash(item)
self.item = item
self.destination = destination
self.cancelled = False
self.paused = False
self._pause = QtCore.QWaitCondition()
self.thread = QtCore.QThread()
self.worker = DownloadWorker(mutex)
self.worker.moveToThread(self.thread)
self.thread.start()
def __init__(self, data, parent=None):
super().__init__(parent)
self.root_item = TreeItem()
self.setupModelData(data, self.root_item)
self.thumb_queue = plexdesktop.utils.Queue()
self._thumb_thread = QtCore.QThread(self)
self._thumb_worker = plexdesktop.workers.QueueThumbWorker()
self._thumb_worker.moveToThread(self._thumb_thread)
self._thumb_worker.result_ready.connect(self._update_thumb, QtCore.Qt.QueuedConnection)
self.work_thumbs.connect(self._thumb_worker.process, QtCore.Qt.QueuedConnection)
self._thumb_thread.start()
self.work_thumbs.emit(self.thumb_queue)
def __init__(self, parent=None):
super().__init__(parent)
self.container = None
self.container_thread = QtCore.QThread()
self.thumb_thread = QtCore.QThread()
self.container_worker = plexdesktop.workers.ContainerWorker()
self.thumb_worker = plexdesktop.workers.QueueThumbWorker()
self.container_worker.moveToThread(self.container_thread)
self.thumb_worker.moveToThread(self.thumb_thread)
self.container_worker.result_ready.connect(self._add_container)
self.container_worker.container_updated.connect(self._update_container)
self.container_worker.finished.connect(self._done)
self.thumb_worker.result_ready.connect(self._update_thumb)
self.work_container.connect(self.container_worker.run)
self.work_container.connect(self.working.emit)
self.work_container_fetch_more.connect(self.container_worker.fetch_more)
self.work_container_fetch_more.connect(self.working.emit)
# self.work_container_fetch_next_page.connect(self.container_worker.fetch_next_page_object)
# self.work_container_fetch_next_page.connect(self.working.emit)
# self.work_thumb.connect(self.thumb_worker.get_thumb)
self.work_thumbs.connect(self.thumb_worker.process)
self.thumb_queue = plexdesktop.utils.Queue()
self.container_thread.start()
self.thumb_thread.start()
self.work_thumbs.emit(self.thumb_queue)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._timeline_thread = QtCore.QThread(self)
self._timeline_updater = TimelineUpdater()
self._timeline_updater.moveToThread(self._timeline_thread)
self.update_timeline.connect(self._timeline_updater.update)
self._timeline_thread.start()
self.plex_play_queue = None
self.plex_current_item = None
self.plex_next_item = None
self.timeline_timer = QtCore.QElapsedTimer()
def __init__(
self, ip, port, key=None, client_arr=None, https_conn=None,
cert_file=None, ui=None):
global thread_signal, media_server_key, client_auth_arr, ui_player, local_ip_arr
QtCore.QThread.__init__(self)
self.ip = ip
self.port = int(port)
media_server_key = key
client_auth_arr = client_arr
self.https_allow = https_conn
self.cert_file = cert_file
ui_player = ui
local_ip_arr = ['127.0.0.1', '0.0.0.0', ip]
if ui is not None:
self.ui = ui
def __init__(self, v1, v2, v3, v4, row=None, from_client=None):
QtCore.QThread.__init__(self)
self.handle = v1
self.cnt = v2
self.cnt_limit = v3
self.session = v4
if row:
self.current_index = row
else:
self.current_index = 0
self.from_client = from_client
self.start_next = False
self.session_signal.connect(session_finished)
self.progress_signal.connect(print_progress)
self.progress_signal_end.connect(print_progress_complete)
def __init__(self, ui_widget, meta, mode, row, epn_arr):
QtCore.QThread.__init__(self)
global ui
ui = ui_widget
self.ui = ui_widget
self.meta = meta
self.mode = mode
self.row = row
self.image_dict_list = {}
self.dest_dir = ''
self.site = ''
self.epn_arr = ''
self.local_arr = epn_arr.copy()
self.mysignal.connect(update_playlist_widget)
self.imagesignal.connect(update_image_list)
def __init__(self, ui_widget, music_db, music_file, music_file_bak):
QtCore.QThread.__init__(self)
global ui
ui = ui_widget
self.music_db = music_db
self.music_file = music_file
self.music_file_bak = music_file_bak
self.music_db_update.connect(update_music_db_onstart)
def __init__(self, ui_widget, command):
QtCore.QThread.__init__(self)
global ui
ui = ui_widget
self.command = command
self.wait_signal.connect(start_new_player_instance)
def __init__(self, ui_widget, interval=None, ip_file=None):
QtCore.QThread.__init__(self)
global ui
ui = ui_widget
if not interval:
self.interval = (3600)
else:
self.interval = interval * (3600)
self.got_ip_signal.connect(set_my_ip_function)
self.ip_file = ip_file
def __init__(self, ui_widget, logr, epn_arr, update_pl, title_list):
QtCore.QThread.__init__(self)
global ui, logger
ui = ui_widget
logger = logr
self.epn_arr = epn_arr.copy()
self.update_pl = update_pl
self.title_list = title_list
self.setThumb.connect(apply_thumbnail_to_playlist)
def __init__(self, ui_widget, logr, browse_cnt, picn, val, fit_size,
widget_size, length, nameEpn, path=None):
QtCore.QThread.__init__(self)
global ui, logger
ui = ui_widget
logger = logr
self.browse_cnt = browse_cnt
self.picn = picn
self.val = val
self.fit_size = fit_size
self.widget_size = widget_size
self.length = length
self.nameEpn = nameEpn
self.setThumbGrid.connect(apply_to_thumbnail_grid)
self.path = path
def __init__(
self, ui_widget, logr, site, opt, siteName, video_local_stream,
name, ei, category, from_cache):
QtCore.QThread.__init__(self)
global ui, logger
ui = ui_widget
logger = logr
self.site = site
self.opt = opt
self.siteName = siteName
self.video_local_stream = video_local_stream
self.name = name
self.ei = ei
self.category = category
self.from_cache = from_cache
def __init__(self, name, logr, tmp):
QtCore.QThread.__init__(self)
global logger, TMPDIR
self.name1 = name
self.interval = 1
logger = logr
TMPDIR = tmp
def __init__(self, ui_widget, broadcast=None):
QtCore.QThread.__init__(self)
global ui
ui = ui_widget
if broadcast:
ui.broadcast_server = True
self.broadcast_signal.connect(broadcast_server_signal)
def __init__(self, parent):
QtCore.QThread.__init__(self, parent=parent)
self.stop = False
def __init__(self, callable, args, kwargs):
QtCore.QThread.__init__(self)
self.args = args
self.kwargs = kwargs
self.callable = callable
def __init__(self, workq):
QtCore.QThread.__init__(self)
self.workq = workq
def __init__(self, opts):
QtCore.QThread.__init__(self)
self.mutex = QtCore.QMutex()
self.opts = opts
self.email, self.password = "", ""
self.anonymous = True
def __init__(self, file: CurseFile, curse: CurseAPI, pack: CurseModpack):
super().__init__()
self.setWindowTitle(translate("downloading.pack").format(pack.project.title))
self.layout = QVBoxLayout(self)
self.label = QLabel()
self.layout.addWidget(self.label)
self.progress = QProgressBar()
self.layout.addWidget(self.progress)
self.prog2 = QProgressBar()
self.layout.addWidget(self.prog2)
self.show()
self.downloader = PackDownloaderThread(file, curse, pack)
self.downloader.done.connect(self.download_done)
self.downloader.bar1.connect(self.progress.setValue)
self.downloader.bar2.connect(self.prog2.setValue)
self.downloader.setLabel.connect(self.label.setText)
self.download_thread = QThread()
self.downloader.moveToThread(self.download_thread)
self.download_thread.started.connect(self.downloader.download)
self.download_thread.start()
def botMove(self):
self.moveThread = QThread()
self.moveCalculation = MoveCalculation(self.on_turn, self.board)
self.moveCalculation.calculated.connect(self.makeBotMove)
self.moveCalculation.moveToThread(self.moveThread)
self.moveThread.started.connect(self.moveCalculation.run)
self.moveCalculation.terminated.connect(self.moveThread.quit)
self.moveThread.start()
def botMove(self):
self.moveThread = QThread()
self.moveCalculation = WaitForMove(self.bot, self.board)
self.moveCalculation.done.connect(self.makeBotMove)
self.moveCalculation.error.connect(self.botMoveError)
self.moveCalculation.moveToThread(self.moveThread)
self.moveThread.started.connect(self.moveCalculation.run)
self.moveCalculation.terminated.connect(self.moveThread.quit)
self.moveThread.start()
def __init__(self, name, port, parent=None):
super(ServerGame, self).__init__(parent)
self.qBoard = QMacroBoard(self.onButtonClick)
self.statusBar = QLabel()
self.statusBar.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
self.statusBar.hide()
self.titleBar = QLabel()
self.titleBar.setAlignment(Qt.AlignHCenter | Qt.AlignVCenter)
self.titleBar.hide()
layout = QVBoxLayout()
layout.addWidget(self.titleBar)
layout.addWidget(self.qBoard)
layout.addWidget(self.statusBar)
self.setLayout(layout)
self.server = game.players.human.ServerPlayer(name, port)
self.opponentConnected = False
self.board = None
self.last_click = None
self.qBoard.updateBoard(game.boards.Macroboard())
self.qBoard.setClickEnabled(False)
self.requestThread = QThread()
self.requestWorker = RequestHandler(self)
self.requestWorker.waitingRequest.connect(self.waitingOpponentMessage)
self.requestWorker.requestAccepted.connect(self.moveRequest)
self.requestWorker.error.connect(self.serverError)
self.requestWorker.moveToThread(self.requestThread)
self.requestThread.started.connect(self.requestWorker.run)
self.requestWorker.terminated.connect(self.requestThread.quit)
self.requestThread.start()
def opponentMove(self):
self.requestThread = QThread()
self.requestMaker = MoveRequestMaker(self)
self.requestMaker.serverResponsed.connect(self.makeOpponentMove)
self.requestMaker.error.connect(self.serverError)
self.requestMaker.moveToThread(self.requestThread)
self.requestThread.started.connect(self.requestMaker.run)
self.requestMaker.terminated.connect(self.requestThread.quit)
self.requestThread.start()