10686238271

Committed 03 Sep 2024 03:42PM UTC coverage: 94.216% (-0.9%) from 95.106%

Build # 10686238271

Build Type

Pull #336

github

Committed by

emcek

Commit Message

Merge branch 'master' into client_in_qthread

Pull Request Pull Request #336: Start DCSpy client as QThread

Run Details

519 of 579 branches covered (89.64%)

Branch coverage included in aggregate %.

50 of 61 new or added lines in 4 files covered. (81.97%)

11 existing lines in 1 file now uncovered.

2071 of 2170 relevant lines covered (95.44%)

0.95 hits per line

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

58.11

/dcspy/starter.py

import socket
import struct
from collections import deque
from collections.abc import Iterator
from logging import getLogger
from threading import Event
from time import gmtime, time

from dcspy import get_config_yaml_item
from dcspy.dcsbios import ProtocolParser
from dcspy.logitech import LogitechDevice
from dcspy.models import MULTICAST_IP, RECV_ADDR, LogitechDeviceModel
from dcspy.utils import SignalHandler, check_bios_ver, get_version_string

LOG = getLogger(__name__)
LOOP_FLAG = True
__version__ = '3.5.1'


def _handle_connection(logi_device: LogitechDevice, parser: ProtocolParser, sock: socket.socket, ver_string: str,
                       event: Event, sig_handler: SignalHandler) -> None:
    """
    Handle main loop where all the magic is happened.

    :param logi_device: type of Logitech keyboard with LCD
    :param parser: DCS protocol parser
    :param sock: multicast UDP socket
    :param ver_string: current version to show
    :param event: stop event for main loop
    :param sig_handler: Qt signal handler for progress notification
    """
    start_time = time()
    LOG.info('Waiting for DCS connection...')
    support_banner = _supporters(text='Huge thanks to: Simon Leigh, Alexander Leschanz, Sireyn, Nick Thain, BrotherBloat and others! For support and help! ',
                                 width=26)
    while not event.is_set():
        try:
            dcs_bios_resp = sock.recv(2048)
            for int_byte in dcs_bios_resp:
                parser.process_byte(int_byte)
            sig_handler.emit(sig_name='count', value=(0, len(dcs_bios_resp)))
            start_time = time()
            _load_new_plane_if_detected(logi_device)
            logi_device.button_handle()
        except OSError as exp:
            _sock_err_handler(logi_device, start_time, ver_string, support_banner, exp)


def _load_new_plane_if_detected(logi_device: LogitechDevice) -> None:
    """
    Load instance when new plane detected.

    :param logi_device: type of Logitech keyboard with LCD
    """
    global LOOP_FLAG
    if logi_device.plane_detected:
        logi_device.unload_old_plane()
        logi_device.load_new_plane()
        LOOP_FLAG = True


def _supporters(text: str, width: int) -> Iterator[str]:
    """
    Scroll text with widow width.

    :param text: text to scroll
    :param width: width of window
    """
    queue = deque(text)
    while True:
        yield ''.join(queue)[:width]
        queue.rotate(-1)


def _sock_err_handler(logi_device: LogitechDevice, start_time: float, ver_string: str, support_iter: Iterator[str], exp: Exception) -> None:
    """
    Show basic data when DCS is disconnected.

    :param logi_device: type of Logitech keyboard with LCD
    :param start_time: time when connection to DCS was lost
    :param ver_string: current version to show
    :param support_iter: iterator for banner supporters
    :param exp: caught exception instance
    """
    global LOOP_FLAG
    if LOOP_FLAG:
        LOG.debug(f'Main loop socket error: {exp}')
        LOOP_FLAG = False
    wait_time = gmtime(time() - start_time)
    logi_device.display = ['Logitech LCD OK',
                           f'No data from DCS:   {wait_time.tm_min:02d}:{wait_time.tm_sec:02d}',
                           f'{next(support_iter)}',
                           ver_string]


def _prepare_socket() -> socket.socket:
    """
    Prepare multicast UDP socket for DCS-BIOS communication.

    :return: socket object
    """
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(RECV_ADDR)
    mreq = struct.pack('=4sl', socket.inet_aton(MULTICAST_IP), socket.INADDR_ANY)
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
    sock.settimeout(0.5)
    return sock


def dcspy_run(model: LogitechDeviceModel, event: Event, sig_handler: SignalHandler) -> None:
    """
    Real starting point of DCSpy.

    :param model: Logitech device model
    :param event: stop event for main loop
    :param sig_handler: Qt signal handler for progress notification
    """
    with _prepare_socket() as dcs_sock:
        parser = ProtocolParser()
        logi_dev = LogitechDevice(parser=parser, sock=dcs_sock, model=model, sig_handler=sig_handler)
        LOG.info(f'Loading: {str(logi_dev)}')
        LOG.debug(f'Loading: {repr(logi_dev)}')
        dcspy_ver = get_version_string(repo='emcek/dcspy', current_ver=__version__, check=bool(get_config_yaml_item('check_ver')))
        _handle_connection(logi_device=logi_dev, parser=parser, sock=dcs_sock, ver_string=dcspy_ver, event=event, sig_handler=sig_handler)
    LOG.info('DCSpy stopped.')
    logi_dev.display = ['DCSpy stopped', '', f'DCSpy: {dcspy_ver}', f'DCS-BIOS: {check_bios_ver(bios_path=str(get_config_yaml_item("dcsbios"))).ver}']

1	import socket	1✔
2	import struct	1✔
3	from collections import deque	1✔
4	from collections.abc import Iterator	1✔
5	from logging import getLogger	1✔
6	from threading import Event	1✔
7	from time import gmtime, time	1✔
8
9	from dcspy import get_config_yaml_item	1✔
10	from dcspy.dcsbios import ProtocolParser	1✔
11	from dcspy.logitech import LogitechDevice	1✔
12	from dcspy.models import MULTICAST_IP, RECV_ADDR, LogitechDeviceModel	1✔
13	from dcspy.utils import SignalHandler, check_bios_ver, get_version_string	1✔
14
15	LOG = getLogger(__name__)	1✔
16	LOOP_FLAG = True	1✔
17	__version__ = '3.5.1'	1✔
18
19
20	def _handle_connection(logi_device: LogitechDevice, parser: ProtocolParser, sock: socket.socket, ver_string: str,	1✔
21	event: Event, sig_handler: SignalHandler) -> None:
22	"""
23	Handle main loop where all the magic is happened.
24
25	:param logi_device: type of Logitech keyboard with LCD
26	:param parser: DCS protocol parser
27	:param sock: multicast UDP socket
28	:param ver_string: current version to show
29	:param event: stop event for main loop
30	:param sig_handler: Qt signal handler for progress notification
31	"""
UNCOV 32	start_time = time()	×
UNCOV 33	LOG.info('Waiting for DCS connection...')	×
UNCOV 34	support_banner = _supporters(text='Huge thanks to: Simon Leigh, Alexander Leschanz, Sireyn, Nick Thain, BrotherBloat and others! For support and help! ',	×
35	width=26)
UNCOV 36	while not event.is_set():	×
37	try:	×
38	dcs_bios_resp = sock.recv(2048)	×
39	for int_byte in dcs_bios_resp:	×
40	parser.process_byte(int_byte)	×
NEW 41	sig_handler.emit(sig_name='count', value=(0, len(dcs_bios_resp)))	×
42	start_time = time()	×
43	_load_new_plane_if_detected(logi_device)	×
44	logi_device.button_handle()	×
45	except OSError as exp:	×
46	_sock_err_handler(logi_device, start_time, ver_string, support_banner, exp)	×
47
48
49	def _load_new_plane_if_detected(logi_device: LogitechDevice) -> None:	1✔
50	"""
51	Load instance when new plane detected.
52
53	:param logi_device: type of Logitech keyboard with LCD
54	"""
55	global LOOP_FLAG
56	if logi_device.plane_detected:	1!
57	logi_device.unload_old_plane()	1✔
58	logi_device.load_new_plane()	1✔
59	LOOP_FLAG = True	1✔
60
61
62	def _supporters(text: str, width: int) -> Iterator[str]:	1✔
63	"""
64	Scroll text with widow width.
65
66	:param text: text to scroll
67	:param width: width of window
68	"""
69	queue = deque(text)	1✔
70	while True:	1✔
71	yield ''.join(queue)[:width]	1✔
72	queue.rotate(-1)	1✔
73
74
75	def _sock_err_handler(logi_device: LogitechDevice, start_time: float, ver_string: str, support_iter: Iterator[str], exp: Exception) -> None:	1✔
76	"""
77	Show basic data when DCS is disconnected.
78
79	:param logi_device: type of Logitech keyboard with LCD
80	:param start_time: time when connection to DCS was lost
81	:param ver_string: current version to show
82	:param support_iter: iterator for banner supporters
83	:param exp: caught exception instance
84	"""
85	global LOOP_FLAG
86	if LOOP_FLAG:	1!
87	LOG.debug(f'Main loop socket error: {exp}')	1✔
88	LOOP_FLAG = False	1✔
89	wait_time = gmtime(time() - start_time)	1✔
90	logi_device.display = ['Logitech LCD OK',	1✔
91	f'No data from DCS: {wait_time.tm_min:02d}:{wait_time.tm_sec:02d}',
92	f'{next(support_iter)}',
93	ver_string]
94
95
96	def _prepare_socket() -> socket.socket:	1✔
97	"""
98	Prepare multicast UDP socket for DCS-BIOS communication.
99
100	:return: socket object
101	"""
102	sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)	1✔
103	sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)	1✔
104	sock.bind(RECV_ADDR)	1✔
105	mreq = struct.pack('=4sl', socket.inet_aton(MULTICAST_IP), socket.INADDR_ANY)	1✔
106	sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)	1✔
107	sock.settimeout(0.5)	1✔
108	return sock	1✔
109
110
111	def dcspy_run(model: LogitechDeviceModel, event: Event, sig_handler: SignalHandler) -> None:	1✔
112	"""
113	Real starting point of DCSpy.
114
115	:param model: Logitech device model
116	:param event: stop event for main loop
117	:param sig_handler: Qt signal handler for progress notification
118	"""
UNCOV 119	with _prepare_socket() as dcs_sock:	×
UNCOV 120	parser = ProtocolParser()	×
NEW 121	logi_dev = LogitechDevice(parser=parser, sock=dcs_sock, model=model, sig_handler=sig_handler)	×
UNCOV 122	LOG.info(f'Loading: {str(logi_dev)}')	×
UNCOV 123	LOG.debug(f'Loading: {repr(logi_dev)}')	×
UNCOV 124	dcspy_ver = get_version_string(repo='emcek/dcspy', current_ver=__version__, check=bool(get_config_yaml_item('check_ver')))	×
NEW 125	_handle_connection(logi_device=logi_dev, parser=parser, sock=dcs_sock, ver_string=dcspy_ver, event=event, sig_handler=sig_handler)	×
UNCOV 126	LOG.info('DCSpy stopped.')	×
UNCOV 127	logi_dev.display = ['DCSpy stopped', '', f'DCSpy: {dcspy_ver}', f'DCS-BIOS: {check_bios_ver(bios_path=str(get_config_yaml_item("dcsbios"))).ver}']	×

emcek / dcspy / 10686238271

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