kytos.core.link.Link.from_dict() - Code Metrics - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

kytos.core.link.Link.from_dict() A
last analyzed 2025-06-18 19:57 UTC

↳ Parent: kytos.core.link

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""Module with all classes related to links.

Links are low level abstractions representing connections between two
interfaces.
"""
import json
import operator
from collections import OrderedDict, defaultdict
from functools import reduce
from threading import Lock
from typing import Union

from kytos.core.common import EntityStatus, GenericEntity
from kytos.core.exceptions import (KytosLinkCreationError,
                                   KytosNoTagAvailableError)
from kytos.core.id import LinkID
from kytos.core.interface import Interface, TAGType
from kytos.core.tag_ranges import range_intersection


class Link(GenericEntity):
    """Define a link between two Endpoints."""

    status_funcs = OrderedDict()
    status_reason_funcs = OrderedDict()
    _get_available_vlans_lock = defaultdict(Lock)

    def __init__(self, endpoint_a, endpoint_b):
        """Create a Link instance and set its attributes.

        Two kytos.core.interface.Interface are required as parameters.
        """
        if endpoint_a is None:
            raise KytosLinkCreationError("endpoint_a cannot be None")
        if endpoint_b is None:
            raise KytosLinkCreationError("endpoint_b cannot be None")
        self._id = LinkID(endpoint_a.id, endpoint_b.id)
        if self._id.interfaces[0] == endpoint_b.id:
            self.endpoint_a: Interface = endpoint_b
            self.endpoint_b: Interface = endpoint_a
        else:
            self.endpoint_a: Interface = endpoint_a
            self.endpoint_b: Interface = endpoint_b

        self.link_lock = Lock()
        super().__init__()

    def __hash__(self):
        return hash(self.id)

    def __repr__(self):
        return f"Link({self.endpoint_a!r}, {self.endpoint_b!r}, {self.id})"

    @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

    @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_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 is_enabled(self):
        """Override the is_enabled method.

        We consider a link enabled when all the interfaces are enabled.

        Returns:
            boolean: True if both interfaces are enabled, False otherwise.

        """
        return (self._enabled and self.endpoint_a.is_enabled() and
                self.endpoint_b.is_enabled())

    def is_active(self):
        """Override the is_active method.

        We consider a link active whether all the interfaces are active.

        Returns:
            boolean: True if the interfaces are active, othewrise False.

        """
        return (self._active and self.endpoint_a.is_active() and
                self.endpoint_b.is_active())

    def __eq__(self, other):
        """Check if two instances of Link are equal."""
        return self.id == other.id

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

        Returns:
            string: link id.

        """
        return self._id

    def available_tags(self, tag_type: str = 'vlan') -> list[list[int]]:
        """Return the available tags for the link.

        Based on the endpoint tags.
        """
        tag_iterator = range_intersection(
            self.endpoint_a.available_tags[tag_type],
            self.endpoint_b.available_tags[tag_type],
        )
        available_tags = list(tag_iterator)
        return available_tags

    def is_tag_available(self, tag: int, tag_type: str = 'vlan'):
        """Check if a tag is available."""
        return (self.endpoint_a.is_tag_available(tag, tag_type) and
                self.endpoint_b.is_tag_available(tag, tag_type))

    def get_next_available_tag(
        self,
        controller,
        link_id: str,
        take_last: bool = False,
        tag_type: str = 'vlan',
        try_avoid_value: int = None,
    ) -> int:
        """Return the next available tag if exists. By default this
         method returns the smallest tag available. Apply options to
         change behavior.
         Options:
           - take_last (bool): Choose the largest tag available.
           - try_avoid_value (int): Avoid given tag if possible. Otherwise
             return what is available.
        """
        with self._get_available_vlans_lock[link_id]:
            with self.endpoint_a._tag_lock:
                with self.endpoint_b._tag_lock:
                    ava_tags_a = self.endpoint_a.available_tags[tag_type]
                    ava_tags_b = self.endpoint_b.available_tags[tag_type]
                    tags = range_intersection(ava_tags_a, ava_tags_b,
                                              take_last)
                    try:
                        tag_range: list = next(tags)
                        if (try_avoid_value is not None and
                                tag_range[take_last] == try_avoid_value):
                            if (tag_range[take_last] !=
                                    tag_range[not take_last]):
                                tag = tag_range[take_last]
                                tag += (+1) if not take_last else (-1)
                            else:
                                try:
                                    tag = next(tags)[take_last]
                                except StopIteration:
                                    tag = tag_range[take_last]
                        else:
                            tag = tag_range[take_last]

                        self.endpoint_a.use_tags(
                            controller, tag, use_lock=False, check_order=False
                        )
                        self.endpoint_b.use_tags(
                            controller, tag, use_lock=False, check_order=False
                        )
                        return tag
                    except StopIteration:
                        raise KytosNoTagAvailableError(self)

    def make_tags_available(
        self,
        controller,
        tags: Union[int, list[int], list[list[int]]],
        link_id,
        tag_type: str = 'vlan',
        check_order: bool = True,
    ) -> tuple[list[list[int]], list[list[int]]]:
        """Add a specific tag in available_tags."""
        with self._get_available_vlans_lock[link_id]:
            with self.endpoint_a._tag_lock:
                with self.endpoint_b._tag_lock:
                    conflict_a = self.endpoint_a.make_tags_available(
                        controller, tags, tag_type, use_lock=False,
                        check_order=check_order
                    )
                    conflict_b = self.endpoint_b.make_tags_available(
                        controller, tags, tag_type, use_lock=False,
                        check_order=check_order
                    )
        return conflict_a, conflict_b

    def available_vlans(self):
        """Get all available vlans from each interface in the link."""
        vlans_a = self._get_available_vlans(self.endpoint_a)
        vlans_b = self._get_available_vlans(self.endpoint_b)
        return [vlan for vlan in vlans_a if vlan in vlans_b]

    @staticmethod
    def _get_available_vlans(endpoint):
        """Return all vlans from endpoint."""
        vlans = endpoint.available_tags
        return [vlan for vlan in vlans if vlan == TAGType.VLAN.value]

    def as_dict(self):
        """Return the Link as a dictionary."""
        return {
            'id': self.id,
            'endpoint_a': self.endpoint_a.as_dict(),
            'endpoint_b': self.endpoint_b.as_dict(),
            'metadata': self.get_metadata_as_dict(),
            'active': self.is_active(),
            'enabled': self.is_enabled(),
            'status': self.status.value,
            'status_reason': sorted(self.status_reason),
        }

    def as_json(self):
        """Return the Link as a JSON string."""
        return json.dumps(self.as_dict())

    @classmethod
    def from_dict(cls, link_dict):
        """Return a Link instance from python dictionary."""
        return cls(link_dict.get('endpoint_a'),
                   link_dict.get('endpoint_b'))


1		"""Module with all classes related to links.
2
3		Links are low level abstractions representing connections between two
4		interfaces.
5		"""
6		import json
7		import operator
8		from collections import OrderedDict, defaultdict
9		from functools import reduce
10		from threading import Lock
11		from typing import Union
12
13		from kytos.core.common import EntityStatus, GenericEntity
14		from kytos.core.exceptions import (KytosLinkCreationError,
15		KytosNoTagAvailableError)
16		from kytos.core.id import LinkID
17		from kytos.core.interface import Interface, TAGType
18		from kytos.core.tag_ranges import range_intersection
19
20
21		class Link(GenericEntity):
22		"""Define a link between two Endpoints."""
23
24		status_funcs = OrderedDict()
25		status_reason_funcs = OrderedDict()
26		_get_available_vlans_lock = defaultdict(Lock)
27
28		def __init__(self, endpoint_a, endpoint_b):
29		"""Create a Link instance and set its attributes.
30
31		Two kytos.core.interface.Interface are required as parameters.
32		"""
33		if endpoint_a is None:
34		raise KytosLinkCreationError("endpoint_a cannot be None")
35		if endpoint_b is None:
36		raise KytosLinkCreationError("endpoint_b cannot be None")
37		self._id = LinkID(endpoint_a.id, endpoint_b.id)
38		if self._id.interfaces[0] == endpoint_b.id:
39		self.endpoint_a: Interface = endpoint_b
40		self.endpoint_b: Interface = endpoint_a
41		else:
42		self.endpoint_a: Interface = endpoint_a
43		self.endpoint_b: Interface = endpoint_b
44
45		self.link_lock = Lock()
46		super().__init__()
47
48		def __hash__(self):
49		return hash(self.id)
50
51		def __repr__(self):
52		return f"Link({self.endpoint_a!r}, {self.endpoint_b!r}, {self.id})"
53
54		@classmethod
55		def register_status_func(cls, name: str, func):
56		"""Register status func given its name and a callable at setup time."""
57		cls.status_funcs[name] = func
58
59	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...
60		def status(self):
61		"""Return the current status of the Entity."""
62		state = super().status
63		if state == EntityStatus.DISABLED:
64		return state
65
66		for status_func in self.status_funcs.values():
67		if status_func(self) == EntityStatus.DOWN:
68		return EntityStatus.DOWN
69		return state
70
71		@classmethod
72		def register_status_reason_func(cls, name: str, func):
73		"""Register status reason func given its name
74		and a callable at setup time."""
75		cls.status_reason_funcs[name] = func
76
77		@property
78		def status_reason(self):
79		"""Return the reason behind the current status of the entity."""
80		return reduce(
81		operator.or_,
82		map(
83		lambda x: x(self),
84		self.status_reason_funcs.values()
85		),
86		super().status_reason
87		)
88
89		def is_enabled(self):
90		"""Override the is_enabled method.
91
92		We consider a link enabled when all the interfaces are enabled.
93
94		Returns:
95		boolean: True if both interfaces are enabled, False otherwise.
96
97		"""
98		return (self._enabled and self.endpoint_a.is_enabled() and
99		self.endpoint_b.is_enabled())
100
101		def is_active(self):
102		"""Override the is_active method.
103
104		We consider a link active whether all the interfaces are active.
105
106		Returns:
107		boolean: True if the interfaces are active, othewrise False.
108
109		"""
110		return (self._active and self.endpoint_a.is_active() and
111		self.endpoint_b.is_active())
112
113		def __eq__(self, other):
114		"""Check if two instances of Link are equal."""
115		return self.id == other.id
116
117		@property
118		def id(self): # pylint: disable=invalid-name
119		"""Return id from Link intance.
120
121		Returns:
122		string: link id.
123
124		"""
125		return self._id
126
127		def available_tags(self, tag_type: str = 'vlan') -> list[list[int]]:
128		"""Return the available tags for the link.
129
130		Based on the endpoint tags.
131		"""
132		tag_iterator = range_intersection(
133		self.endpoint_a.available_tags[tag_type],
134		self.endpoint_b.available_tags[tag_type],
135		)
136		available_tags = list(tag_iterator)
137		return available_tags
138
139		def is_tag_available(self, tag: int, tag_type: str = 'vlan'):
140		"""Check if a tag is available."""
141		return (self.endpoint_a.is_tag_available(tag, tag_type) and
142		self.endpoint_b.is_tag_available(tag, tag_type))
143
144		def get_next_available_tag(
145		self,
146		controller,
147		link_id: str,
148		take_last: bool = False,
149		tag_type: str = 'vlan',
150		try_avoid_value: int = None,
151		) -> int:
152		"""Return the next available tag if exists. By default this
153		method returns the smallest tag available. Apply options to
154		change behavior.
155		Options:
156		- take_last (bool): Choose the largest tag available.
157		- try_avoid_value (int): Avoid given tag if possible. Otherwise
158		return what is available.
159		"""
160		with self._get_available_vlans_lock[link_id]:
161		with self.endpoint_a._tag_lock:
162		with self.endpoint_b._tag_lock:
163		ava_tags_a = self.endpoint_a.available_tags[tag_type]
164		ava_tags_b = self.endpoint_b.available_tags[tag_type]
165		tags = range_intersection(ava_tags_a, ava_tags_b,
166		take_last)
167		try:
168		tag_range: list = next(tags)
169		if (try_avoid_value is not None and
170		tag_range[take_last] == try_avoid_value):
171		if (tag_range[take_last] !=
172		tag_range[not take_last]):
173		tag = tag_range[take_last]
174		tag += (+1) if not take_last else (-1)
175		else:
176		try:
177		tag = next(tags)[take_last]
178		except StopIteration:
179		tag = tag_range[take_last]
180		else:
181		tag = tag_range[take_last]
182
183		self.endpoint_a.use_tags(
184		controller, tag, use_lock=False, check_order=False
185		)
186		self.endpoint_b.use_tags(
187		controller, tag, use_lock=False, check_order=False
188		)
189		return tag
190		except StopIteration:
191		raise KytosNoTagAvailableError(self)
192
193		def make_tags_available(
194		self,
195		controller,
196		tags: Union[int, list[int], list[list[int]]],
197		link_id,
198		tag_type: str = 'vlan',
199		check_order: bool = True,
200		) -> tuple[list[list[int]], list[list[int]]]:
201		"""Add a specific tag in available_tags."""
202		with self._get_available_vlans_lock[link_id]:
203		with self.endpoint_a._tag_lock:
204		with self.endpoint_b._tag_lock:
205		conflict_a = self.endpoint_a.make_tags_available(
206		controller, tags, tag_type, use_lock=False,
207		check_order=check_order
208		)
209		conflict_b = self.endpoint_b.make_tags_available(
210		controller, tags, tag_type, use_lock=False,
211		check_order=check_order
212		)
213		return conflict_a, conflict_b
214
215		def available_vlans(self):
216		"""Get all available vlans from each interface in the link."""
217		vlans_a = self._get_available_vlans(self.endpoint_a)
218		vlans_b = self._get_available_vlans(self.endpoint_b)
219		return [vlan for vlan in vlans_a if vlan in vlans_b]
220
221		@staticmethod
222		def _get_available_vlans(endpoint):
223		"""Return all vlans from endpoint."""
224		vlans = endpoint.available_tags
225		return [vlan for vlan in vlans if vlan == TAGType.VLAN.value]
226
227		def as_dict(self):
228		"""Return the Link as a dictionary."""
229		return {
230		'id': self.id,
231		'endpoint_a': self.endpoint_a.as_dict(),
232		'endpoint_b': self.endpoint_b.as_dict(),
233		'metadata': self.get_metadata_as_dict(),
234		'active': self.is_active(),
235		'enabled': self.is_enabled(),
236		'status': self.status.value,
237		'status_reason': sorted(self.status_reason),
238		}
239
240		def as_json(self):
241		"""Return the Link as a JSON string."""
242		return json.dumps(self.as_dict())
243
244		@classmethod
245		def from_dict(cls, link_dict):
246		"""Return a Link instance from python dictionary."""
247		return cls(link_dict.get('endpoint_a'),
248		link_dict.get('endpoint_b'))
249

kytos-ng / kytos

kytos.core.link.Link.from_dict() A last analyzed 2025-06-18 19:57 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

kytos.core.link.Link.from_dict() A
last analyzed 2025-06-18 19:57 UTC