11670323424

Committed 04 Nov 2024 06:24PM UTC coverage: 93.525% (-0.9%) from 94.423%

Build # 11670323424

Build Type

Pull #336

github

Committed by

emcek

Commit Message

use new type annotations in logitech module

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

Run Details

611 of 695 branches covered (87.91%)

Branch coverage included in aggregate %.

49 of 60 new or added lines in 4 files covered. (81.67%)

11 existing lines in 1 file now uncovered.

2090 of 2193 relevant lines covered (95.3%)

0.95 hits per line

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

58.67

/src/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
SUPPORTERS = ['Jon Wardell', 'Simon Leigh', 'Alexander Leschanz', 'Sireyn', 'Nick Thain', 'BrotherBloat']
__version__ = '3.6.0'


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

emcek / dcspy / 11670323424

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