bricknil.bricknil.main() - Code Metrics - Inspection of "Code simplification" - virantha/bricknil - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#11)

unknown

created 2020-01-14 13:24 UTC

bricknil.bricknil.main() C

↳ Parent: bricknil.bricknil

Complexity

Conditions

Size

Total Lines	64
Code Lines	33

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	10
eloc	33
nop	1
dl	0
loc	64
rs	5.9999
c	0
b	0
f	0

How to fix Long Method Complexity

# Copyright 2019 Virantha N. Ekanayake
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Utility functions to attach sensors/motors and start the whole event loop

    #. The decorator :class:`attach` to specify peripherals that
       connect to a hub (which enables sensing and motor control functions),
    #. The function :func:`start` that starts running the BLE communication queue, and all the hubs, in the event-loop system

"""

import logging
import pprint
from asyncio import run, sleep, Queue, get_event_loop
from asyncio import create_task as spawn
from functools import partial, wraps
import uuid

# Local imports
from .process import Process
from .ble_queue import BLEventQ
from .hub import PoweredUpHub, BoostHub, Hub
from .sockets import bricknil_socket_server

import threading

# Actual decorator that sets up the peripheral classes
# noinspection PyPep8Naming
class attach:
    """ Class-decorator to attach peripherals onto a Hub

        Injects sub-classes of `Peripheral` as instance variables on a Hub
        such as the PoweredUp Hub, akin to "attaching" a physical sensor or
        motor onto the Hub.

        Before you attach a peripheral with sensing capabilities,
        you need to ensure your `Peripheral` sub-class has the matching
        call-back method 'peripheralname_change'.

        Examples::

            @attach(PeripheralType,
                    name="instance name",
                    port='port',
                    capabilities=[])

        Warnings:
            - No support for checking to make sure user put in correct parameters
            - Identifies capabilities that need a callback update handler based purely on
              checking if the capability name starts with the string "sense*"

    """
    def __init__(self, peripheral_type, **kwargs):
        # TODO: check here to make sure parameters were entered
        if logging.getLogger().getEffectiveLevel() == logging.DEBUG:
            print(f'decorating with {peripheral_type}')
        self.peripheral_type = peripheral_type
        self.kwargs = kwargs

    def __call__ (self, cls):
        """
            Since the actual Hub sub-class being decorated can have __init__ params,
            we need to have a wrapper function inside here to capture the arguments
            going into that __init__ call.

            Inside that wrapper, we do the following:

            # Instance the peripheral that was decorated with the saved **kwargs
            # Check that any `sense_*` capabiilities in the peripheral have an
              appropriate handler method in the hub class being decorated.
            # Instance the Hub
            # Set the peripheral instance as an instance variable on the hub via the
              `Hub.attach_sensor` method

        """
        # Define a wrapper function to capture the actual instantiation and __init__ params
        @wraps(cls)
        def wrapper_f(*args, **kwargs):
            #print(f'type of cls is {type(cls)}')
            peripheral = self.peripheral_type(**self.kwargs)

            # Ugly, but scan through and check if any of the capabilities are sense_*
            if any([cap.name.startswith('sense') for cap in peripheral.capabilities]):
                handler_name = f'{peripheral.name}_change'
                assert hasattr(cls, handler_name), f'{cls.__name__} needs a handler {handler_name}'
            # Create the hub process and attach this peripheral
            o = cls(*args, **kwargs)
            o.message_debug(f"Decorating class {cls.__name__} with {self.peripheral_type.__name__}")
            o.attach_sensor(peripheral)
            return o
        return wrapper_f

async def main(system):
    """
    Entry-point coroutine that handles everything. This is to be run run
    in bricknil's main loop.

    You normally don't need to use this directly, instead use start()
    """

    # Instantiate the Bluetooth LE handler/queue
    ble_q = BLEventQ()
    # The web client out_going queue
    web_out_queue = Queue()
    # Instantiate socket listener
    # task_socket = await spawn(socket_server, web_out_queue, ('',25000))
    # task_tcp = await spawn(bricknil_socket_server, web_out_queue, ('',25000))
    # await task_tcp.join()

    # Call the user's system routine to instantiate the processes
    await system()

    hub_tasks = []
    hub_peripheral_listen_tasks = [] # Need to cancel these at the end

    # Connect all the hubs first before enabling any of them
    for hub in Hub.hubs:
        hub.web_queue_out = web_out_queue
        task_connect = spawn(ble_q.connect(hub))
        await task_connect

    for hub in Hub.hubs:
        # Start the peripheral listening loop in each hub
        task_listen = spawn(hub.peripheral_message_loop())
        hub_peripheral_listen_tasks.append(task_listen)

        # Need to wait here until all the ports are set
        # Use a faster timeout the first time (for speeding up testing)
        first_delay = True
        for name, peripheral in hub.peripherals.items():
            while peripheral.port is None:
                hub.message_info(f"Waiting for peripheral {name} to attach to a port")
                if first_delay:
                    first_delay = False
                    await sleep(0.1)
                else:
                    await sleep(1)

        # Start each hub
        task_run = spawn(hub.run())
        hub_tasks.append(task_run)

    # Now wait for the tasks to finish
    ble_q.message_info(f'Waiting for hubs to end')

    for task in hub_tasks:
        await task
    ble_q.message_info(f'Hubs end')
    await ble_q.disconnect()
    for task in hub_peripheral_listen_tasks:
        task.cancel()

    # Print out the port information in debug mode
    for hub in Hub.hubs:
        if hub.query_port_info:
            hub.message_info(pprint.pformat(hub.port_info))

def start(user_system_setup_func): #pragma: no cover
    """
        Main entry point into running everything.

        Just pass in the async co-routine that instantiates all your hubs, and this
        function will take care of the rest.  This includes:

        - Initializing the bluetooth interface object
        - Starting up the user async co-routines inside the asyncio event loop
    """
    loop = get_event_loop()
    loop.run_until_complete(main(user_system_setup_func))




1			# Copyright 2019 Virantha N. Ekanayake
2			#
3			# Licensed under the Apache License, Version 2.0 (the "License");
4			# you may not use this file except in compliance with the License.
5			# You may obtain a copy of the License at
6			#
7			# http://www.apache.org/licenses/LICENSE-2.0
8			#
9			# Unless required by applicable law or agreed to in writing, software
10			# distributed under the License is distributed on an "AS IS" BASIS,
11			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12			# See the License for the specific language governing permissions and
13			# limitations under the License.
14
15			"""Utility functions to attach sensors/motors and start the whole event loop
16
17			#. The decorator :class:`attach` to specify peripherals that
18			connect to a hub (which enables sensing and motor control functions),
19			#. The function :func:`start` that starts running the BLE communication queue, and all the hubs, in the event-loop system
20
21			"""
22
23			import logging
24			import pprint
25			from asyncio import run, sleep, Queue, get_event_loop
26			from asyncio import create_task as spawn
27			from functools import partial, wraps
28			import uuid
29
30			# Local imports
31			from .process import Process
32			from .ble_queue import BLEventQ
33			from .hub import PoweredUpHub, BoostHub, Hub
34			from .sockets import bricknil_socket_server
35
36			import threading
37
38			# Actual decorator that sets up the peripheral classes
39			# noinspection PyPep8Naming
40			class attach:
41			""" Class-decorator to attach peripherals onto a Hub
42
43			Injects sub-classes of `Peripheral` as instance variables on a Hub
44			such as the PoweredUp Hub, akin to "attaching" a physical sensor or
45			motor onto the Hub.
46
47			Before you attach a peripheral with sensing capabilities,
48			you need to ensure your `Peripheral` sub-class has the matching
49			call-back method 'peripheralname_change'.
50
51			Examples::
52
53			@attach(PeripheralType,
54			name="instance name",
55			port='port',
56			capabilities=[])
57
58			Warnings:
59			- No support for checking to make sure user put in correct parameters
60			- Identifies capabilities that need a callback update handler based purely on
61			checking if the capability name starts with the string "sense*"
62
63			"""
64			def __init__(self, peripheral_type, **kwargs):
65			# TODO: check here to make sure parameters were entered
66			if logging.getLogger().getEffectiveLevel() == logging.DEBUG:
67			print(f'decorating with {peripheral_type}')
68			self.peripheral_type = peripheral_type
69			self.kwargs = kwargs
70
71			def __call__ (self, cls):
72			"""
73			Since the actual Hub sub-class being decorated can have __init__ params,
74			we need to have a wrapper function inside here to capture the arguments
75			going into that __init__ call.
76
77			Inside that wrapper, we do the following:
78
79			# Instance the peripheral that was decorated with the saved **kwargs
80			# Check that any `sense_*` capabiilities in the peripheral have an
81			appropriate handler method in the hub class being decorated.
82			# Instance the Hub
83			# Set the peripheral instance as an instance variable on the hub via the
84			`Hub.attach_sensor` method
85
86			"""
87			# Define a wrapper function to capture the actual instantiation and __init__ params
88			@wraps(cls)
89			def wrapper_f(args, *kwargs):
90			#print(f'type of cls is {type(cls)}')
91			peripheral = self.peripheral_type(**self.kwargs)
92
93			# Ugly, but scan through and check if any of the capabilities are sense_*
94			if any([cap.name.startswith('sense') for cap in peripheral.capabilities]):
95			handler_name = f'{peripheral.name}_change'
96			assert hasattr(cls, handler_name), f'{cls.__name__} needs a handler {handler_name}'
97			# Create the hub process and attach this peripheral
98			o = cls(args, *kwargs)
99			o.message_debug(f"Decorating class {cls.__name__} with {self.peripheral_type.__name__}")
100			o.attach_sensor(peripheral)
101			return o
102			return wrapper_f
103
104			async def main(system):
105			"""
106			Entry-point coroutine that handles everything. This is to be run run
107			in bricknil's main loop.
108
109			You normally don't need to use this directly, instead use start()
110			"""
111
112			# Instantiate the Bluetooth LE handler/queue
113			ble_q = BLEventQ()
114			# The web client out_going queue
115			web_out_queue = Queue()
116			# Instantiate socket listener
117			# task_socket = await spawn(socket_server, web_out_queue, ('',25000))
118			# task_tcp = await spawn(bricknil_socket_server, web_out_queue, ('',25000))
119			# await task_tcp.join()
120
121			# Call the user's system routine to instantiate the processes
122			await system()
123
124			hub_tasks = []
125			hub_peripheral_listen_tasks = [] # Need to cancel these at the end
126
127			# Connect all the hubs first before enabling any of them
128			for hub in Hub.hubs:
129			hub.web_queue_out = web_out_queue
130			task_connect = spawn(ble_q.connect(hub))
131			await task_connect
132
133			for hub in Hub.hubs:
134			# Start the peripheral listening loop in each hub
135			task_listen = spawn(hub.peripheral_message_loop())
136			hub_peripheral_listen_tasks.append(task_listen)
137
138			# Need to wait here until all the ports are set
139			# Use a faster timeout the first time (for speeding up testing)
140			first_delay = True
141			for name, peripheral in hub.peripherals.items():
142			while peripheral.port is None:
143			hub.message_info(f"Waiting for peripheral {name} to attach to a port")
144			if first_delay:
145			first_delay = False
146			await sleep(0.1)
147			else:
148			await sleep(1)
149
150			# Start each hub
151			task_run = spawn(hub.run())
152			hub_tasks.append(task_run)
153
154			# Now wait for the tasks to finish
155			ble_q.message_info(f'Waiting for hubs to end')
156
157			for task in hub_tasks:
158			await task
159			ble_q.message_info(f'Hubs end')
160			await ble_q.disconnect()
161			for task in hub_peripheral_listen_tasks:
162			task.cancel()
163
164			# Print out the port information in debug mode
165			for hub in Hub.hubs:
166			if hub.query_port_info:
167			hub.message_info(pprint.pformat(hub.port_info))
168
169			def start(user_system_setup_func): #pragma: no cover
170			"""
171			Main entry point into running everything.
172
173			Just pass in the async co-routine that instantiates all your hubs, and this
174			function will take care of the rest. This includes:
175
176			- Initializing the bluetooth interface object
177			- Starting up the user async co-routines inside the asyncio event loop
178			"""
179			loop = get_event_loop()
180			loop.run_until_complete(main(user_system_setup_func))
181
182
183

virantha / bricknil

GitHub Access Token became invalid

Pull Request — master (#11)

bricknil.bricknil.main() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like