adapters/adtran_olt/codec/olt_state.py

# 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 OltState(object):
    """
    Class to wrap decode of olt-state container from the ADTRAN
    gpon-olt-hw.yang YANG model
    """

    def __init__(self, packet):
        self._packet = packet
        self._pons = None

    def __str__(self):
        return "OltState: {}".format(self.software_version)

    @property
    def software_version(self):
        """The software version of olt driver"""
        return self._packet.get('software-version', '')

    @property
    def pons(self):
        if self._pons is None:
            self._pons = OltState.Pon.decode(self._packet.get('pon', None))
        return self._pons

    #############################################################
    # Act like a container for simple access into PON list

    def __len__(self):
        return len(self.pons)

    def __getitem__(self, key):
        if not isinstance(key, int):
            raise TypeError('Key should be of type int')
        if key not in self.pons:
            raise KeyError("key '{}' not found".format(key))

        return self.pons[key]

    def __iter__(self):
        raise NotImplementedError('TODO: Not yet implemented')

    def __contains__(self, item):
        if not isinstance(item, int):
            raise TypeError('Item should be of type int')
        return item in self.pons

    # TODO: Look at generator support and if it is useful

    class Pon(object):
        """
        Provides decode of PON list from within
        """
        def __init__(self, packet):
            assert 'pon-id' in packet
            self._packet = packet
            self._onus = None
            self._gems = None

        def __str__(self):
            return "OltState.Pon: pon-id: {}".format(self.pon_id)

        @staticmethod
        def decode(pon_list):
            # log.debug('Decoding PON List:{}{}'.format(os.linesep,
            #                                           pprint.PrettyPrinter().pformat(pon_list)))
            pons = {}
            for pon_data in pon_list:
                pon = OltState.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 downstream_wavelength(self):
            """The wavelength, in nanometers, being used in the downstream direction"""
            return self._packet.get('downstream-wavelength', 0)

        @property
        def upstream_wavelength(self):
            """The wavelength, in nanometers, being used in the upstream direction"""
            return self._packet.get('upstream-wavelength', 0)

        @property
        def downstream_channel_id(self):
            """Downstream wavelength channel identifier associated with this PON."""
            return self._packet.get('downstream-channel-id', 0)

        @property
        def rx_packets(self):
            """Sum all of the RX Packets of GEM ports that are not base TCONT's"""
            return int(self._packet.get('rx-packets', 0))

        @property
        def tx_packets(self):
            """Sum all of the TX Packets of GEM ports that are not base TCONT's"""
            return int(self._packet.get('tx-packets', 0))

        @property
        def rx_bytes(self):
            """Sum all of the RX Octets of GEM ports that are not base TCONT's"""
            return int(self._packet.get('rx-bytes', 0))

        @property
        def tx_bytes(self):
            """Sum all of the TX Octets of GEM ports that are not base TCONT's"""
            return int(self._packet.get('tx-bytes', 0))

        @property
        def tx_bip_errors(self):
            """Sum the TX ONU bip errors to get TX BIP's per PON"""
            return int(self._packet.get('tx-bip-errors', 0))

        @property
        def wm_tuned_out_onus(self):
            """
            bit array indicates the list of tuned out ONU's that are in wavelength
            mobility protecting state.
                onu-bit-octects:
                  type binary { length "4 .. 1024"; }
                  description  each bit position indicates corresponding ONU's status
                               (true or false) whether that ONU's is in
                               wavelength mobility protecting state or not
                               For 128 ONTs per PON, the size of this
                               array will be 16. onu-bit-octects[0] and MSB bit in that byte
                               represents ONU 0 etc.
            """
            return self._packet.get('wm-tuned-out-onus', bytes(0))

        @property
        def ont_los(self):
            """List of configured ONTs that have been previously discovered and are in a los of signal state"""
            return self._packet.get('ont-los', [])

        @property
        def discovered_onu(self):
            """
            Immutable Set of each Optical Network Unit(ONU) that has been activated via discovery
                key/value: serial-number (string)
            """
            return frozenset([sn['serial-number'] for sn in self._packet.get('discovered-onu', [])
                              if 'serial-number' in sn and sn['serial-number'] != 'AAAAAAAAAAA='])

        @property
        def gems(self):
            """This list is not in the proposed BBF model, the stats are part of ietf-interfaces"""
            if self._gems is None:
                self._gems = OltState.Pon.Gem.decode(self._packet.get('gem', []))
            return self._gems

        @property
        def onus(self):
            """
            The map of each Optical Network Unit(ONU).  Key: ONU ID (int)
            """
            if self._onus is None:
                self._onus = OltState.Pon.Onu.decode(self._packet.get('onu', []))
            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 in packet'
                self._packet = packet

            def __str__(self):
                return "OltState.Pon.Onu: onu-id: {}".format(self.onu_id)

            @staticmethod
            def decode(onu_list):
                # log.debug('onus:{}{}'.format(os.linesep,
                #                              pprint.PrettyPrinter().pformat(onu_list)))
                onus = {}
                for onu_data in onu_list:
                    onu = OltState.Pon.Onu(onu_data)
                    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 oper_status(self):
                """The operational state of each ONU"""
                return self._packet.get('oper-status', 'unknown')

            @property
            def reported_password(self):
                """The password reported by the ONU (binary)"""
                return self._packet.get('reported-password', bytes(0))

            @property
            def rssi(self):
                """The received signal strength indication of the ONU"""
                return self._packet.get('rssi', -9999)

            @property
            def equalization_delay(self):
                """Equalization delay (bits)"""
                return self._packet.get('equalization-delay', 0)

            @property
            def fiber_length(self):
                """Distance to ONU"""
                return self._packet.get('fiber-length', 0)

        class Gem(object):
            """
            Provides decode of onu list for a PON port
            """
            def __init__(self, packet):
                assert 'onu-id' in packet, 'onu-id not found in packet'
                assert 'port-id' in packet, 'port-id not found in packet'
                assert 'alloc-id' in packet, 'alloc-id not found in packet'
                self._packet = packet

            def __str__(self):
                return "OltState.Pon.Gem: onu-id: {}, gem-id".\
                    format(self.onu_id, self.gem_id)

            @staticmethod
            def decode(gem_list):
                # log.debug('gems:{}{}'.format(os.linesep,
                #                              pprint.PrettyPrinter().pformat(gem_list)))
                gems = {}
                for gem_data in gem_list:
                    gem = OltState.Pon.Gem(gem_data)
                    assert gem.gem_id not in gems
                    gems[gem.gem_id] = gem

                return gems

            @property
            def onu_id(self):
                """The ID used to identify the ONU"""
                return self._packet['onu-id']

            @property
            def alloc_id(self):
                return self._packet['alloc-id']

            @property
            def gem_id(self):
                return self._packet['port-id']

            @property
            def tx_packets(self):
                return int(self._packet.get('tx-packets', 0))

            @property
            def tx_bytes(self):
                return int(self._packet.get('tx-bytes', 0))

            @property
            def rx_packets(self):
                return int(self._packet.get('rx-packets', 0))

            @property
            def rx_bytes(self):
                return int(self._packet.get('rx-bytes', 0))