mirror of
https://github.com/kbumsik/VirtScreen.git
synced 2025-03-09 15:40:18 +00:00
split single virtscreen.py into submodules
This commit is contained in:
Bumsik Kim
parent
2ea15b8943
commit
96c6066a91
9 changed files with 929 additions and 897 deletions
355
virtscreen/qt_backend.py
Normal file
355
virtscreen/qt_backend.py
Normal file
|
|
@ -0,0 +1,355 @@
|
|||
"""GUI backend"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import os
|
||||
import shutil
|
||||
import atexit
|
||||
import time
|
||||
|
||||
from PyQt5.QtCore import QObject, pyqtProperty, pyqtSlot, pyqtSignal, Q_ENUMS
|
||||
from PyQt5.QtGui import QCursor
|
||||
from PyQt5.QtQml import QQmlListProperty
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
from netifaces import interfaces, ifaddresses, AF_INET
|
||||
|
||||
from .display import DisplayProperty
|
||||
from .xrandr import XRandR
|
||||
from .process import AsyncSubprocess, SubprocessWrapper
|
||||
from .path import (DATA_PATH, CONFIG_PATH, DEFAULT_CONFIG_PATH,
|
||||
X11VNC_PASSWORD_PATH, X11VNC_LOG_PATH)
|
||||
|
||||
|
||||
class Backend(QObject):
|
||||
""" Backend class for QML frontend """
|
||||
|
||||
class VNCState:
|
||||
""" Enum to indicate a state of the VNC server """
|
||||
OFF = 0
|
||||
ERROR = 1
|
||||
WAITING = 2
|
||||
CONNECTED = 3
|
||||
|
||||
Q_ENUMS(VNCState)
|
||||
|
||||
# Signals
|
||||
onVirtScreenCreatedChanged = pyqtSignal(bool)
|
||||
onVncUsePasswordChanged = pyqtSignal(bool)
|
||||
onVncStateChanged = pyqtSignal(VNCState)
|
||||
onDisplaySettingClosed = pyqtSignal()
|
||||
onError = pyqtSignal(str)
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super(Backend, self).__init__(parent)
|
||||
# Virtual screen properties
|
||||
self.xrandr: XRandR = XRandR()
|
||||
self._virtScreenCreated: bool = False
|
||||
# VNC server properties
|
||||
self._vncUsePassword: bool = False
|
||||
self._vncState: self.VNCState = self.VNCState.OFF
|
||||
# Primary screen and mouse posistion
|
||||
self.vncServer: AsyncSubprocess
|
||||
# Check config file
|
||||
# and initialize if needed
|
||||
need_init = False
|
||||
if not os.path.exists(CONFIG_PATH):
|
||||
shutil.copy(DEFAULT_CONFIG_PATH, CONFIG_PATH)
|
||||
need_init = True
|
||||
# Version check
|
||||
file_match = True
|
||||
with open(CONFIG_PATH, 'r') as f_config, open(DATA_PATH, 'r') as f_data:
|
||||
config = json.load(f_config)
|
||||
data = json.load(f_data)
|
||||
if config['version'] != data['version']:
|
||||
file_match = False
|
||||
# Override config with default when version doesn't match
|
||||
if not file_match:
|
||||
shutil.copy(DEFAULT_CONFIG_PATH, CONFIG_PATH)
|
||||
need_init = True
|
||||
# initialize config file
|
||||
if need_init:
|
||||
# 1. Available x11vnc options
|
||||
# Get available x11vnc options from x11vnc first
|
||||
p = SubprocessWrapper()
|
||||
arg = 'x11vnc -opts'
|
||||
ret = p.run(arg)
|
||||
options = tuple(m.group(1) for m in re.finditer("\s*(-\w+)\s+", ret))
|
||||
# Set/unset available x11vnc options flags in config
|
||||
with open(CONFIG_PATH, 'r') as f, open(DATA_PATH, 'r') as f_data:
|
||||
config = json.load(f)
|
||||
data = json.load(f_data)
|
||||
for key, value in config["x11vncOptions"].items():
|
||||
if key in options:
|
||||
value["available"] = True
|
||||
else:
|
||||
value["available"] = False
|
||||
# Default Display settings app for a Desktop Environment
|
||||
desktop_environ = os.environ['XDG_CURRENT_DESKTOP'].lower()
|
||||
for key, value in data['displaySettingApps'].items():
|
||||
for de in value['XDG_CURRENT_DESKTOP']:
|
||||
if de in desktop_environ:
|
||||
config["displaySettingApp"] = key
|
||||
# Save the new config
|
||||
with open(CONFIG_PATH, 'w') as f:
|
||||
f.write(json.dumps(config, indent=4, sort_keys=True))
|
||||
|
||||
# Qt properties
|
||||
@pyqtProperty(str, constant=True)
|
||||
def settings(self):
|
||||
with open(CONFIG_PATH, "r") as f:
|
||||
return f.read()
|
||||
|
||||
@settings.setter
|
||||
def settings(self, json_str):
|
||||
with open(CONFIG_PATH, "w") as f:
|
||||
f.write(json_str)
|
||||
|
||||
@pyqtProperty(bool, notify=onVirtScreenCreatedChanged)
|
||||
def virtScreenCreated(self):
|
||||
return self._virtScreenCreated
|
||||
|
||||
@virtScreenCreated.setter
|
||||
def virtScreenCreated(self, value):
|
||||
self._virtScreenCreated = value
|
||||
self.onVirtScreenCreatedChanged.emit(value)
|
||||
|
||||
@pyqtProperty(QQmlListProperty, constant=True)
|
||||
def screens(self):
|
||||
try:
|
||||
return QQmlListProperty(DisplayProperty, self, [DisplayProperty(x) for x in self.xrandr.screens])
|
||||
except RuntimeError as e:
|
||||
self.onError.emit(str(e))
|
||||
return QQmlListProperty(DisplayProperty, self, [])
|
||||
|
||||
@pyqtProperty(bool, notify=onVncUsePasswordChanged)
|
||||
def vncUsePassword(self):
|
||||
if os.path.isfile(X11VNC_PASSWORD_PATH):
|
||||
self._vncUsePassword = True
|
||||
else:
|
||||
if self._vncUsePassword:
|
||||
self.vncUsePassword = False
|
||||
return self._vncUsePassword
|
||||
|
||||
@vncUsePassword.setter
|
||||
def vncUsePassword(self, use):
|
||||
self._vncUsePassword = use
|
||||
self.onVncUsePasswordChanged.emit(use)
|
||||
|
||||
@pyqtProperty(VNCState, notify=onVncStateChanged)
|
||||
def vncState(self):
|
||||
return self._vncState
|
||||
|
||||
@vncState.setter
|
||||
def vncState(self, state):
|
||||
self._vncState = state
|
||||
self.onVncStateChanged.emit(self._vncState)
|
||||
|
||||
# Qt Slots
|
||||
@pyqtSlot(str, int, int, bool, bool)
|
||||
def createVirtScreen(self, device, width, height, portrait, hidpi, pos=''):
|
||||
self.xrandr.virt_name = device
|
||||
print("Creating a Virtual Screen...")
|
||||
try:
|
||||
self.xrandr.create_virtual_screen(width, height, portrait, hidpi, pos)
|
||||
except subprocess.CalledProcessError as e:
|
||||
self.onError.emit(str(e.cmd) + '\n' + e.stdout.decode('utf-8'))
|
||||
return
|
||||
except RuntimeError as e:
|
||||
self.onError.emit(str(e))
|
||||
return
|
||||
self.virtScreenCreated = True
|
||||
|
||||
@pyqtSlot()
|
||||
def deleteVirtScreen(self):
|
||||
print("Deleting the Virtual Screen...")
|
||||
if self.vncState is not self.VNCState.OFF:
|
||||
self.onError.emit("Turn off the VNC server first")
|
||||
self.virtScreenCreated = True
|
||||
return
|
||||
try:
|
||||
self.xrandr.delete_virtual_screen()
|
||||
except RuntimeError as e:
|
||||
self.onError.emit(str(e))
|
||||
return
|
||||
self.virtScreenCreated = False
|
||||
|
||||
@pyqtSlot(str)
|
||||
def createVNCPassword(self, password):
|
||||
if password:
|
||||
password += '\n' + password + '\n\n' # verify + confirm
|
||||
p = SubprocessWrapper()
|
||||
try:
|
||||
p.run(f"x11vnc -storepasswd {X11VNC_PASSWORD_PATH}", input=password, check=True)
|
||||
except subprocess.CalledProcessError as e:
|
||||
self.onError.emit(str(e.cmd) + '\n' + e.stdout.decode('utf-8'))
|
||||
return
|
||||
self.vncUsePassword = True
|
||||
else:
|
||||
self.onError.emit("Empty password")
|
||||
|
||||
@pyqtSlot()
|
||||
def deleteVNCPassword(self):
|
||||
if os.path.isfile(X11VNC_PASSWORD_PATH):
|
||||
os.remove(X11VNC_PASSWORD_PATH)
|
||||
self.vncUsePassword = False
|
||||
else:
|
||||
self.onError.emit("Failed deleting the password file")
|
||||
|
||||
@pyqtSlot(int)
|
||||
def startVNC(self, port):
|
||||
# Check if a virtual screen created
|
||||
if not self.virtScreenCreated:
|
||||
self.onError.emit("Virtual Screen not crated.")
|
||||
return
|
||||
if self.vncState is not self.VNCState.OFF:
|
||||
self.onError.emit("VNC Server is already running.")
|
||||
return
|
||||
# regex used in callbacks
|
||||
patter_connected = re.compile(r"^.*Got connection from client.*$", re.M)
|
||||
patter_disconnected = re.compile(r"^.*client_count: 0*$", re.M)
|
||||
|
||||
# define callbacks
|
||||
def _connected():
|
||||
print("VNC started.")
|
||||
self.vncState = self.VNCState.WAITING
|
||||
|
||||
def _received(data):
|
||||
data = data.decode("utf-8")
|
||||
if (self._vncState is not self.VNCState.CONNECTED) and patter_connected.search(data):
|
||||
print("VNC connected.")
|
||||
self.vncState = self.VNCState.CONNECTED
|
||||
if (self._vncState is self.VNCState.CONNECTED) and patter_disconnected.search(data):
|
||||
print("VNC disconnected.")
|
||||
self.vncState = self.VNCState.WAITING
|
||||
|
||||
def _ended(exitCode):
|
||||
if exitCode is not 0:
|
||||
self.vncState = self.VNCState.ERROR
|
||||
self.onError.emit('X11VNC: Error occurred.\n'
|
||||
'Double check if the port is already used.')
|
||||
self.vncState = self.VNCState.OFF # TODO: better handling error state
|
||||
else:
|
||||
self.vncState = self.VNCState.OFF
|
||||
print("VNC Exited.")
|
||||
atexit.unregister(self.stopVNC)
|
||||
# load settings
|
||||
with open(CONFIG_PATH, 'r') as f:
|
||||
config = json.load(f)
|
||||
options = ''
|
||||
if config['customX11vncArgs']['enabled']:
|
||||
options = config['customX11vncArgs']['value']
|
||||
else:
|
||||
for key, value in config['x11vncOptions'].items():
|
||||
if value['available'] and value['enabled']:
|
||||
options += key + ' '
|
||||
if value['arg'] is not None:
|
||||
options += str(value['arg']) + ' '
|
||||
# Sart x11vnc, turn settings object into VNC arguments format
|
||||
logfile = open(X11VNC_LOG_PATH, "wb")
|
||||
self.vncServer = AsyncSubprocess(_connected, _received, _received, _ended, logfile)
|
||||
try:
|
||||
virt = self.xrandr.get_virtual_screen()
|
||||
except RuntimeError as e:
|
||||
self.onError.emit(str(e))
|
||||
return
|
||||
clip = f"{virt.width}x{virt.height}+{virt.x_offset}+{virt.y_offset}"
|
||||
arg = f"x11vnc -rfbport {port} -clip {clip} {options}"
|
||||
if self.vncUsePassword:
|
||||
arg += f" -rfbauth {X11VNC_PASSWORD_PATH}"
|
||||
self.vncServer.run(arg)
|
||||
# auto stop on exit
|
||||
atexit.register(self.stopVNC, force=True)
|
||||
|
||||
@pyqtSlot(str)
|
||||
def openDisplaySetting(self, app: str = "arandr"):
|
||||
# define callbacks
|
||||
def _connected():
|
||||
print("External Display Setting opened.")
|
||||
|
||||
def _received(data):
|
||||
pass
|
||||
|
||||
def _ended(exitCode):
|
||||
print("External Display Setting closed.")
|
||||
self.onDisplaySettingClosed.emit()
|
||||
if exitCode is not 0:
|
||||
self.onError.emit(f'Error opening "{running_program}".')
|
||||
with open(DATA_PATH, 'r') as f:
|
||||
data = json.load(f)['displaySettingApps']
|
||||
if app not in data:
|
||||
self.onError.emit('Wrong display settings program')
|
||||
return
|
||||
program_list = [data[app]['args'], "arandr"]
|
||||
program = AsyncSubprocess(_connected, _received, _received, _ended, None)
|
||||
running_program = ''
|
||||
for arg in program_list:
|
||||
if not shutil.which(arg.split()[0]):
|
||||
continue
|
||||
running_program = arg
|
||||
program.run(arg)
|
||||
return
|
||||
self.onError.emit('Failed to find a display settings program.\n'
|
||||
'Please install ARandR package.\n'
|
||||
'(e.g. sudo apt-get install arandr)\n'
|
||||
'Please issue a feature request\n'
|
||||
'if you wish to add a display settings\n'
|
||||
'program for your Desktop Environment.')
|
||||
|
||||
@pyqtSlot()
|
||||
def stopVNC(self, force=False):
|
||||
if force:
|
||||
# Usually called from atexit().
|
||||
self.vncServer.close()
|
||||
time.sleep(3) # Make sure X11VNC shutdown before execute next atexit().
|
||||
if self._vncState in (self.VNCState.WAITING, self.VNCState.CONNECTED):
|
||||
self.vncServer.close()
|
||||
else:
|
||||
self.onError.emit("stopVNC called while it is not running")
|
||||
|
||||
@pyqtSlot()
|
||||
def clearCache(self):
|
||||
# engine.clearComponentCache()
|
||||
pass
|
||||
|
||||
@pyqtSlot()
|
||||
def quitProgram(self):
|
||||
self.blockSignals(True) # This will prevent invoking auto-restart or etc.
|
||||
QApplication.instance().quit()
|
||||
|
||||
|
||||
class Cursor(QObject):
|
||||
""" Global mouse cursor position """
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super(Cursor, self).__init__(parent)
|
||||
|
||||
@pyqtProperty(int)
|
||||
def x(self):
|
||||
cursor = QCursor().pos()
|
||||
return cursor.x()
|
||||
|
||||
@pyqtProperty(int)
|
||||
def y(self):
|
||||
cursor = QCursor().pos()
|
||||
return cursor.y()
|
||||
|
||||
|
||||
class Network(QObject):
|
||||
""" Backend class for network interfaces """
|
||||
onIPAddressesChanged = pyqtSignal()
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super(Network, self).__init__(parent)
|
||||
|
||||
@pyqtProperty('QStringList', notify=onIPAddressesChanged)
|
||||
def ipAddresses(self):
|
||||
for interface in interfaces():
|
||||
if interface == 'lo':
|
||||
continue
|
||||
addresses = ifaddresses(interface).get(AF_INET, None)
|
||||
if addresses is None:
|
||||
continue
|
||||
for link in addresses:
|
||||
if link is not None:
|
||||
yield link['addr']
|
||||
Loading…
Add table
Add a link
Reference in a new issue