1 # -*- coding: utf-8 -*-
3 #######################################################################################
4 # This file is part of OSM RO module
6 # Copyright ETSI Contributors and Others.
8 # Licensed under the Apache License, Version 2.0 (the "License"); you may
9 # not use this file except in compliance with the License. You may obtain
10 # a copy of the License at
12 # http://www.apache.org/licenses/LICENSE-2.0
14 # Unless required by applicable law or agreed to in writing, software
15 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
16 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
17 # License for the specific language governing permissions and limitations
19 #######################################################################################
20 # This work has been performed in the context of the TeraFlow Project -
21 # funded by the European Commission under Grant number 101015857 through the
22 # Horizon 2020 program.
24 # - Lluis Gifre <lluis.gifre@cttc.es>
25 # - Ricard Vilalta <ricard.vilalta@cttc.es>
26 #######################################################################################
28 """This file contains the ServiceComposer class used by the Transport API (TAPI) WIM
29 connector to compose the services based on the service_endpoint_ids and their
32 from .exceptions
import (
33 WimTapiIncongruentDirectionality
,
34 WimTapiIncongruentEndPoints
,
35 WimTapiMissingMappingField
,
38 from .message_composers
import (
40 compose_requested_capacity
,
41 # compose_vlan_constraint,
45 class ServicesComposer
:
46 def __init__(self
, service_interface_points
) -> None:
47 self
.sips
= service_interface_points
50 # - a single service_endpoint item is created
51 # - the service_endpoint item contains with the 2 bidirectional SIPs
53 # - two service_endpoint items are created
54 # - each service_endpoint item containing a list of 2 unidirectional SIPs (in, out)
55 self
.services
= list()
57 # TODO: populate dynamically capacity of the connection
58 self
.requested_capacity
= compose_requested_capacity(1, unit
="GBPS")
60 self
.vlan_constraint
= None
61 # TODO: VLAN needs to be processed by connection point; by now deactivated
62 # if connection_point.get("service_endpoint_encapsulation_type") == "dot1q":
63 # encap_info = connection_point.get("service_endpoint_encapsulation_info", {})
64 # vlan_id = encap_info.get("vlan")
65 # if vlan_id is not None:
66 # vlan_constraint = compose_vlan_constraint(vlan_id)
68 def add_bidirectional(self
, service_endpoint_id
):
69 if len(self
.services
) == 0:
70 # assume bidirectional, SIP is service_endpoint_id
71 service_interface_point
= self
.sips
[service_endpoint_id
]
72 self
.services
.append([compose_endpoint(service_interface_point
)])
73 elif len(self
.services
) == 1:
74 # is bidirectional, SIP is service_endpoint_id
75 if len(self
.services
[0]) > 1:
76 # too much endpoints per service
77 raise WimTapiIncongruentEndPoints(self
.services
, service_endpoint_id
)
78 self
.services
[0].append(compose_endpoint(self
.sips
[service_endpoint_id
]))
80 raise WimTapiIncongruentDirectionality(self
.services
, service_endpoint_id
)
82 def add_unidirectional(self
, service_endpoint_id
, sip_input
, sip_output
):
83 if len(self
.services
) == 0:
84 # assume unidirectional
85 self
.services
.append([compose_endpoint(self
.sips
[sip_output
])]) # AZ
86 self
.services
.append([compose_endpoint(self
.sips
[sip_input
])]) # ZA
87 elif len(self
.services
) == 2:
90 if len(self
.services
[0]) > 1:
91 # too much endpoints per service
92 raise WimTapiIncongruentEndPoints(self
.services
[0], service_endpoint_id
)
93 self
.services
[0].append(compose_endpoint(self
.sips
[sip_input
])) # AZ
95 if len(self
.services
[1]) > 1:
96 # too much endpoints per service
97 raise WimTapiIncongruentEndPoints(self
.services
[1], service_endpoint_id
)
98 self
.services
[1].insert(0, compose_endpoint(self
.sips
[sip_output
])) # ZA
100 raise WimTapiIncongruentDirectionality(self
.services
, service_endpoint_id
)
102 def add_service_endpoint(self
, service_endpoint_id
, mapping
):
103 service_mapping_info
= mapping
.get("service_mapping_info", {})
106 len(service_mapping_info
) == 0
107 or "sip_input" not in service_mapping_info
108 or "sip_output" not in service_mapping_info
110 # bidirectional (no mapping or no sip_input or no sip_output)
111 if service_endpoint_id
not in self
.sips
:
112 raise WimTapiSipNotFound(service_endpoint_id
, self
.sips
)
113 self
.add_bidirectional(service_endpoint_id
)
116 # unidirectional, sip_input and sip_output provided in mapping
118 sip_input
= service_mapping_info
.get("sip_input")
119 if sip_input
is None:
120 raise WimTapiMissingMappingField(
121 mapping
, "service_mapping_info.sip_input"
124 if sip_input
not in self
.sips
:
125 raise WimTapiSipNotFound(sip_input
, self
.sips
)
127 sip_output
= service_mapping_info
.get("sip_output")
128 if sip_output
is None:
129 raise WimTapiMissingMappingField(
130 mapping
, "service_mapping_info.sip_output"
133 if sip_output
not in self
.sips
:
134 raise WimTapiSipNotFound(sip_output
, self
.sips
)
136 self
.add_unidirectional(service_endpoint_id
, sip_input
, sip_output
)
138 def is_bidirectional(self
):
139 return len(self
.services
) == 1
141 def dump(self
, logger
):
142 str_data
= "\n".join(
144 "services_composer {",
145 " services={:s}".format(str(self
.services
)),
146 " requested_capacity={:s}".format(str(self
.requested_capacity
)),
147 " vlan_constraint={:s}".format(str(self
.vlan_constraint
)),
151 logger
.debug(str_data
)