18953976799

Committed 30 Oct 2025 08:21PM UTC coverage: 80.119% (-1.4%) from 81.509%

Build # 18953976799

Build Type

Pull #798

github

Committed by

web-flow

Commit Message

Merge a4477b844 into 50be529a6

Pull Request Pull Request #798: Refactoring

Coverage Stats

858 of 1175 new or added lines in 51 files covered. (73.02%)

23 existing lines in 5 files now uncovered.

2974 of 3712 relevant lines covered (80.12%)

4.57 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.49

/normcap/gui/socket_server.py

import atexit
import logging

from PySide6 import QtCore, QtNetwork

from normcap import __version__

logger = logging.getLogger(__name__)


class Communicate(QtCore.QObject):
    """Application's communication bus."""

    on_capture_message = QtCore.Signal()
    on_other_instance_running = QtCore.Signal()


class SocketServer(QtCore.QObject):
    _name = f"v{__version__}-normcap"
    _out: QtNetwork.QLocalSocket | None = None
    _in: QtNetwork.QLocalSocket | None = None
    _server: QtNetwork.QLocalServer | None = None
    is_first_instance: bool

    def __init__(self) -> None:
        super().__init__()
        self.com = Communicate()

        self._out = self._connect_to_other_instance()
        if self._out:
            # Send message to other instance
            logger.debug("Another instance is already running. Sending capture signal.")
            self._out.write(b"capture")
            self._out.waitForBytesWritten(1000)
            self.is_first_instance = False
        else:
            # Start server
            self._create_socket_server()
            self.is_first_instance = True
            atexit.register(self.close)

    def _connect_to_other_instance(self) -> QtNetwork.QLocalSocket | None:
        """Test if connection to another NormCap instance socket can be established."""
        self._out = QtNetwork.QLocalSocket(self)
        self._out.connectToServer(self._name)
        if not self._out.waitForConnected():
            # Couldn't connect, cleaning up out
            self._out.close()
            self._out = None

        return self._out

    def _create_socket_server(self) -> None:
        """Open socket server to listen for other NormCap instances."""
        QtNetwork.QLocalServer().removeServer(self._name)
        self._server = QtNetwork.QLocalServer(self)
        self._server.newConnection.connect(self._on_socket_connect)
        self._server.listen(self._name)
        logger.debug("Listen on local socket %s.", self._server.serverName())

    @QtCore.Slot()
    def _on_socket_connect(self) -> None:
        """Open incoming socket to listen for messages from other NormCap instances."""
        if not self._server:
            return
        self._in = self._server.nextPendingConnection()
        if self._in:
            logger.debug("Connect to incoming socket.")
            self._in.readyRead.connect(self._on_socket_ready_read)

    @QtCore.Slot()
    def _on_socket_ready_read(self) -> None:
        """Process messages received from other NormCap instances."""
        if not self._in:
            return

        message = self._in.readAll().data().decode("utf-8", errors="ignore")
        logger.info("Received socket message '%s'", message)

        if message == "capture":
            self.com.on_capture_message.emit()

    def close(self) -> None:
        if self._out:
            self._out.close()
            self._out = None

        if self._server:
            self._server.close()
            self._server.removeServer(self._name)
            self._server = None

1	import atexit	6✔
2	import logging	6✔
3
4	from PySide6 import QtCore, QtNetwork	6✔
5
6	from normcap import __version__	6✔
7
8	logger = logging.getLogger(__name__)	6✔
9
10
11	class Communicate(QtCore.QObject):	6✔
12	"""Application's communication bus."""
13
14	on_capture_message = QtCore.Signal()	6✔
15	on_other_instance_running = QtCore.Signal()	6✔
16
17
18	class SocketServer(QtCore.QObject):	6✔
19	_name = f"v{__version__}-normcap"	6✔
20	_out: QtNetwork.QLocalSocket \| None = None	6✔
21	_in: QtNetwork.QLocalSocket \| None = None	6✔
22	_server: QtNetwork.QLocalServer \| None = None	6✔
23	is_first_instance: bool	6✔
24
25	def __init__(self) -> None:	6✔
26	super().__init__()	6✔
27	self.com = Communicate()	6✔
28
29	self._out = self._connect_to_other_instance()	6✔
30	if self._out:	6✔
31	# Send message to other instance
NEW 32	logger.debug("Another instance is already running. Sending capture signal.")	×
NEW 33	self._out.write(b"capture")	×
NEW 34	self._out.waitForBytesWritten(1000)	×
NEW 35	self.is_first_instance = False	×
36	else:
37	# Start server
38	self._create_socket_server()	6✔
39	self.is_first_instance = True	6✔
40	atexit.register(self.close)	6✔
41
42	def _connect_to_other_instance(self) -> QtNetwork.QLocalSocket \| None:	6✔
43	"""Test if connection to another NormCap instance socket can be established."""
44	self._out = QtNetwork.QLocalSocket(self)	6✔
45	self._out.connectToServer(self._name)	6✔
46	if not self._out.waitForConnected():	6✔
47	# Couldn't connect, cleaning up out
48	self._out.close()	6✔
49	self._out = None	6✔
50
51	return self._out	6✔
52
53	def _create_socket_server(self) -> None:	6✔
54	"""Open socket server to listen for other NormCap instances."""
55	QtNetwork.QLocalServer().removeServer(self._name)	6✔
56	self._server = QtNetwork.QLocalServer(self)	6✔
57	self._server.newConnection.connect(self._on_socket_connect)	6✔
58	self._server.listen(self._name)	6✔
59	logger.debug("Listen on local socket %s.", self._server.serverName())	6✔
60
61	@QtCore.Slot()	6✔
62	def _on_socket_connect(self) -> None:	6✔
63	"""Open incoming socket to listen for messages from other NormCap instances."""
NEW 64	if not self._server:	×
NEW 65	return	×
NEW 66	self._in = self._server.nextPendingConnection()	×
NEW 67	if self._in:	×
NEW 68	logger.debug("Connect to incoming socket.")	×
NEW 69	self._in.readyRead.connect(self._on_socket_ready_read)	×
70
71	@QtCore.Slot()	6✔
72	def _on_socket_ready_read(self) -> None:	6✔
73	"""Process messages received from other NormCap instances."""
NEW 74	if not self._in:	×
NEW 75	return	×
76
NEW 77	message = self._in.readAll().data().decode("utf-8", errors="ignore")	×
NEW 78	logger.info("Received socket message '%s'", message)	×
79
NEW 80	if message == "capture":	×
NEW 81	self.com.on_capture_message.emit()	×
82
83	def close(self) -> None:	6✔
NEW 84	if self._out:	×
NEW 85	self._out.close()	×
NEW 86	self._out = None	×
87
NEW 88	if self._server:	×
NEW 89	self._server.close()	×
NEW 90	self._server.removeServer(self._name)	×
NEW 91	self._server = None	×

dynobo / normcap / 18953976799

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous