During the development process, a validator is needed to constantly verify and validate. I personally use it while developing the rtmp-publisher and http-flv server.
I, agathakuan, am the same person as the author CHTA-TE KUAN, who is known as agathakuan on CSDN and the owner of the Blogger site "Programming notes".
CSDN: agathakuan-CSDN博客
Currently, I am expanding my publishing platforms and plan to publish articles on CSDN. Therefore, I am transferring some content from my original Blogger site, "Programming notes," to my CSDN blog under the username agathakuan.
In the future, I will publish the same content in English on Google Blogger and in Chinese on CSDN. All articles are originally created by me, and any content cited from other sources will be properly referenced with the URL.
git log --oneline
git reset HEAD~<numberOfCommit>
git commit --amend -m "reset message >second line >third line "
class TcpServer_Tool(QWidget, Ui_Form):
class client_handler(BaseRequestHandler):
class check_socket_sum(threading.Thread):
class client_sendBack(threading.Thread):
client_msg_queue = queue.Queue() client_add_addr_queue = queue.Queue() client_remove_addr_queue = queue.Queue() client_recv_queue =queue.Queue() client_send_msg_queue = queue.Queue() client_socket_list = []
# -*- coding: utf-8 -*- # Form implementation generated from reading ui file 'tcpServer_tool.ui' # # Created by: PyQt5 UI code generator 5.14.1 # # WARNING! All changes made in this file will be lost! from PyQt5 import QtCore, QtGui, QtWidgets class Ui_Form(object): def setupUi(self, Form): Form.setObjectName("Form") Form.resize(1078, 728) font = QtGui.QFont() font.setFamily("Century Gothic") font.setPointSize(11) Form.setFont(font) self.gridLayout = QtWidgets.QGridLayout(Form) self.gridLayout.setObjectName("gridLayout") self.groupBox = QtWidgets.QGroupBox(Form) sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Preferred, QtWidgets.QSizePolicy.Preferred) sizePolicy.setHorizontalStretch(2) sizePolicy.setVerticalStretch(0) sizePolicy.setHeightForWidth(self.groupBox.sizePolicy().hasHeightForWidth()) self.groupBox.setSizePolicy(sizePolicy) self.groupBox.setObjectName("groupBox") self.gridLayout_2 = QtWidgets.QGridLayout(self.groupBox) self.gridLayout_2.setObjectName("gridLayout_2") self.label = QtWidgets.QLabel(self.groupBox) self.label.setObjectName("label") self.gridLayout_2.addWidget(self.label, 0, 0, 1, 1) self.s1_line_1 = QtWidgets.QLineEdit(self.groupBox) self.s1_line_1.setObjectName("s1_line_1") self.gridLayout_2.addWidget(self.s1_line_1, 0, 1, 1, 1) self.label_2 = QtWidgets.QLabel(self.groupBox) self.label_2.setObjectName("label_2") self.gridLayout_2.addWidget(self.label_2, 1, 0, 1, 1) self.s1_line_2 = QtWidgets.QLineEdit(self.groupBox) self.s1_line_2.setObjectName("s1_line_2") self.gridLayout_2.addWidget(self.s1_line_2, 1, 1, 1, 1) self.label_6 = QtWidgets.QLabel(self.groupBox) self.label_6.setObjectName("label_6") self.gridLayout_2.addWidget(self.label_6, 8, 0, 1, 1) self.s1_bt_2 = QtWidgets.QPushButton(self.groupBox) self.s1_bt_2.setObjectName("s1_bt_2") self.gridLayout_2.addWidget(self.s1_bt_2, 4, 1, 1, 1) self.label_3 = QtWidgets.QLabel(self.groupBox) self.label_3.setObjectName("label_3") self.gridLayout_2.addWidget(self.label_3, 2, 0, 1, 1) self.label_4 = QtWidgets.QLabel(self.groupBox) self.label_4.setObjectName("label_4") self.gridLayout_2.addWidget(self.label_4, 5, 0, 1, 1) self.s1_line_4 = QtWidgets.QLineEdit(self.groupBox) self.s1_line_4.setObjectName("s1_line_4") self.gridLayout_2.addWidget(self.s1_line_4, 8, 1, 1, 1) self.s1_line_3 = QtWidgets.QLineEdit(self.groupBox) self.s1_line_3.setObjectName("s1_line_3") self.gridLayout_2.addWidget(self.s1_line_3, 2, 1, 1, 1) self.label_5 = QtWidgets.QLabel(self.groupBox) self.label_5.setObjectName("label_5") self.gridLayout_2.addWidget(self.label_5, 6, 0, 1, 1) self.s1_line_5 = QtWidgets.QLineEdit(self.groupBox) self.s1_line_5.setObjectName("s1_line_5") self.gridLayout_2.addWidget(self.s1_line_5, 9, 1, 1, 1) self.s1_bt_1 = QtWidgets.QPushButton(self.groupBox) self.s1_bt_1.setObjectName("s1_bt_1") self.gridLayout_2.addWidget(self.s1_bt_1, 3, 1, 1, 1) self.s1_com_1 = QtWidgets.QComboBox(self.groupBox) self.s1_com_1.setObjectName("s1_com_1") self.gridLayout_2.addWidget(self.s1_com_1, 5, 1, 1, 1) self.label_7 = QtWidgets.QLabel(self.groupBox) self.label_7.setObjectName("label_7") self.gridLayout_2.addWidget(self.label_7, 9, 0, 1, 1) self.s1_infoBrowser = QtWidgets.QTextBrowser(self.groupBox) self.s1_infoBrowser.setObjectName("s1_infoBrowser") self.gridLayout_2.addWidget(self.s1_infoBrowser, 7, 0, 1, 2) self.gridLayout.addWidget(self.groupBox, 0, 0, 2, 1) self.groupBox_2 = QtWidgets.QGroupBox(Form) sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Preferred, QtWidgets.QSizePolicy.Preferred) sizePolicy.setHorizontalStretch(0) sizePolicy.setVerticalStretch(1) sizePolicy.setHeightForWidth(self.groupBox_2.sizePolicy().hasHeightForWidth()) self.groupBox_2.setSizePolicy(sizePolicy) font = QtGui.QFont() font.setPointSize(11) self.groupBox_2.setFont(font) self.groupBox_2.setObjectName("groupBox_2") self.gridLayout_3 = QtWidgets.QGridLayout(self.groupBox_2) self.gridLayout_3.setObjectName("gridLayout_3") self.s3_com_1 = QtWidgets.QComboBox(self.groupBox_2) self.s3_com_1.setObjectName("s3_com_1") self.gridLayout_3.addWidget(self.s3_com_1, 0, 2, 1, 1) self.s3_check_1 = QtWidgets.QCheckBox(self.groupBox_2) self.s3_check_1.setObjectName("s3_check_1") self.gridLayout_3.addWidget(self.s3_check_1, 0, 0, 1, 1) self.s3_bt_1 = QtWidgets.QPushButton(self.groupBox_2) self.s3_bt_1.setObjectName("s3_bt_1") self.gridLayout_3.addWidget(self.s3_bt_1, 0, 3, 1, 1) self.label_8 = QtWidgets.QLabel(self.groupBox_2) self.label_8.setObjectName("label_8") self.gridLayout_3.addWidget(self.label_8, 0, 1, 1, 1) self.s3_line_1 = QtWidgets.QLineEdit(self.groupBox_2) sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Minimum) sizePolicy.setHorizontalStretch(0) sizePolicy.setVerticalStretch(0) sizePolicy.setHeightForWidth(self.s3_line_1.sizePolicy().hasHeightForWidth()) self.s3_line_1.setSizePolicy(sizePolicy) self.s3_line_1.setObjectName("s3_line_1") self.gridLayout_3.addWidget(self.s3_line_1, 1, 0, 1, 4) self.gridLayout.addWidget(self.groupBox_2, 1, 1, 1, 1) self.tabWidget = QtWidgets.QTabWidget(Form) sizePolicy = QtWidgets.QSizePolicy(QtWidgets.QSizePolicy.Expanding, QtWidgets.QSizePolicy.Expanding) sizePolicy.setHorizontalStretch(7) sizePolicy.setVerticalStretch(3) sizePolicy.setHeightForWidth(self.tabWidget.sizePolicy().hasHeightForWidth()) self.tabWidget.setSizePolicy(sizePolicy) font = QtGui.QFont() font.setPointSize(13) self.tabWidget.setFont(font) self.tabWidget.setObjectName("tabWidget") self.tab = QtWidgets.QWidget() self.tab.setObjectName("tab") self.gridLayout_5 = QtWidgets.QGridLayout(self.tab) self.gridLayout_5.setObjectName("gridLayout_5") self.s2_browser_1 = QtWidgets.QTextBrowser(self.tab) self.s2_browser_1.setObjectName("s2_browser_1") self.gridLayout_5.addWidget(self.s2_browser_1, 0, 0, 1, 1) self.tabWidget.addTab(self.tab, "") self.tab_3 = QtWidgets.QWidget() self.tab_3.setObjectName("tab_3") self.gridLayout_4 = QtWidgets.QGridLayout(self.tab_3) self.gridLayout_4.setObjectName("gridLayout_4") self.label_9 = QtWidgets.QLabel(self.tab_3) self.label_9.setObjectName("label_9") self.gridLayout_4.addWidget(self.label_9, 0, 0, 1, 1) self.tabWidget.addTab(self.tab_3, "") self.gridLayout.addWidget(self.tabWidget, 0, 1, 1, 1) self.retranslateUi(Form) self.tabWidget.setCurrentIndex(0) QtCore.QMetaObject.connectSlotsByName(Form) def retranslateUi(self, Form): _translate = QtCore.QCoreApplication.translate Form.setWindowTitle(_translate("Form", "Form")) self.groupBox.setTitle(_translate("Form", "TCP Server setting")) self.label.setText(_translate("Form", "host IP")) self.label_2.setText(_translate("Form", "Port")) self.label_6.setText(_translate("Form", "user")) self.s1_bt_2.setText(_translate("Form", "re-config")) self.label_3.setText(_translate("Form", "timeout")) self.label_4.setText(_translate("Form", "client")) self.label_5.setText(_translate("Form", "info ")) self.s1_bt_1.setText(_translate("Form", "open")) self.label_7.setText(_translate("Form", "time")) self.groupBox_2.setTitle(_translate("Form", "Send")) self.s3_check_1.setText(_translate("Form", "send in hex")) self.s3_bt_1.setText(_translate("Form", "send")) self.label_8.setText(_translate("Form", "send to :")) self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab), _translate("Form", "general recv")) self.label_9.setText(_translate("Form", "power by Ingrid kuan contact me at：agathakuannew@gmail.com")) self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_3), _translate("Form", "info page"))
#prototype@2020-01-13 #coding: utf-8 from socketserver import BaseRequestHandler, ThreadingTCPServer import socket import threading import time import sys import datetime import queue import struct from PyQt5 import QtWidgets from PyQt5.QtWidgets import * from PyQt5.QtCore import * from PyQt5.QtGui import * from PyQt5.QtCore import QTimer,QTime,QDateTime,Qt,QSize from tcpServer_tool import Ui_Form client_msg_queue = queue.Queue() client_add_addr_queue = queue.Queue() client_remove_addr_queue = queue.Queue() client_recv_queue =queue.Queue() client_send_msg_queue = queue.Queue() client_socket_list = [] class client_sendBack(threading.Thread): # 另一个线程任务，用于计数，暂且叫他计数线程 def __init__(self, conn, msg): super(client_sendBack, self).__init__() self.conn= conn self.addrAndPort = conn.getpeername() self.send_msg = msg def run(self): #https://stackoverflow.com/questions/41250805/how-do-i-print-the-local-and-remote-address-and-port-of-a-connected-socket self.conn.sendall(self.send_msg.encode('utf-8')) class check_socket_sum(threading.Thread): def __init__(self, input_q): super(check_socket_sum, self).__init__() self.communicate_queue = input_q self.count = 0 def run(self): global client_socket_list while True: self.communicate_queue.put(str(len(client_socket_list))) self.count = self.count+1 time.sleep(1) class client_handler(BaseRequestHandler): ip = '' port = 0 timeOut = 10*60 def setup(self): global client_add_addr_queue,client_remove_addr_queue global client_msg_queue, client_socket_list self.ip = self.client_address[0].strip() # 獲取客戶端的ip self.port = self.client_address[1] # 獲取客戶端的port self.request.settimeout(self.timeOut) # 對socket設定超時時間 self.msg_queue = client_msg_queue client_socket_list.append(self.request) client_add_addr_queue.put(self.ip) self.disconnect_queue = client_remove_addr_queue def handle(self): global client_send_msg_queue address, pid = self.client_address print('%s connected!' % address) while True: data = self.request.recv(128) if len(data) > 0: response = '{}:{}'.format(self.ip, data) client_recv_queue.put(response) def finish(self): print("client is disconnect!") self.disconnect_queue.put(self.ip) try: client_socket_list.remove(self.server) except: pass class TcpServer_Tool(QWidget, Ui_Form): def __init__(self): super(TcpServer_Tool, self).__init__() self.setupUi(self) self.setWindowTitle("TCP Server to multiple client") self.client_list = [] self.client_num_queue = queue.Queue() self.infoInit() self.functionInit() def infoInit(self): self.s1_line_1.setText(str(self.get_host_ip())) self.s1_line_2.setText("8080") self.s1_line_3.setText("10") self.s1_line_4.setText("default user") def functionInit(self): self.sys_timer = QTimer(self) self.sys_timer.timeout.connect(self.update_time) self.sys_timer.start(500) self.print_update_timer = QTimer(self) self.print_update_timer.timeout.connect(self.update_ui) self.print_update_timer.start(100) watch_client_thread = check_socket_sum(self.client_num_queue) watch_client_thread.start() self.s1_bt_1.clicked.connect(self.server_open) self.s3_bt_1.clicked.connect(self.send_to_client) self.s1_browser_print(char="system start\r\n") def send_to_client(self): global client_socket_list ip = self.s3_com_1.currentText() msg = self.s3_line_1.text() send_msg = ip+","+msg self.s1_browser_print(char=send_msg) if ip == client_socket_list[self.s3_com_1.currentIndex()].getpeername()[0]: terminal_socket = client_socket_list[self.s3_com_1.currentIndex()] client_send_thread = client_sendBack(terminal_socket, msg) client_send_thread.start() def server_open(self): HOST = str(self.get_host_ip()) PORT = int(self.s1_line_2.text()) ADDR = (HOST, PORT) self.tcpserver = ThreadingTCPServer(ADDR, client_handler) # 参数为监听地址和已建立连接的处理类 self.tcpserver_thread = threading.Thread(target=self.tcpserver.serve_forever) # 创建线程，线程用于TCP多线程 self.tcpserver_thread.start() self.s1_line_1.setEnabled(False) self.s1_line_2.setEnabled(False) self.s1_line_3.setEnabled(False) def update_ui(self): global client_add_addr_queue, client_remove_addr_queue global client_recv_queue data = self.client_num_queue.get() data = "connected " + data + " " + "client" self.s1_browser_print(char=data) if client_add_addr_queue.empty() == False: self.updateClientList(func='add',arg=client_add_addr_queue.get()) if client_remove_addr_queue.empty() == False: #https://stackoverflow.com/questions/53828161/pyqt-selected-combobox-item-should-remove-an-item-in-another-combobox idx = self.s3_com_1.findText(client_remove_addr_queue.get()) self.updateClientList(func='remove', arg=idx) if client_recv_queue.empty() == False: self.s2_browser_print(char=client_recv_queue.get()) def update_time(self): time_format = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.s1_line_5.setText(time_format) def updateClientList(self, func, arg): if func == 'add': self.s3_com_1.addItem(arg) self.s1_com_1.addItem(arg) if func == 'remove': self.s3_com_1.removeItem(arg) self.s1_com_1.removeItem(arg) def get_host_ip(self): #https://www.chenyudong.com/archives/python-get-local-ip-graceful.html#hostnameIP try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) ip = s.getsockname()[0] finally: s.close() return ip def s1_browser_print(self, char=" "): self.s1_infoBrowser.insertPlainText(char+'\r\n') textCursor = self.s1_infoBrowser.textCursor() textCursor.movePosition(textCursor.End) self.s1_infoBrowser.setTextCursor(textCursor) def s2_browser_print(self, char=" "): self.s2_browser_1.insertPlainText(char+'\r\n') textCursor = self.s2_browser_1.textCursor() textCursor.movePosition(textCursor.End) self.s2_browser_1.setTextCursor(textCursor) if __name__ == "__main__": print("system start") app = QApplication(sys.argv) ui = TcpServer_Tool() ui.show() sys.exit(app.exec_())
