10963943887

Committed 20 Sep 2024 05:50PM UTC coverage: 40.161%. Remained the same

Build # 10963943887

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #480 from ChristianTremblay/develop

Develop

Run Details

0 of 4 new or added lines in 1 file covered. (0.0%)

2149 of 5351 relevant lines covered (40.16%)

0.4 hits per line

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

16.67

/BAC0/core/functions/Discover.py

import asyncio
import typing as t

from bacpypes3.app import Application
from bacpypes3.pdu import Address, LocalBroadcast
from bacpypes3.primitivedata import ObjectIdentifier

from BAC0.core.app.asyncApp import BAC0Application

from ...core.utils.notes import note_and_log


@note_and_log
class Discover:
    """
    Main function to explore the network and find devices.
    """

    @property
    def known_network_numbers(self) -> t.Set[int]:
        return self.this_application._learnedNetworks

    def discover(
        self,
        networks: t.Union[str, t.List[int], int] = "known",
        limits: t.Tuple[int, int] = (0, 4194303),
        global_broadcast: bool = False,
        reset: bool = False,
    ) -> None:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        asyncio.create_task(
            self._discover(
                networks=networks,
                limits=limits,
                global_broadcast=global_broadcast,
                reset=reset,
            )
        )

    async def _discover(
        self,
        networks: t.Union[str, t.List[int], int] = "known",
        limits: t.Tuple[int, int] = (0, 4194303),
        global_broadcast: bool = False,
        timeout: int = 3,
        reset: bool = False,
    ) -> None:
        """
        Discover is meant to be the function used to explore the network when we
        connect.
        It will trigger whois request using the different options provided with
        parameters.

        By default, a local broadcast will be used. This is required as in big
        BACnet network, global broadcast can lead to network flood and loss of data.

        If not parameters are given, BAC0 will try to :

            * Find the network on which it is
            * Find routers for other networks (accessible via local broadcast)
            * Detect "known networks"
            * Use the list of known networks and create whois request to find all devices on those networks

        This should be sufficient for most cases.

        Once discovery is done, user may access the list of "discovered devices" using ::

            bacnet.discoveredDevices

        :param networks (list, integer) : A simple integer or a list of integer
            representing the network numbers used to issue whois request.

        :param limits (tuple) : tuple made of 2 integer, the low limit and the high
            limit. Those are the device instances used in the creation of the
            whois request. Min : 0 ; Max : 4194303

        :param global_broadcast (boolean) : If set to true, a global broadcast
            will be used for the whois. Use with care.
        """
        if reset:
            self.discoveredDevices = {}
        found = []

        _this_application: BAC0Application = self.this_application
        _app: Application = _this_application.app

        _networks = _this_application._learnedNetworks

        deviceInstanceRangeLowLimit, deviceInstanceRangeHighLimit = limits
        # Try to find on which network we are
        _this_network = await self.what_is_network_number()
        if _this_network:
            _networks.add(_this_network)
        # Try to find local routers...
        _other_networks = await self.whois_router_to_network()
        if _other_networks == []:
            _other_networks = await self.whois_router_to_network(global_broadcast=True)
        for each in _other_networks:
            network_adapter, _iamrtn = each
            _networks.update(_iamrtn.iartnNetworkList)
        for net in _networks:
            _this_application._learnedNetworks.add(net)
        self.log(f"Found those networks : {self.known_network_numbers}", level="info")

        if networks:
            if isinstance(networks, list):
                # we'll make multiple whois...
                for network in networks:
                    if network < 65535:
                        _networks.add(network)
            elif networks == "known":
                _networks = self.known_network_numbers.copy()
            else:
                if isinstance(networks, int) and networks < 65535:
                    _networks.add(networks)

        if _networks and not global_broadcast:
            for each_network in _networks:
                self.log(f"Discovering network {each_network}", level="info")
                _res = await self.this_application.app.who_is(
                    low_limit=deviceInstanceRangeLowLimit,
                    high_limit=deviceInstanceRangeHighLimit,
                    address=Address(f"{each_network}:*"),
                    timeout=timeout,
                )
                for each in _res:
                    found.append((each, each_network))

        else:
            msg = (
                "Global braodacast required"
                if global_broadcast
                else "No BACnet network found"
            )
            self.log(
                f"{msg}, attempting a simple whois using provided device instances limits ({deviceInstanceRangeLowLimit} - {deviceInstanceRangeHighLimit})",
                level="info",
            )
            if global_broadcast is True:
                self._log.warning(
                    "Issuing a global Broadcast whois can create network flood. Use with care."
                )
            else:
                self.log("Issuing a local broadcast whois request.", level="info")
            _address = None if global_broadcast is True else LocalBroadcast()
            _res = await _app.who_is(
                low_limit=deviceInstanceRangeLowLimit,
                high_limit=deviceInstanceRangeHighLimit,
                address=_address,
                timeout=timeout,
            )
            for each in _res:
                found.append((each, _this_network))

        for iam_request, network_number in found:
            self.log(
                f"Found device {iam_request} on network {network_number}", level="debug"
            )
            if not self.discoveredDevices:
                self.discoveredDevices = {}  # we can add device as we found some...
            device_address: Address = iam_request.pduSource
            objid: ObjectIdentifier = iam_request.iAmDeviceIdentifier
            self.log(str(objid), level="debug")
            self.log(f"Discovered : {self.discoveredDevices.keys()}", level="debug")
            key = str(objid)
            if key in self.discoveredDevices:
                self._log.debug(
                    f"{objid} already in discovered devices. Adding network number {network_number} to the list."
                )
                self.discoveredDevices[key]["network_number"].add(network_number)
            else:
                self._log.debug(
                    f"Adding {objid} to discovered devices in network {network_number}."
                )
                self.discoveredDevices[key] = {
                    "object_instance": objid,
                    "address": device_address,
                    "network_number": {network_number},
                    "vendor_id": iam_request.vendorID,
                    "vendor_name": "unknown",
                }

        self.log(
            f"Discovery done. Found {len(self.discoveredDevices) if self.discoveredDevices else 0} devices on {len(_networks) if _networks else 0} BACnet networks.",
            level="info",
        )

1	import asyncio	1✔
2	import typing as t	1✔
3
4	from bacpypes3.app import Application	1✔
5	from bacpypes3.pdu import Address, LocalBroadcast	1✔
6	from bacpypes3.primitivedata import ObjectIdentifier	1✔
7
8	from BAC0.core.app.asyncApp import BAC0Application	1✔
9
10	from ...core.utils.notes import note_and_log	1✔
11
12
13	@note_and_log	1✔
14	class Discover:	1✔
15	"""
16	Main function to explore the network and find devices.
17	"""
18
19	@property	1✔
20	def known_network_numbers(self) -> t.Set[int]:	1✔
21	return self.this_application._learnedNetworks	×
22
23	def discover(	1✔
24	self,
25	networks: t.Union[str, t.List[int], int] = "known",
26	limits: t.Tuple[int, int] = (0, 4194303),
27	global_broadcast: bool = False,
28	reset: bool = False,
29	) -> None:
30	try:	×
31	loop = asyncio.get_running_loop()	×
32	except RuntimeError:	×
33	loop = asyncio.new_event_loop()	×
34	asyncio.set_event_loop(loop)	×
35	asyncio.create_task(	×
36	self._discover(
37	networks=networks,
38	limits=limits,
39	global_broadcast=global_broadcast,
40	reset=reset,
41	)
42	)
43
44	async def _discover(	1✔
45	self,
46	networks: t.Union[str, t.List[int], int] = "known",
47	limits: t.Tuple[int, int] = (0, 4194303),
48	global_broadcast: bool = False,
49	timeout: int = 3,
50	reset: bool = False,
51	) -> None:
52	"""
53	Discover is meant to be the function used to explore the network when we
54	connect.
55	It will trigger whois request using the different options provided with
56	parameters.
57
58	By default, a local broadcast will be used. This is required as in big
59	BACnet network, global broadcast can lead to network flood and loss of data.
60
61	If not parameters are given, BAC0 will try to :
62
63	* Find the network on which it is
64	* Find routers for other networks (accessible via local broadcast)
65	* Detect "known networks"
66	* Use the list of known networks and create whois request to find all devices on those networks
67
68	This should be sufficient for most cases.
69
70	Once discovery is done, user may access the list of "discovered devices" using ::
71
72	bacnet.discoveredDevices
73
74	:param networks (list, integer) : A simple integer or a list of integer
75	representing the network numbers used to issue whois request.
76
77	:param limits (tuple) : tuple made of 2 integer, the low limit and the high
78	limit. Those are the device instances used in the creation of the
79	whois request. Min : 0 ; Max : 4194303
80
81	:param global_broadcast (boolean) : If set to true, a global broadcast
82	will be used for the whois. Use with care.
83	"""
84	if reset:	×
85	self.discoveredDevices = {}	×
86	found = []	×
87
88	_this_application: BAC0Application = self.this_application	×
89	_app: Application = _this_application.app	×
90
91	_networks = _this_application._learnedNetworks	×
92
93	deviceInstanceRangeLowLimit, deviceInstanceRangeHighLimit = limits	×
94	# Try to find on which network we are
95	_this_network = await self.what_is_network_number()	×
96	if _this_network:	×
NEW 97	_networks.add(_this_network)	×
98	# Try to find local routers...
99	_other_networks = await self.whois_router_to_network()	×
100	if _other_networks == []:	×
101	_other_networks = await self.whois_router_to_network(global_broadcast=True)	×
102	for each in _other_networks:	×
103	network_adapter, _iamrtn = each	×
NEW 104	_networks.update(_iamrtn.iartnNetworkList)	×
105	for net in _networks:	×
106	_this_application._learnedNetworks.add(net)	×
107	self.log(f"Found those networks : {self.known_network_numbers}", level="info")	×
108
109	if networks:	×
110	if isinstance(networks, list):	×
111	# we'll make multiple whois...
112	for network in networks:	×
113	if network < 65535:	×
NEW 114	_networks.add(network)	×
115	elif networks == "known":	×
116	_networks = self.known_network_numbers.copy()	×
117	else:
118	if isinstance(networks, int) and networks < 65535:	×
NEW 119	_networks.add(networks)	×
120
121	if _networks and not global_broadcast:	×
122	for each_network in _networks:	×
123	self.log(f"Discovering network {each_network}", level="info")	×
124	_res = await self.this_application.app.who_is(	×
125	low_limit=deviceInstanceRangeLowLimit,
126	high_limit=deviceInstanceRangeHighLimit,
127	address=Address(f"{each_network}:*"),
128	timeout=timeout,
129	)
130	for each in _res:	×
131	found.append((each, each_network))	×
132
133	else:
134	msg = (	×
135	"Global braodacast required"
136	if global_broadcast
137	else "No BACnet network found"
138	)
139	self.log(	×
140	f"{msg}, attempting a simple whois using provided device instances limits ({deviceInstanceRangeLowLimit} - {deviceInstanceRangeHighLimit})",
141	level="info",
142	)
143	if global_broadcast is True:	×
144	self._log.warning(	×
145	"Issuing a global Broadcast whois can create network flood. Use with care."
146	)
147	else:
148	self.log("Issuing a local broadcast whois request.", level="info")	×
149	_address = None if global_broadcast is True else LocalBroadcast()	×
150	_res = await _app.who_is(	×
151	low_limit=deviceInstanceRangeLowLimit,
152	high_limit=deviceInstanceRangeHighLimit,
153	address=_address,
154	timeout=timeout,
155	)
156	for each in _res:	×
157	found.append((each, _this_network))	×
158
159	for iam_request, network_number in found:	×
160	self.log(	×
161	f"Found device {iam_request} on network {network_number}", level="debug"
162	)
163	if not self.discoveredDevices:	×
164	self.discoveredDevices = {} # we can add device as we found some...	×
165	device_address: Address = iam_request.pduSource	×
166	objid: ObjectIdentifier = iam_request.iAmDeviceIdentifier	×
167	self.log(str(objid), level="debug")	×
168	self.log(f"Discovered : {self.discoveredDevices.keys()}", level="debug")	×
169	key = str(objid)	×
170	if key in self.discoveredDevices:	×
171	self._log.debug(	×
172	f"{objid} already in discovered devices. Adding network number {network_number} to the list."
173	)
174	self.discoveredDevices[key]["network_number"].add(network_number)	×
175	else:
176	self._log.debug(	×
177	f"Adding {objid} to discovered devices in network {network_number}."
178	)
179	self.discoveredDevices[key] = {	×
180	"object_instance": objid,
181	"address": device_address,
182	"network_number": {network_number},
183	"vendor_id": iam_request.vendorID,
184	"vendor_name": "unknown",
185	}
186
187	self.log(	×
188	f"Discovery done. Found {len(self.discoveredDevices) if self.discoveredDevices else 0} devices on {len(_networks) if _networks else 0} BACnet networks.",
189	level="info",
190	)

ChristianTremblay / BAC0 / 10963943887

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