VOL-1397: Adtran-OLT - Initial containerization commit
- Need to move VERSION to base directory
Change-Id: I9d62d0607a011ce642e379fd92b35ec48b300070
diff --git a/adapters/adtran_olt/codec/olt_config.py b/adapters/adtran_olt/codec/olt_config.py
new file mode 100644
index 0000000..473e2f6
--- /dev/null
+++ b/adapters/adtran_olt/codec/olt_config.py
@@ -0,0 +1,329 @@
+# Copyright 2017-present Adtran, Inc.
+#
+# 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.
+
+import structlog
+
+log = structlog.get_logger()
+
+
+class OltConfig(object):
+ """
+ Class to wrap decode of olt container (config) from the ADTRAN
+ gpon-olt-hw.yang YANG model
+ """
+ def __init__(self, packet):
+ self._packet = packet
+ self._pons = None
+
+ def __str__(self):
+ return "OltConfig: {}".format(self.olt_id)
+
+ @property
+ def olt_id(self):
+ """Unique OLT identifier"""
+ return self._packet.get('olt-id', '')
+
+ @property
+ def debug_output(self):
+ """least important level that will output everything"""
+ return self._packet.get('debug-output', 'warning')
+
+ @property
+ def pons(self):
+ if self._pons is None:
+ self._pons = OltConfig.Pon.decode(self._packet.get('pon', None))
+ return self._pons
+
+ class Pon(object):
+ """
+ Provides decode of PON list from within
+ """
+ def __init__(self, packet):
+ assert 'pon-id' in packet, 'pon-id not found'
+ self._packet = packet
+ self._onus = None
+
+ def __str__(self):
+ return "OltConfig.Pon: pon-id: {}".format(self.pon_id)
+
+ @staticmethod
+ def decode(pon_list):
+ pons = {}
+
+ if pon_list is not None:
+ for pon_data in pon_list:
+ pon = OltConfig.Pon(pon_data)
+ assert pon.pon_id not in pons
+ pons[pon.pon_id] = pon
+
+ return pons
+
+ @property
+ def pon_id(self):
+ """PON identifier"""
+ return self._packet['pon-id']
+
+ @property
+ def enabled(self):
+ """The desired state of the interface"""
+ return self._packet.get('enabled', False)
+
+ @property
+ def downstream_fec_enable(self):
+ """Enables downstream Forward Error Correction"""
+ return self._packet.get('downstream-fec-enable', False)
+
+ @property
+ def upstream_fec_enable(self):
+ """Enables upstream Forward Error Correction"""
+ return self._packet.get('upstream-fec-enable', False)
+
+ @property
+ def deployment_range(self):
+ """Maximum deployment distance (meters)"""
+ return self._packet.get('deployment-range', 25000)
+
+ @property
+ def onus(self):
+ if self._onus is None:
+ self._onus = OltConfig.Pon.Onu.decode(self._packet.get('onus', None))
+ return self._onus
+
+ class Onu(object):
+ """
+ Provides decode of onu list for a PON port
+ """
+ def __init__(self, packet):
+ assert 'onu-id' in packet, 'onu-id not found'
+ self._packet = packet
+ self._tconts = None
+ self._tconts_dict = None
+ self._gem_ports = None
+ self._gem_ports_dict = None
+
+ def __str__(self):
+ return "OltConfig.Pon.Onu: onu-id: {}".format(self.onu_id)
+
+ @staticmethod
+ def decode(onu_dict):
+ onus = {}
+
+ if onu_dict is not None:
+ if 'onu' in onu_dict:
+ for onu_data in onu_dict['onu']:
+ onu = OltConfig.Pon.Onu(onu_data)
+ assert onu.onu_id not in onus
+ onus[onu.onu_id] = onu
+ elif len(onu_dict) > 0 and 'onu-id' in onu_dict[0]:
+ onu = OltConfig.Pon.Onu(onu_dict[0])
+ assert onu.onu_id not in onus
+ onus[onu.onu_id] = onu
+
+ return onus
+
+ @property
+ def onu_id(self):
+ """The ID used to identify the ONU"""
+ return self._packet['onu-id']
+
+ @property
+ def serial_number_64(self):
+ """The serial number (base-64) is unique for each ONU"""
+ return self._packet.get('serial-number', '')
+
+ @property
+ def password(self):
+ """ONU Password"""
+ return self._packet.get('password', bytes(0))
+
+ @property
+ def enable(self):
+ """If true, places the ONU in service"""
+ return self._packet.get('enable', False)
+
+ @property
+ def tconts(self):
+ if self._tconts is None:
+ self._tconts = OltConfig.Pon.Onu.TCont.decode(self._packet.get('t-conts', None))
+ return self._tconts
+
+ @property
+ def tconts_dict(self): # TODO: Remove if not used
+ if self._tconts_dict is None:
+ self._tconts_dict = {tcont.alloc_id: tcont for tcont in self.tconts}
+ return self._tconts_dict
+
+ @property
+ def gem_ports(self):
+ if self._gem_ports is None:
+ self._gem_ports = OltConfig.Pon.Onu.GemPort.decode(self._packet.get('gem-ports', None))
+ return self._gem_ports
+
+ @property
+ def gem_ports_dict(self): # TODO: Remove if not used
+ if self._gem_ports_dict is None:
+ self._gem_ports_dict = {gem.gem_id: gem for gem in self.gem_ports}
+ return self._gem_ports_dict
+
+ class TCont(object):
+ """
+ Provides decode of onu list for the T-CONT container
+ """
+ def __init__(self, packet):
+ assert 'alloc-id' in packet, 'alloc-id not found'
+ self._packet = packet
+ self._traffic_descriptor = None
+ self._best_effort = None
+
+ def __str__(self):
+ return "OltConfig.Pon.Onu.TCont: alloc-id: {}".format(self.alloc_id)
+
+ @staticmethod
+ def decode(tcont_container):
+ tconts = {}
+
+ if tcont_container is not None:
+ for tcont_data in tcont_container.get('t-cont', []):
+ tcont = OltConfig.Pon.Onu.TCont(tcont_data)
+ assert tcont.alloc_id not in tconts
+ tconts[tcont.alloc_id] = tcont
+
+ return tconts
+
+ @property
+ def alloc_id(self):
+ """The ID used to identify the T-CONT"""
+ return self._packet['alloc-id']
+
+ @property
+ def traffic_descriptor(self):
+ """
+ Each Alloc-ID is provisioned with a traffic descriptor that specifies
+ the three bandwidth component parameters: fixed bandwidth, assured
+ bandwidth, and maximum bandwidth, as well as the ternary eligibility
+ indicator for additional bandwidth assignment
+ """
+ if self._traffic_descriptor is None and 'traffic-descriptor' in self._packet:
+ self._traffic_descriptor = OltConfig.Pon.Onu.TCont.\
+ TrafficDescriptor(self._packet['traffic-descriptor'])
+ return self._traffic_descriptor
+
+ class TrafficDescriptor(object):
+ def __init__(self, packet):
+ self._packet = packet
+
+ def __str__(self):
+ return "OltConfig.Pon.Onu.TCont.TrafficDescriptor: {}/{}/{}".\
+ format(self.fixed_bandwidth, self.assured_bandwidth,
+ self.maximum_bandwidth)
+
+ @property
+ def fixed_bandwidth(self):
+ try:
+ return int(self._packet.get('fixed-bandwidth', 0))
+ except:
+ return 0
+
+ @property
+ def assured_bandwidth(self):
+ try:
+ return int(self._packet.get('assured-bandwidth', 0))
+ except:
+ return 0
+
+ @property
+ def maximum_bandwidth(self):
+ try:
+ return int(self._packet.get('maximum-bandwidth', 0))
+ except:
+ return 0
+
+ @property
+ def additional_bandwidth_eligibility(self):
+ return self._packet.get('additional-bandwidth-eligibility', 'none')
+
+ @property
+ def best_effort(self):
+ if self._best_effort is None:
+ self._best_effort = OltConfig.Pon.Onu.TCont.BestEffort.decode(
+ self._packet.get('best-effort', None))
+ return self._best_effort
+
+ class BestEffort(object):
+ def __init__(self, packet):
+ self._packet = packet
+
+ def __str__(self):
+ return "OltConfig.Pon.Onu.TCont.BestEffort: {}".format(self.bandwidth)
+
+ @property
+ def bandwidth(self):
+ return self._packet['bandwidth']
+
+ @property
+ def priority(self):
+ return self._packet['priority']
+
+ @property
+ def weight(self):
+ return self._packet['weight']
+
+ class GemPort(object):
+ """
+ Provides decode of onu list for the gem-ports container
+ """
+ def __init__(self, packet):
+ assert 'port-id' in packet, 'port-id not found'
+ self._packet = packet
+
+ def __str__(self):
+ return "OltConfig.Pon.Onu.GemPort: port-id: {}/{}".\
+ format(self.port_id, self.alloc_id)
+
+ @staticmethod
+ def decode(gem_port_container):
+ gem_ports = {}
+
+ if gem_port_container is not None:
+ for gem_port_data in gem_port_container.get('gem-port', []):
+ gem_port = OltConfig.Pon.Onu.GemPort(gem_port_data)
+ assert gem_port.port_id not in gem_ports
+ gem_ports[gem_port.port_id] = gem_port
+
+ return gem_ports
+
+ @property
+ def port_id(self):
+ """The ID used to identify the GEM Port"""
+ return self._packet['port-id']
+
+ @property
+ def gem_id(self):
+ """The ID used to identify the GEM Port"""
+ return self.port_id
+
+ @property
+ def alloc_id(self):
+ """The Alloc-ID of the T-CONT to which this GEM port is mapped"""
+ return self._packet['alloc-id']
+
+ @property
+ def omci_transport(self):
+ """If true, this GEM port is used to transport the OMCI virtual connection"""
+ return self._packet.get('omci-transport', False)
+
+ @property
+ def encryption(self):
+ """If true, enable encryption using the advanced encryption standard(AES)"""
+ return self._packet.get('encryption', False)