kytos.core.switch.Switch.where_is_mac() - Code Metrics - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

kytos.core.switch.Switch.where_is_mac() A
last analyzed 2025-06-18 19:57 UTC

↳ Parent: kytos.core.switch

Complexity

Conditions

Size

Total Lines	14
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	5
nop	2
dl	0
loc	14
rs	10
c	0
b	0
f	0

"""Module with main classes related to Switches."""
import json
import logging
import operator
from collections import OrderedDict
from functools import reduce
from threading import Lock

from kytos.core.common import EntityStatus, GenericEntity
from kytos.core.constants import CONNECTION_TIMEOUT, FLOOD_TIMEOUT
from kytos.core.helpers import get_time, now
from kytos.core.interface import Interface

__all__ = ('Switch',)

LOG = logging.getLogger(__name__)


class Switch(GenericEntity):
    """Switch class is an abstraction from switches.

    A new Switch will be created every time the handshake process is done
    (after receiving the first FeaturesReply). Considering this, the
    :attr:`socket`, :attr:`connection_id`, :attr:`of_version` and
    :attr:`features` need to be passed on init. But when the connection to the
    switch is lost, then this attributes can be set to None (actually some of
    them must be).

    The :attr:`dpid` attribute will be the unique identifier of a Switch.
    It is the :attr:`pyof.*.controller2switch.SwitchFeatures.datapath-id` that
    defined by the OpenFlow Specification, it is a 64 bit field that should be
    thought of as analogous to a Ethernet Switches bridge MAC, its a unique
    identifier for the specific packet processing pipeline being managed. One
    physical switch may have more than one datapath-id (think virtualization of
    the switch).

    :attr:`socket` is the request from a TCP connection, it represents the
    effective connection between the switch and the controller.

    :attr:`connection_id` is a tuple, composed by the ip and port of the
    established connection (if any). It will be used to help map the connection
    to the Switch and vice-versa.

    :attr:`ofp_version` is a string representing the accorded version of
    python-openflow that will be used on the communication between the
    Controller and the Switch.

    :attr:`features` is an instance of
    :class:`pyof.*.controller2switch.FeaturesReply` representing the current
    features of the switch.
    """

    status_funcs = OrderedDict()
    status_reason_funcs = OrderedDict()

    def __init__(self, dpid, connection=None, features=None):
        """Contructor of switches have the below parameters.

        Args:
          dpid (|DPID|): datapath_id of the switch
          connection (:class:`~.Connection`): Connection used by switch.
          features (|features_reply|): FeaturesReply instance.

        """
        self.dpid = dpid
        self.connection = connection
        self.features = features
        self.firstseen = now()
        self.lastseen = get_time("0001-01-01T00:00:00")
        self.sent_xid = None
        self.waiting_for_reply = False
        self.request_timestamp = 0
        #: Dict associating mac addresses to switch ports.
        #:      the key of this dict is a mac_address, and the value is a set
        #:      containing the ports of this switch in which that mac can be
        #:      found.
        self.mac2port = {}
        #: This flood_table will keep track of flood packets to avoid over
        #:     flooding on the network. Its key is a hash composed by
        #:     (eth_type, mac_src, mac_dst) and the value is the timestamp of
        #:     the last flood.
        self.flood_table = {}
        self.interfaces: dict[int, Interface] = {}
        self.flows = []
        self.description = {}
        self._id = dpid
        self._interface_lock = Lock()

        if connection:
            connection.switch = self

        super().__init__()

    def __repr__(self):
        return f"Switch('{self.dpid}')"

    def update_description(self, desc):
        """Update switch'descriptions from Switch instance.

        Args:
            desc (|desc_stats|):
                Description Class with new values of switch's descriptions.
        """
        self.description['manufacturer'] = desc.mfr_desc.value
        self.description['hardware'] = desc.hw_desc.value
        self.description['software'] = desc.sw_desc.value
        self.description['serial'] = desc.serial_num.value
        self.description['data_path'] = desc.dp_desc.value

    @property
    def id(self):  # pylint: disable=invalid-name
        """Return id from Switch instance.

        Returns:
            string: the switch id is the data_path_id from switch.

        """
        return self._id

    @property
    def ofp_version(self):
        """Return the OFP version of this switch."""
        if self.connection:
            return '0x0' + str(self.connection.protocol.version)
        return None

    @property

    def status(self):
        """Return the current status of the Entity."""
        state = super().status
        if state == EntityStatus.DISABLED:
            return state

        for status_func in self.status_funcs.values():
            if status_func(self) == EntityStatus.DOWN:
                return EntityStatus.DOWN
        return state

    @classmethod
    def register_status_func(cls, name: str, func):
        """Register status func given its name and a callable at setup time."""
        cls.status_funcs[name] = func

    @classmethod
    def register_status_reason_func(cls, name: str, func):
        """Register status reason func given its name
        and a callable at setup time."""
        cls.status_reason_funcs[name] = func

    @property
    def status_reason(self):
        """Return the reason behind the current status of the entity."""
        return reduce(
            operator.or_,
            map(
                lambda x: x(self),
                self.status_reason_funcs.values()
            ),
            super().status_reason
        )

    def disable(self):
        """Disable this switch instance.

        Also disable this switch's interfaces and their respective links.
        """
        for interface in self.interfaces.values():
            interface.disable()
        self._enabled = False

    def disconnect(self):
        """Disconnect the switch instance."""
        self.connection.close()
        self.connection = None
        LOG.info("Switch %s is disconnected", self.dpid)

    def get_interface_by_port_no(self, port_no):
        """Get interface by port number from Switch instance.

        Returns:
            :class:`~.core.switch.Interface`: Interface from specific port.

        """
        # pyof v0x01 stores port_no in the value attribute
        port_no = getattr(port_no, 'value', port_no)

        return self.interfaces.get(port_no)

    def update_or_create_interface(self, port_no, name=None, address=None,
                                   state=None, features=None, speed=None,
                                   config=None):
        """Get and upated an interface or create one if it does not exist."""
        with self._interface_lock:
            interface = self.get_interface_by_port_no(port_no)
            if interface:
                interface.name = name or interface.name
                interface.address = address or interface.address
                interface.state = state or interface.state
                interface.features = features or interface.features
                interface.config = config
                if speed:
                    interface.set_custom_speed(speed)
            else:
                interface = Interface(name=name,
                                      address=address,
                                      port_number=port_no,
                                      switch=self,
                                      state=state,
                                      features=features,
                                      speed=speed,
                                      config=config)
                self.update_interface(interface)
            return interface

    def get_flow_by_id(self, flow_id):
        """Return a Flow using the flow_id given. None if not found in flows.

        Args:
            flow_id (int): identifier from specific flow stored.
        """
        for flow in self.flows:
            if flow_id == flow.id:
                return flow
        return None

    def is_active(self):
        """Return true if the switch connection is alive."""
        return self.is_connected()

    def is_connected(self):
        """Verify if the switch is connected to a socket."""
        return (self.connection is not None and
                self.connection.is_alive() and
                self.connection.is_established() and
                (now() - self.lastseen).seconds <= CONNECTION_TIMEOUT)

    def update_connection(self, connection):
        """Update switch connection.

        Args:
            connection (:class:`~.core.switch.Connection`):
                New connection to this instance of switch.
        """
        self.connection = connection
        self.connection.switch = self

    def update_features(self, features):
        """Update :attr:`features` attribute."""
        self.features = features

    def send(self, buffer):
        """Send a buffer data to the real switch.

        Args:
          buffer (bytes): bytes to be sent to the switch throught its
                            connection.
        """
        if self.connection:
            self.connection.send(buffer)

    def update_lastseen(self):
        """Update the lastseen attribute."""
        self.lastseen = now()

    def update_interface(self, interface):
        """Update or associate a interface from switch instance.

        Args:
            interface (:class:`~kytos.core.switch.Interface`):
                Interface object to be stored.
        """
        self.interfaces[interface.port_number] = interface

    def remove_interface(self, interface):
        """Remove a interface from switch instance.

        Args:
            interface (:class:`~kytos.core.switch.Interface`):
                Interface object to be removed.
        """
        del self.interfaces[interface.port_number]

    def update_mac_table(self, mac, port_number):
        """Link the mac address with a port number.

        Args:
            mac (|hw_address|): mac address from switch.
            port (int): port linked in mac address.
        """
        if mac.value in self.mac2port:
            self.mac2port[mac.value].add(port_number)
        else:
            self.mac2port[mac.value] = set([port_number])

    def last_flood(self, ethernet_frame):
        """Return the timestamp when the ethernet_frame was flooded.

        This method is usefull to check if a frame was flooded before or not.

        Args:
            ethernet_frame (|ethernet|): Ethernet instance to be verified.

        Returns:
            datetime.datetime.now:
                Last time when the ethernet_frame was flooded.

        """
        try:
            return self.flood_table[ethernet_frame.get_hash()]
        except KeyError:
            return None

    def should_flood(self, ethernet_frame):
        """Verify if the ethernet frame should flood.

        Args:
            ethernet_frame (|ethernet|): Ethernet instance to be verified.

        Returns:
            bool: True if the ethernet_frame should flood.

        """
        last_flood = self.last_flood(ethernet_frame)
        diff = (now() - last_flood).microseconds

        return last_flood is None or diff > FLOOD_TIMEOUT

    def update_flood_table(self, ethernet_frame):
        """Update a flood table using the given ethernet frame.

        Args:
            ethernet_frame (|ethernet|): Ethernet frame to be updated.
        """
        self.flood_table[ethernet_frame.get_hash()] = now()

    def where_is_mac(self, mac):
        """Return all ports from specific mac address.

        Args:
            mac (|hw_address|): Mac address from switch.

        Returns:
            :class:`list`: A list of ports. None otherswise.

        """
        try:
            return list(self.mac2port[mac.value])
        except KeyError:
            return None

    def as_dict(self):
        """Return a dictionary with switch attributes.

        Example of output:

        .. code-block:: python3

               {'id': '00:00:00:00:00:00:00:03:2',
                'name': '00:00:00:00:00:00:00:03:2',
                'dpid': '00:00:00:00:03',
                'connection':  connection,
                'ofp_version': '0x01',
                'type': 'switch',
                'manufacturer': "",
                'serial': "",
                'hardware': "Open vSwitch",
                'software': 2.5,
                'data_path': "",
                'interfaces': {},
                'metadata': {},
                'active': True,
                'enabled': False,
                'status': 'DISABLED',
                }

        Returns:
            dict: Dictionary filled with interface attributes.

        """
        connection = ""
        if self.connection is not None:
            address = self.connection.address
            port = self.connection.port
            connection = f"{address}:{port}"

        return {
            'id': self.id,
            'name': self.id,
            'dpid': self.dpid,
            'connection': connection,
            'ofp_version': self.ofp_version,
            'type': 'switch',
            'manufacturer': self.description.get('manufacturer', ''),
            'serial': self.description.get('serial', ''),
            'hardware': self.description.get('hardware', ''),
            'software': self.description.get('software'),
            'data_path': self.description.get('data_path', ''),
            'interfaces': {
                i.id: i.as_dict()
                for i in self.interfaces.values()
            },
            'metadata': self.metadata,
            'active': self.is_active(),
            'enabled': self.is_enabled(),
            'status': self.status.value,
            'status_reason': sorted(self.status_reason),
        }

    def as_json(self):
        """Return JSON with switch's attributes.

        Example of output:

        .. code-block:: json

            {"data_path": "",
             "hardware": "Open vSwitch",
             "dpid": "00:00:00:00:03",
             "name": "00:00:00:00:00:00:00:03:2",
             "manufacturer": "",
             "serial": "",
             "software": 2.5,
             "id": "00:00:00:00:00:00:00:03:2",
             "ofp_version": "0x01",
             "type": "switch",
             "connection": ""}

        Returns:
            string: JSON filled with switch's attributes.

        """
        return json.dumps(self.as_dict())

    @classmethod
    def from_dict(cls, switch_dict):
        """Return a Switch instance from python dictionary."""
        return cls(switch_dict.get('dpid'),
                   switch_dict.get('connection'),
                   switch_dict.get('features'))


1		"""Module with main classes related to Switches."""
2		import json
3		import logging
4		import operator
5		from collections import OrderedDict
6		from functools import reduce
7		from threading import Lock
8
9		from kytos.core.common import EntityStatus, GenericEntity
10		from kytos.core.constants import CONNECTION_TIMEOUT, FLOOD_TIMEOUT
11		from kytos.core.helpers import get_time, now
12		from kytos.core.interface import Interface
13
14		__all__ = ('Switch',)
15
16		LOG = logging.getLogger(__name__)
17
18
19		class Switch(GenericEntity):
20		"""Switch class is an abstraction from switches.
21
22		A new Switch will be created every time the handshake process is done
23		(after receiving the first FeaturesReply). Considering this, the
24		:attr:`socket`, :attr:`connection_id`, :attr:`of_version` and
25		:attr:`features` need to be passed on init. But when the connection to the
26		switch is lost, then this attributes can be set to None (actually some of
27		them must be).
28
29		The :attr:`dpid` attribute will be the unique identifier of a Switch.
30		It is the :attr:`pyof.*.controller2switch.SwitchFeatures.datapath-id` that
31		defined by the OpenFlow Specification, it is a 64 bit field that should be
32		thought of as analogous to a Ethernet Switches bridge MAC, its a unique
33		identifier for the specific packet processing pipeline being managed. One
34		physical switch may have more than one datapath-id (think virtualization of
35		the switch).
36
37		:attr:`socket` is the request from a TCP connection, it represents the
38		effective connection between the switch and the controller.
39
40		:attr:`connection_id` is a tuple, composed by the ip and port of the
41		established connection (if any). It will be used to help map the connection
42		to the Switch and vice-versa.
43
44		:attr:`ofp_version` is a string representing the accorded version of
45		python-openflow that will be used on the communication between the
46		Controller and the Switch.
47
48		:attr:`features` is an instance of
49		:class:`pyof.*.controller2switch.FeaturesReply` representing the current
50		features of the switch.
51		"""
52
53		status_funcs = OrderedDict()
54		status_reason_funcs = OrderedDict()
55
56		def __init__(self, dpid, connection=None, features=None):
57		"""Contructor of switches have the below parameters.
58
59		Args:
60		dpid (\|DPID\|): datapath_id of the switch
61		connection (:class:`~.Connection`): Connection used by switch.
62		features (\|features_reply\|): FeaturesReply instance.
63
64		"""
65		self.dpid = dpid
66		self.connection = connection
67		self.features = features
68		self.firstseen = now()
69		self.lastseen = get_time("0001-01-01T00:00:00")
70		self.sent_xid = None
71		self.waiting_for_reply = False
72		self.request_timestamp = 0
73		#: Dict associating mac addresses to switch ports.
74		#: the key of this dict is a mac_address, and the value is a set
75		#: containing the ports of this switch in which that mac can be
76		#: found.
77		self.mac2port = {}
78		#: This flood_table will keep track of flood packets to avoid over
79		#: flooding on the network. Its key is a hash composed by
80		#: (eth_type, mac_src, mac_dst) and the value is the timestamp of
81		#: the last flood.
82		self.flood_table = {}
83		self.interfaces: dict[int, Interface] = {}
84		self.flows = []
85		self.description = {}
86		self._id = dpid
87		self._interface_lock = Lock()
88
89		if connection:
90		connection.switch = self
91
92		super().__init__()
93
94		def __repr__(self):
95		return f"Switch('{self.dpid}')"
96
97		def update_description(self, desc):
98		"""Update switch'descriptions from Switch instance.
99
100		Args:
101		desc (\|desc_stats\|):
102		Description Class with new values of switch's descriptions.
103		"""
104		self.description['manufacturer'] = desc.mfr_desc.value
105		self.description['hardware'] = desc.hw_desc.value
106		self.description['software'] = desc.sw_desc.value
107		self.description['serial'] = desc.serial_num.value
108		self.description['data_path'] = desc.dp_desc.value
109
110		@property
111		def id(self): # pylint: disable=invalid-name
112		"""Return id from Switch instance.
113
114		Returns:
115		string: the switch id is the data_path_id from switch.
116
117		"""
118		return self._id
119
120		@property
121		def ofp_version(self):
122		"""Return the OFP version of this switch."""
123		if self.connection:
124		return '0x0' + str(self.connection.protocol.version)
125		return None
126
127	View Code Duplication	@property
		0 ignored issues – show Duplication introduced 2022-11-17 14:25 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
128		def status(self):
129		"""Return the current status of the Entity."""
130		state = super().status
131		if state == EntityStatus.DISABLED:
132		return state
133
134		for status_func in self.status_funcs.values():
135		if status_func(self) == EntityStatus.DOWN:
136		return EntityStatus.DOWN
137		return state
138
139		@classmethod
140		def register_status_func(cls, name: str, func):
141		"""Register status func given its name and a callable at setup time."""
142		cls.status_funcs[name] = func
143
144		@classmethod
145		def register_status_reason_func(cls, name: str, func):
146		"""Register status reason func given its name
147		and a callable at setup time."""
148		cls.status_reason_funcs[name] = func
149
150		@property
151		def status_reason(self):
152		"""Return the reason behind the current status of the entity."""
153		return reduce(
154		operator.or_,
155		map(
156		lambda x: x(self),
157		self.status_reason_funcs.values()
158		),
159		super().status_reason
160		)
161
162		def disable(self):
163		"""Disable this switch instance.
164
165		Also disable this switch's interfaces and their respective links.
166		"""
167		for interface in self.interfaces.values():
168		interface.disable()
169		self._enabled = False
170
171		def disconnect(self):
172		"""Disconnect the switch instance."""
173		self.connection.close()
174		self.connection = None
175		LOG.info("Switch %s is disconnected", self.dpid)
176
177		def get_interface_by_port_no(self, port_no):
178		"""Get interface by port number from Switch instance.
179
180		Returns:
181		:class:`~.core.switch.Interface`: Interface from specific port.
182
183		"""
184		# pyof v0x01 stores port_no in the value attribute
185		port_no = getattr(port_no, 'value', port_no)
186
187		return self.interfaces.get(port_no)
188
189		def update_or_create_interface(self, port_no, name=None, address=None,
190		state=None, features=None, speed=None,
191		config=None):
192		"""Get and upated an interface or create one if it does not exist."""
193		with self._interface_lock:
194		interface = self.get_interface_by_port_no(port_no)
195		if interface:
196		interface.name = name or interface.name
197		interface.address = address or interface.address
198		interface.state = state or interface.state
199		interface.features = features or interface.features
200		interface.config = config
201		if speed:
202		interface.set_custom_speed(speed)
203		else:
204		interface = Interface(name=name,
205		address=address,
206		port_number=port_no,
207		switch=self,
208		state=state,
209		features=features,
210		speed=speed,
211		config=config)
212		self.update_interface(interface)
213		return interface
214
215		def get_flow_by_id(self, flow_id):
216		"""Return a Flow using the flow_id given. None if not found in flows.
217
218		Args:
219		flow_id (int): identifier from specific flow stored.
220		"""
221		for flow in self.flows:
222		if flow_id == flow.id:
223		return flow
224		return None
225
226		def is_active(self):
227		"""Return true if the switch connection is alive."""
228		return self.is_connected()
229
230		def is_connected(self):
231		"""Verify if the switch is connected to a socket."""
232		return (self.connection is not None and
233		self.connection.is_alive() and
234		self.connection.is_established() and
235		(now() - self.lastseen).seconds <= CONNECTION_TIMEOUT)
236
237		def update_connection(self, connection):
238		"""Update switch connection.
239
240		Args:
241		connection (:class:`~.core.switch.Connection`):
242		New connection to this instance of switch.
243		"""
244		self.connection = connection
245		self.connection.switch = self
246
247		def update_features(self, features):
248		"""Update :attr:`features` attribute."""
249		self.features = features
250
251		def send(self, buffer):
252		"""Send a buffer data to the real switch.
253
254		Args:
255		buffer (bytes): bytes to be sent to the switch throught its
256		connection.
257		"""
258		if self.connection:
259		self.connection.send(buffer)
260
261		def update_lastseen(self):
262		"""Update the lastseen attribute."""
263		self.lastseen = now()
264
265		def update_interface(self, interface):
266		"""Update or associate a interface from switch instance.
267
268		Args:
269		interface (:class:`~kytos.core.switch.Interface`):
270		Interface object to be stored.
271		"""
272		self.interfaces[interface.port_number] = interface
273
274		def remove_interface(self, interface):
275		"""Remove a interface from switch instance.
276
277		Args:
278		interface (:class:`~kytos.core.switch.Interface`):
279		Interface object to be removed.
280		"""
281		del self.interfaces[interface.port_number]
282
283		def update_mac_table(self, mac, port_number):
284		"""Link the mac address with a port number.
285
286		Args:
287		mac (\|hw_address\|): mac address from switch.
288		port (int): port linked in mac address.
289		"""
290		if mac.value in self.mac2port:
291		self.mac2port[mac.value].add(port_number)
292		else:
293		self.mac2port[mac.value] = set([port_number])
294
295		def last_flood(self, ethernet_frame):
296		"""Return the timestamp when the ethernet_frame was flooded.
297
298		This method is usefull to check if a frame was flooded before or not.
299
300		Args:
301		ethernet_frame (\|ethernet\|): Ethernet instance to be verified.
302
303		Returns:
304		datetime.datetime.now:
305		Last time when the ethernet_frame was flooded.
306
307		"""
308		try:
309		return self.flood_table[ethernet_frame.get_hash()]
310		except KeyError:
311		return None
312
313		def should_flood(self, ethernet_frame):
314		"""Verify if the ethernet frame should flood.
315
316		Args:
317		ethernet_frame (\|ethernet\|): Ethernet instance to be verified.
318
319		Returns:
320		bool: True if the ethernet_frame should flood.
321
322		"""
323		last_flood = self.last_flood(ethernet_frame)
324		diff = (now() - last_flood).microseconds
325
326		return last_flood is None or diff > FLOOD_TIMEOUT
327
328		def update_flood_table(self, ethernet_frame):
329		"""Update a flood table using the given ethernet frame.
330
331		Args:
332		ethernet_frame (\|ethernet\|): Ethernet frame to be updated.
333		"""
334		self.flood_table[ethernet_frame.get_hash()] = now()
335
336		def where_is_mac(self, mac):
337		"""Return all ports from specific mac address.
338
339		Args:
340		mac (\|hw_address\|): Mac address from switch.
341
342		Returns:
343		:class:`list`: A list of ports. None otherswise.
344
345		"""
346		try:
347		return list(self.mac2port[mac.value])
348		except KeyError:
349		return None
350
351		def as_dict(self):
352		"""Return a dictionary with switch attributes.
353
354		Example of output:
355
356		.. code-block:: python3
357
358		{'id': '00:00:00:00:00:00:00:03:2',
359		'name': '00:00:00:00:00:00:00:03:2',
360		'dpid': '00:00:00:00:03',
361		'connection': connection,
362		'ofp_version': '0x01',
363		'type': 'switch',
364		'manufacturer': "",
365		'serial': "",
366		'hardware': "Open vSwitch",
367		'software': 2.5,
368		'data_path': "",
369		'interfaces': {},
370		'metadata': {},
371		'active': True,
372		'enabled': False,
373		'status': 'DISABLED',
374		}
375
376		Returns:
377		dict: Dictionary filled with interface attributes.
378
379		"""
380		connection = ""
381		if self.connection is not None:
382		address = self.connection.address
383		port = self.connection.port
384		connection = f"{address}:{port}"
385
386		return {
387		'id': self.id,
388		'name': self.id,
389		'dpid': self.dpid,
390		'connection': connection,
391		'ofp_version': self.ofp_version,
392		'type': 'switch',
393		'manufacturer': self.description.get('manufacturer', ''),
394		'serial': self.description.get('serial', ''),
395		'hardware': self.description.get('hardware', ''),
396		'software': self.description.get('software'),
397		'data_path': self.description.get('data_path', ''),
398		'interfaces': {
399		i.id: i.as_dict()
400		for i in self.interfaces.values()
401		},
402		'metadata': self.metadata,
403		'active': self.is_active(),
404		'enabled': self.is_enabled(),
405		'status': self.status.value,
406		'status_reason': sorted(self.status_reason),
407		}
408
409		def as_json(self):
410		"""Return JSON with switch's attributes.
411
412		Example of output:
413
414		.. code-block:: json
415
416		{"data_path": "",
417		"hardware": "Open vSwitch",
418		"dpid": "00:00:00:00:03",
419		"name": "00:00:00:00:00:00:00:03:2",
420		"manufacturer": "",
421		"serial": "",
422		"software": 2.5,
423		"id": "00:00:00:00:00:00:00:03:2",
424		"ofp_version": "0x01",
425		"type": "switch",
426		"connection": ""}
427
428		Returns:
429		string: JSON filled with switch's attributes.
430
431		"""
432		return json.dumps(self.as_dict())
433
434		@classmethod
435		def from_dict(cls, switch_dict):
436		"""Return a Switch instance from python dictionary."""
437		return cls(switch_dict.get('dpid'),
438		switch_dict.get('connection'),
439		switch_dict.get('features'))
440

kytos-ng / kytos

kytos.core.switch.Switch.where_is_mac() A last analyzed 2025-06-18 19:57 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

kytos.core.switch.Switch.where_is_mac() A
last analyzed 2025-06-18 19:57 UTC