PyDMXControl.controllers.utils.debug.Debugger.run_fixture() - Code Metrics - Inspection of "refactor profiles.default.Fixture (move helpers to..." - MattIPv4/PyDMXControl - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — refactoring (fbf932)

by Matt

created 2018-12-15 18:08 UTC

Debugger.run_fixture() C

↳ Parent: PyDMXControl.controllers.utils.debug

Complexity

Conditions

Size

Total Lines	48
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	27
dl	0
loc	48
rs	5.9999
c	0
b	0
f	0
cc	10
nop	1

How to fix Complexity

"""
 *  PyDMXControl: A Python 3 module to control DMX via Python. Featuring fixture profiles and working with uDMX.
 *  <https://github.com/MattIPv4/PyDMXControl/>
 *  Copyright (C) 2018 Matt Cowley (MattIPv4) ([email protected])
"""

from collections import namedtuple
from inspect import signature, Parameter
from re import compile as re_compile
from typing import List, Tuple, Union, Dict, Callable

from ... import Colors
from ...profiles.defaults import Fixture, Vdim


class Debugger:

    def __init__(self, controller: 'Controller', callbacks: Dict[str, Callable] = None):
        self.cont = controller
        self.cbs = {} if callbacks is None else callbacks

    def __default_callbacks(self):
        # Some default callbacks
        if 'all_on' not in self.cbs:
            self.cbs['all_on'] = self.cont.all_on
        if 'on' not in self.cbs:
            self.cbs['on'] = self.cont.all_on

        if 'all_off' not in self.cbs:
            self.cbs['all_off'] = self.cont.all_off
        if 'off' not in self.cbs:
            self.cbs['off'] = self.cont.all_off

        if 'all_locate' not in self.cbs:
            self.cbs['all_locate'] = self.cont.all_locate
        if 'locate' not in self.cbs:
            self.cbs['locate'] = self.cont.all_locate

    def __check_callbacks(self):
        for key in self.cbs.keys():
            if not self.cbs[key] or not callable(self.cbs[key]):
                del self.cbs[key]

    @staticmethod
    def __callbacks_parameters(parameters: List[Parameter]) -> Tuple[list, dict]:
        # Given params
        ordered_params = []
        keyword_params = {}

        # Go through all params
        for param in parameters:
            # Basic param information
            has_default = (param.default != Parameter.empty)
            has_type = (param.annotation != Parameter.empty)
            param_type = str(param.annotation) if has_type else "Unknown"
            param_default = (", leave blank for default " + str(param.default)) if has_default else ""

            # Validate the parameter input
            given_param = False

            def valid(this):
                # Not started
                if this is None:
                    return False

                # Default?
                if this.strip() == "":
                    if has_default:
                        return param.default
                    return False

                # Normal
                try:
                    return param.annotation(this)
                except Exception:
                    return this

            # Get input
            while given_param is False:
                given_param = valid(input(
                    "[Callbacks Debug] Parameter '" + param.name + "' (expects " + param_type + param_default + "): "))

            # Add to return
            if param.kind == Parameter.POSITIONAL_ONLY:
                ordered_params.append(given_param)
            else:
                keyword_params[param.name] = given_param

        return ordered_params, keyword_params

    def run_callbacks(self):
        # Defaults
        self.__default_callbacks()
        self.__check_callbacks()

        # Give callbacks
        print("\n[Callbacks Debug] Available callbacks:",
              ", ".join(["'" + f + "'" for f in self.cbs.keys()]))
        while True:
            # Selection
            callback = input("\n[Callbacks Debug] Callback Name (or 'exit'): ").strip()

            # Allow exit
            if callback == 'exit':
                break

            # Check it exists
            if callback not in self.cbs:
                continue

            # Run the callback
            cb = self.cbs[callback]
            try:
                # Get all parameters
                params = signature(cb).parameters.values()
                ordered, keyword = self.__callbacks_parameters(params)

                # Run
                res = cb(*ordered, **keyword)
            except Exception as e:
                print(e)
                print("[Callbacks Debug] '" + callback + "' failed.")
            else:
                print("[Callbacks Debug] Callback '" + callback + "' succeed and returned:", res)

    @staticmethod
    def __fixture_channels(fixture: Fixture) -> List[str]:
        names = ["'" + f['name'] + "'" for f in fixture.channels.values()]
        if issubclass(type(fixture), Vdim):
            names.append("'dimmer'")
        return names

    @staticmethod
    def __fixture_channel_value(fixture: Fixture, channel: Union[str, int]) -> int:
        if issubclass(type(fixture), Vdim):
            return fixture.get_channel_value(channel, False)[0]
        return fixture.get_channel_value(channel)[0]

    @staticmethod
    def run_fixture_color(fixture: Fixture):
        # Select
        select = input("\n[Fixture Debug] Color Name: ").strip().lower()

        # Try finding enum
        color = [c for c in Colors if c.name.lower() == select]

        # If not enum, try regex, else fetch enum
        if not color:
            pattern = re_compile(
                r"^\s*(\d{1,3})\s*[, ]\s*(\d{1,3})\s*[, ]\s*(\d{1,3})\s*(?:[, ]\s*(\d{1,3})\s*)*$")
            match = pattern.match(select)
            if not match:
                return
            color = {"name": "User Input", "value": [int(f) for f in match.groups() if f]}
            color = namedtuple("Color", color.keys())(*color.values())
        else:
            color = color[0]

        # Apply
        fixture.color(color.value)
        print("[Fixture Debug] Color set to " + color.name + " (" + Colors.to_print(color.value) + ")")

    def run_fixture_channel(self, fixture: Fixture):
        # Select
        channel = input("\n[Channel Debug] Channel Number/Name: ").strip()

        # Find
        channel = fixture.get_channel_id(channel)

        # Abort if not found
        if channel == -1:
            return

        # Value
        value = str(self.__fixture_channel_value(fixture, channel))
        value = input("[Channel Debug] Channel Value (Currently " + value + ", leave blank to abort): ").strip()

        # Abort if bad value
        if value == "":
            return
        if not value.isdigit():
            return

        # Apply
        value = int(value)
        fixture.set_channel(channel, value)
        print("[Channel Debug] Channel '" + str(channel) + "' set to " + str(
            self.__fixture_channel_value(fixture, channel)))

    def run_fixture(self):
        fixture = input("\n[Fixture Debug] Fixture ID/Name: ").strip()

        # Find the fixture
        if not fixture.isdigit():
            fixture = self.cont.get_fixtures_by_name(fixture)
            if fixture:
                fixture = fixture[0]
        else:
            fixture = self.cont.get_fixture(int(fixture))
        if not fixture:
            return

        # Fixture debug control
        print("\n[Fixture Debug] Fixture selected:", fixture)
        while True:

            # Selection
            select = input("\n[Fixture Debug] '1': Channel Select by Number/Name"
                           "\n            '2': Channel List"
                           "\n            '3': Color (if fixture supports)"
                           "\n            '4': Color List"
                           "\n            '5': Exit"
                           "\nSelection: ").strip()

            # Channel number/name
            if select == '1':
                self.run_fixture_channel(fixture)
                continue

            # Channel list
            if select == '2':
                print("\n[Fixture Debug] Channel List:", ", ".join(self.__fixture_channels(fixture)))
                continue

            # Color select
            if select == '3':
                self.run_fixture_color(fixture)
                continue

            # Color list
            if select == '4':
                print("\n[Fixture Debug] Color List:", ", ".join([color.name for color in Colors]))
                continue

            # Exit
            if select == '5':
                break

    def run(self):
        # DMX debug control
        print("[DMX Debug] Currently operating in channels: {}".format("1->{}.".format(self.cont.next_channel - 1) if
                                                                       self.cont.next_channel > 1 else "None"))
        while True:
            # Selection
            select = input("\n[DMX Debug] '1': Fixture Select by ID/Name"
                           "\n            '2': Fixture List"
                           "\n            '3': Callbacks"
                           "\n            '4': Exit"
                           "\nSelection: ").strip()

            # Fixture id/name
            if select == '1':
                self.run_fixture()
                continue

            # Fixture list
            if select == '2':
                # Compile list
                fixtures = []
                for f in self.cont.get_all_fixtures():
                    fixtures.extend(["\n", f])
                # Output
                print("\n[DMX Debug] Fixture List:", *fixtures)
                continue

            # Callbacks
            if select == '3':
                self.run_callbacks()
                continue

            # Exit
            if select == '4':
                break


1			"""
2			* PyDMXControl: A Python 3 module to control DMX via Python. Featuring fixture profiles and working with uDMX.
3			* <https://github.com/MattIPv4/PyDMXControl/>
4			* Copyright (C) 2018 Matt Cowley (MattIPv4) ([email protected])
5			"""
6
7			from collections import namedtuple
8			from inspect import signature, Parameter
9			from re import compile as re_compile
10			from typing import List, Tuple, Union, Dict, Callable
11
12			from ... import Colors
13			from ...profiles.defaults import Fixture, Vdim
14
15
16			class Debugger:
17
18			def __init__(self, controller: 'Controller', callbacks: Dict[str, Callable] = None):
19			self.cont = controller
20			self.cbs = {} if callbacks is None else callbacks
21
22			def __default_callbacks(self):
23			# Some default callbacks
24			if 'all_on' not in self.cbs:
25			self.cbs['all_on'] = self.cont.all_on
26			if 'on' not in self.cbs:
27			self.cbs['on'] = self.cont.all_on
28
29			if 'all_off' not in self.cbs:
30			self.cbs['all_off'] = self.cont.all_off
31			if 'off' not in self.cbs:
32			self.cbs['off'] = self.cont.all_off
33
34			if 'all_locate' not in self.cbs:
35			self.cbs['all_locate'] = self.cont.all_locate
36			if 'locate' not in self.cbs:
37			self.cbs['locate'] = self.cont.all_locate
38
39			def __check_callbacks(self):
40			for key in self.cbs.keys():
41			if not self.cbs[key] or not callable(self.cbs[key]):
42			del self.cbs[key]
43
44			@staticmethod
45			def __callbacks_parameters(parameters: List[Parameter]) -> Tuple[list, dict]:
46			# Given params
47			ordered_params = []
48			keyword_params = {}
49
50			# Go through all params
51			for param in parameters:
52			# Basic param information
53			has_default = (param.default != Parameter.empty)
54			has_type = (param.annotation != Parameter.empty)
55			param_type = str(param.annotation) if has_type else "Unknown"
56			param_default = (", leave blank for default " + str(param.default)) if has_default else ""
57
58			# Validate the parameter input
59			given_param = False
60
61			def valid(this):
62			# Not started
63			if this is None:
64			return False
65
66			# Default?
67			if this.strip() == "":
68			if has_default:
69			return param.default
70			return False
71
72			# Normal
73			try:
74			return param.annotation(this)
75			except Exception:
76			return this
77
78			# Get input
79			while given_param is False:
80			given_param = valid(input(
81			"[Callbacks Debug] Parameter '" + param.name + "' (expects " + param_type + param_default + "): "))
82
83			# Add to return
84			if param.kind == Parameter.POSITIONAL_ONLY:
85			ordered_params.append(given_param)
86			else:
87			keyword_params[param.name] = given_param
88
89			return ordered_params, keyword_params
90
91			def run_callbacks(self):
92			# Defaults
93			self.__default_callbacks()
94			self.__check_callbacks()
95
96			# Give callbacks
97			print("\n[Callbacks Debug] Available callbacks:",
98			", ".join(["'" + f + "'" for f in self.cbs.keys()]))
99			while True:
100			# Selection
101			callback = input("\n[Callbacks Debug] Callback Name (or 'exit'): ").strip()
102
103			# Allow exit
104			if callback == 'exit':
105			break
106
107			# Check it exists
108			if callback not in self.cbs:
109			continue
110
111			# Run the callback
112			cb = self.cbs[callback]
113			try:
114			# Get all parameters
115			params = signature(cb).parameters.values()
116			ordered, keyword = self.__callbacks_parameters(params)
117
118			# Run
119			res = cb(ordered, *keyword)
120			except Exception as e:
121			print(e)
122			print("[Callbacks Debug] '" + callback + "' failed.")
123			else:
124			print("[Callbacks Debug] Callback '" + callback + "' succeed and returned:", res)
125
126			@staticmethod
127			def __fixture_channels(fixture: Fixture) -> List[str]:
128			names = ["'" + f['name'] + "'" for f in fixture.channels.values()]
129			if issubclass(type(fixture), Vdim):
130			names.append("'dimmer'")
131			return names
132
133			@staticmethod
134			def __fixture_channel_value(fixture: Fixture, channel: Union[str, int]) -> int:
135			if issubclass(type(fixture), Vdim):
136			return fixture.get_channel_value(channel, False)[0]
137			return fixture.get_channel_value(channel)[0]
138
139			@staticmethod
140			def run_fixture_color(fixture: Fixture):
141			# Select
142			select = input("\n[Fixture Debug] Color Name: ").strip().lower()
143
144			# Try finding enum
145			color = [c for c in Colors if c.name.lower() == select]
146
147			# If not enum, try regex, else fetch enum
148			if not color:
149			pattern = re_compile(
150			r"^\s(\d{1,3})\s[, ]\s(\d{1,3})\s[, ]\s(\d{1,3})\s(?:[, ]\s(\d{1,3})\s)*$")
151			match = pattern.match(select)
152			if not match:
153			return
154			color = {"name": "User Input", "value": [int(f) for f in match.groups() if f]}
155			color = namedtuple("Color", color.keys())(*color.values())
156			else:
157			color = color[0]
158
159			# Apply
160			fixture.color(color.value)
161			print("[Fixture Debug] Color set to " + color.name + " (" + Colors.to_print(color.value) + ")")
162
163			def run_fixture_channel(self, fixture: Fixture):
164			# Select
165			channel = input("\n[Channel Debug] Channel Number/Name: ").strip()
166
167			# Find
168			channel = fixture.get_channel_id(channel)
169
170			# Abort if not found
171			if channel == -1:
172			return
173
174			# Value
175			value = str(self.__fixture_channel_value(fixture, channel))
176			value = input("[Channel Debug] Channel Value (Currently " + value + ", leave blank to abort): ").strip()
177
178			# Abort if bad value
179			if value == "":
180			return
181			if not value.isdigit():
182			return
183
184			# Apply
185			value = int(value)
186			fixture.set_channel(channel, value)
187			print("[Channel Debug] Channel '" + str(channel) + "' set to " + str(
188			self.__fixture_channel_value(fixture, channel)))
189
190			def run_fixture(self):
191			fixture = input("\n[Fixture Debug] Fixture ID/Name: ").strip()
192
193			# Find the fixture
194			if not fixture.isdigit():
195			fixture = self.cont.get_fixtures_by_name(fixture)
196			if fixture:
197			fixture = fixture[0]
198			else:
199			fixture = self.cont.get_fixture(int(fixture))
200			if not fixture:
201			return
202
203			# Fixture debug control
204			print("\n[Fixture Debug] Fixture selected:", fixture)
205			while True:
206
207			# Selection
208			select = input("\n[Fixture Debug] '1': Channel Select by Number/Name"
209			"\n '2': Channel List"
210			"\n '3': Color (if fixture supports)"
211			"\n '4': Color List"
212			"\n '5': Exit"
213			"\nSelection: ").strip()
214
215			# Channel number/name
216			if select == '1':
217			self.run_fixture_channel(fixture)
218			continue
219
220			# Channel list
221			if select == '2':
222			print("\n[Fixture Debug] Channel List:", ", ".join(self.__fixture_channels(fixture)))
223			continue
224
225			# Color select
226			if select == '3':
227			self.run_fixture_color(fixture)
228			continue
229
230			# Color list
231			if select == '4':
232			print("\n[Fixture Debug] Color List:", ", ".join([color.name for color in Colors]))
233			continue
234
235			# Exit
236			if select == '5':
237			break
238
239			def run(self):
240			# DMX debug control
241			print("[DMX Debug] Currently operating in channels: {}".format("1->{}.".format(self.cont.next_channel - 1) if
242			self.cont.next_channel > 1 else "None"))
243			while True:
244			# Selection
245			select = input("\n[DMX Debug] '1': Fixture Select by ID/Name"
246			"\n '2': Fixture List"
247			"\n '3': Callbacks"
248			"\n '4': Exit"
249			"\nSelection: ").strip()
250
251			# Fixture id/name
252			if select == '1':
253			self.run_fixture()
254			continue
255
256			# Fixture list
257			if select == '2':
258			# Compile list
259			fixtures = []
260			for f in self.cont.get_all_fixtures():
261			fixtures.extend(["\n", f])
262			# Output
263			print("\n[DMX Debug] Fixture List:", *fixtures)
264			continue
265
266			# Callbacks
267			if select == '3':
268			self.run_callbacks()
269			continue
270
271			# Exit
272			if select == '4':
273			break
274

MattIPv4 / PyDMXControl

Branch — refactoring (fbf932)

Debugger.run_fixture() C

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like