sonic-buildimage/platform/mellanox/mlnx-platform-api/sonic_platform/chassis.py

#
# Copyright (c) 2019-2023 NVIDIA CORPORATION & AFFILIATES.
# Apache-2.0
#
# 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.
#
#############################################################################
# Mellanox
#
# Module contains an implementation of SONiC Platform Base API and
# provides the Chassis information which are available in the platform
#
#############################################################################

try:
    from sonic_platform_base.chassis_base import ChassisBase
    from sonic_py_common.logger import Logger
    import os
    from functools import reduce
    from .utils import extract_RJ45_ports_index
    from . import utils
    from .device_data import DeviceDataManager
    import re
    import queue
    import threading
    import time
    from sonic_platform import modules_mgmt
except ImportError as e:
    raise ImportError (str(e) + "- required module not found")


RJ45_TYPE = "RJ45"

DMI_FILE = '/sys/firmware/dmi/entries/2-0/raw'
DMI_HEADER_LEN = 15
DMI_PRODUCT_NAME = "Product Name"
DMI_MANUFACTURER = "Manufacturer"
DMI_VERSION = "Version"
DMI_SERIAL = "Serial Number"
DMI_ASSET_TAG = "Asset Tag"
DMI_LOC = "Location In Chassis"
DMI_TABLE_MAP = {
                    DMI_PRODUCT_NAME: 0,
                    DMI_MANUFACTURER: 1,
                    DMI_VERSION: 2,
                    DMI_SERIAL: 3,
                    DMI_ASSET_TAG: 4,
                    DMI_LOC: 5
                }

HWMGMT_SYSTEM_ROOT = '/var/run/hw-management/system/'

#reboot cause related definitions
REBOOT_CAUSE_ROOT = HWMGMT_SYSTEM_ROOT
REBOOT_CAUSE_MAX_WAIT_TIME = 45
REBOOT_CAUSE_CHECK_INTERVAL = 5
REBOOT_CAUSE_READY_FILE = '/run/hw-management/config/reset_attr_ready'

REBOOT_TYPE_KEXEC_FILE = "/proc/cmdline"
REBOOT_TYPE_KEXEC_PATTERN_WARM = ".*SONIC_BOOT_TYPE=(warm|fastfast).*"
REBOOT_TYPE_KEXEC_PATTERN_FAST = ".*SONIC_BOOT_TYPE=(fast|fast-reboot).*"

# Global logger class instance
logger = Logger()

class Chassis(ChassisBase):
    """Platform-specific Chassis class"""

    # System status LED
    _led = None

    # System UID LED
    _led_uid = None

    chassis_instance = None

    def __init__(self):
        super(Chassis, self).__init__()

        # Initialize DMI data
        self.dmi_data = None

        # move the initialization of each components to their dedicated initializer
        # which will be called from platform
        #
        # Multiple scenarios need to be taken into consideration regarding the SFP modules initialization.
        # - Platform daemons
        #   - Can access multiple or all SFP modules
        # - sfputil
        #   - Sometimes can access only one SFP module
        #   - Call get_sfp to get one SFP module object
        #
        # We should initialize all SFP modules only if it is necessary because initializing SFP module is time-consuming.
        # This means,
        # - If get_sfp is called,
        #    - If the _sfp_list isn't initialized, initialize it first.
        #    - Only the SFP module being required should be initialized.
        # - If get_all_sfps is called,
        #    - If the _sfp_list isn't initialized, initialize it first.
        #    - All SFP modules need to be initialized.
        #      But the SFP modules that have already been initialized should not be initialized for the second time.
        #      This can caused by get_sfp being called before.
        #
        # Due to the complexity of SFP modules initialization, we have to introduce two initialized flags for SFP modules
        # - sfp_module_partial_initialized:
        #    - False: The _sfp_list is [] (SFP stuff has never been accessed)
        #    - True: The _sfp_list is a list whose length is number of SFP modules supported by the platform
        # - sfp_module_full_initialized:
        #    - False: All SFP modules have not been created
        #    - True: All SFP modules have been created
        #
        self.sfp_initialized_count = 0
        self.sfp_event = None
        self.reboot_cause_initialized = False

        self.sfp_module = None
        self.sfp_lock = threading.Lock()

        # Build the RJ45 port list from platform.json and hwsku.json
        self._RJ45_port_inited = False
        self._RJ45_port_list = None

        Chassis.chassis_instance = self

        self.modules_mgmt_thread = threading.Thread()
        self.modules_changes_queue = queue.Queue()
        self.modules_mgmt_task_stopping_event = threading.Event()

        logger.log_info("Chassis loaded successfully")

    def __del__(self):
        if self.sfp_event:
            self.sfp_event.deinitialize()

        if self._sfp_list:
            if self.sfp_module.SFP.shared_sdk_handle:
                self.sfp_module.deinitialize_sdk_handle(self.sfp_module.SFP.shared_sdk_handle)

    @property
    def RJ45_port_list(self):
        if not self._RJ45_port_inited:
            self._RJ45_port_list = extract_RJ45_ports_index()
            self._RJ45_port_inited = True
        return self._RJ45_port_list

    ##############################################
    # PSU methods
    ##############################################

    def initialize_psu(self):
        if not self._psu_list:
            from .psu import Psu, FixedPsu
            psu_count = DeviceDataManager.get_psu_count()
            hot_swapable = DeviceDataManager.is_psu_hotswapable()

            # Initialize PSU list
            for index in range(psu_count):
                if hot_swapable:
                    psu = Psu(index)
                else:
                    psu = FixedPsu(index)
                self._psu_list.append(psu)

    def get_num_psus(self):
        """
        Retrieves the number of power supply units available on this chassis

        Returns:
            An integer, the number of power supply units available on this
            chassis
        """
        self.initialize_psu()
        return len(self._psu_list)

    def get_all_psus(self):
        """
        Retrieves all power supply units available on this chassis

        Returns:
            A list of objects derived from PsuBase representing all power
            supply units available on this chassis
        """
        self.initialize_psu()
        return self._psu_list

    def get_psu(self, index):
        """
        Retrieves power supply unit represented by (0-based) index <index>

        Args:
            index: An integer, the index (0-based) of the power supply unit to
            retrieve

        Returns:
            An object dervied from PsuBase representing the specified power
            supply unit
        """
        self.initialize_psu()
        return super(Chassis, self).get_psu(index)

    ##############################################
    # Fan methods
    ##############################################

    def initialize_fan(self):
        if not self._fan_drawer_list:
            from .fan import Fan
            from .fan_drawer import RealDrawer, VirtualDrawer

            hot_swapable = DeviceDataManager.is_fan_hotswapable()
            drawer_num = DeviceDataManager.get_fan_drawer_count()
            fan_num = DeviceDataManager.get_fan_count()
            fan_num_per_drawer = fan_num // drawer_num
            drawer_ctor = RealDrawer if hot_swapable else VirtualDrawer
            fan_index = 0
            for drawer_index in range(drawer_num):
                drawer = drawer_ctor(drawer_index)
                self._fan_drawer_list.append(drawer)
                for index in range(fan_num_per_drawer):
                    fan = Fan(fan_index, drawer, index + 1)
                    fan_index += 1
                    drawer._fan_list.append(fan)

    def get_num_fan_drawers(self):
        """
        Retrieves the number of fan drawers available on this chassis

        Returns:
            An integer, the number of fan drawers available on this chassis
        """
        return DeviceDataManager.get_fan_drawer_count()

    def get_all_fan_drawers(self):
        """
        Retrieves all fan drawers available on this chassis

        Returns:
            A list of objects derived from FanDrawerBase representing all fan
            drawers available on this chassis
        """
        self.initialize_fan()
        return self._fan_drawer_list

    def get_fan_drawer(self, index):
        """
        Retrieves fan drawers represented by (0-based) index <index>

        Args:
            index: An integer, the index (0-based) of the fan drawer to
            retrieve

        Returns:
            An object dervied from FanDrawerBase representing the specified fan
            drawer
        """
        self.initialize_fan()
        return super(Chassis, self).get_fan_drawer(index)

    ##############################################
    # SFP methods
    ##############################################

    def _import_sfp_module(self):
        if not self.sfp_module:
            from . import sfp as sfp_module
            self.sfp_module = sfp_module
        return self.sfp_module

    def initialize_single_sfp(self, index):
        sfp_count = self.get_num_sfps()
        # Use double checked locking mechanism for:
        #     1. protect shared resource self._sfp_list
        #     2. performance (avoid locking every time)
        if index < sfp_count:
            if not self._sfp_list or not self._sfp_list[index]:
                with self.sfp_lock:
                    if not self._sfp_list:
                        self._sfp_list = [None] * sfp_count

                    if not self._sfp_list[index]:
                        sfp_module = self._import_sfp_module()
                        if self.RJ45_port_list and index in self.RJ45_port_list:
                            self._sfp_list[index] = sfp_module.RJ45Port(index)
                        else:
                            self._sfp_list[index] = sfp_module.SFP(index)
                        self.sfp_initialized_count += 1

    def initialize_sfp(self):
        sfp_count = self.get_num_sfps()
        # Use double checked locking mechanism for:
        #     1. protect shared resource self._sfp_list
        #     2. performance (avoid locking every time)
        if sfp_count != self.sfp_initialized_count:
            with self.sfp_lock:
                if sfp_count != self.sfp_initialized_count:
                    if not self._sfp_list:
                        sfp_module = self._import_sfp_module()
                        for index in range(sfp_count):
                            if self.RJ45_port_list and index in self.RJ45_port_list:
                                sfp_object = sfp_module.RJ45Port(index)
                            else:
                                sfp_object = sfp_module.SFP(index)
                            self._sfp_list.append(sfp_object)
                        self.sfp_initialized_count = sfp_count
                    elif self.sfp_initialized_count != len(self._sfp_list):
                        sfp_module = self._import_sfp_module()
                        for index in range(len(self._sfp_list)):
                            if self._sfp_list[index] is None:
                                if self.RJ45_port_list and index in self.RJ45_port_list:
                                    self._sfp_list[index] = sfp_module.RJ45Port(index)
                                else:
                                    self._sfp_list[index] = sfp_module.SFP(index)
                        self.sfp_initialized_count = len(self._sfp_list)

    def get_num_sfps(self):
        """
        Retrieves the number of sfps available on this chassis

        Returns:
            An integer, the number of sfps available on this chassis
        """
        return DeviceDataManager.get_sfp_count()

    def get_all_sfps(self):
        """
        Retrieves all sfps available on this chassis

        Returns:
            A list of objects derived from SfpBase representing all sfps
            available on this chassis
        """
        self.initialize_sfp()
        return self._sfp_list

    def get_sfp(self, index):
        """
        Retrieves sfp represented by (1-based) index <index>

        Args:
            index: An integer, the index (1-based) of the sfp to retrieve.
                   The index should be the sequence of a physical port in a chassis,
                   starting from 1.
                   For example, 1 for Ethernet0, 2 for Ethernet4 and so on.

        Returns:
            An object dervied from SfpBase representing the specified sfp
        """
        index = index - 1
        self.initialize_single_sfp(index)
        return super(Chassis, self).get_sfp(index)

    def get_port_or_cage_type(self, index):
        """
        Retrieves sfp port or cage type corresponding to physical port <index>

        Args:
            index: An integer (>=0), the index of the sfp to retrieve.
                   The index should correspond to the physical port in a chassis.
                   For example:-
                   1 for Ethernet0, 2 for Ethernet4 and so on for one platform.
                   0 for Ethernet0, 1 for Ethernet4 and so on for another platform.

        Returns:
            The masks of all types of port or cage that can be supported on the port
            Types are defined in sfp_base.py
            Eg.
                Both SFP and SFP+ are supported on the port, the return value should be 0x0a
                which is 0x02 | 0x08
        """
        index = index - 1
        if self.RJ45_port_list and index in self.RJ45_port_list:
            from sonic_platform_base.sfp_base import SfpBase
            return SfpBase.SFP_PORT_TYPE_BIT_RJ45
        raise NotImplementedError

    def get_change_event(self, timeout=0):
        """
        Returns a nested dictionary containing all devices which have
        experienced a change at chassis level

        Args:
            timeout: Timeout in milliseconds (optional). If timeout == 0,
                this method will block until a change is detected.

        Returns:
            (bool, dict):
                - True if call successful, False if not; - Deprecated, will always return True
                - A nested dictionary where key is a device type,
                  value is a dictionary with key:value pairs in the format of
                  {'device_id':'device_event'},
                  where device_id is the device ID for this device and
                        device_event,
                             status='1' represents device inserted,
                             status='0' represents device removed.
                  Ex. {'fan':{'0':'0', '2':'1'}, 'sfp':{'11':'0'}}
                      indicates that fan 0 has been removed, fan 2
                      has been inserted and sfp 11 has been removed.
        """
        if not self.modules_mgmt_thread.is_alive():
            # open new SFP change events thread
            self.modules_mgmt_thread = modules_mgmt.ModulesMgmtTask(q=self.modules_changes_queue
                                                    , main_thread_stop_event = self.modules_mgmt_task_stopping_event)
            # Set the thread as daemon so when pmon/xcvrd are shutting down, modules_mgmt will shut down immedietly.
            self.modules_mgmt_thread.daemon = True
            self.modules_mgmt_thread.start()
        self.initialize_sfp()
        wait_for_ever = (timeout == 0)
        # poll timeout should be no more than 1000ms to ensure fast shutdown flow
        timeout = 1000.0 if timeout >= 1000 else float(timeout)
        port_dict = {}
        error_dict = {}
        begin = time.time()
        i = 0
        while True:
            try:
                logger.log_info(f'get_change_event() trying to get changes from queue on iteration {i}')
                port_dict = self.modules_changes_queue.get(timeout=timeout / 1000)
                logger.log_info(f'get_change_event() iteration {i} port_dict: {port_dict}')
            except queue.Empty:
                logger.log_info(f"failed to get item from modules changes queue on itertaion {i}")

            if port_dict:
                self.reinit_sfps(port_dict)
                result_dict = {'sfp': port_dict}
                result_dict['sfp_error'] = error_dict
                return True, result_dict
            else:
                if not wait_for_ever:
                    elapse = time.time() - begin
                    logger.log_info(f"get_change_event: wait_for_ever {wait_for_ever} elapse {elapse} iteartion {i}")
                    if elapse * 1000 >= timeout:
                        logger.log_info(f"elapse {elapse} > timeout {timeout} iteartion {i} returning empty dict")
                        return True, {'sfp': {}}
            i += 1

    def reinit_sfps(self, port_dict):
        """
        Re-initialize SFP if there is any newly inserted SFPs
        :param port_dict: SFP event data
        :return:
        """
        from . import sfp
        for index, status in port_dict.items():
            if status == sfp.SFP_STATUS_INSERTED:
                try:
                    self._sfp_list[int(index) - 1].reinit()
                except Exception as e:
                    logger.log_error("Fail to re-initialize SFP {} - {}".format(index, repr(e)))

    def _show_capabilities(self):
        """
        This function is for debug purpose
        Some features require a xSFP module to support some capabilities but it's unrealistic to
        check those modules one by one.
        So this function is introduce to show some capabilities of all xSFP modules mounted on the device.
        """
        self.initialize_sfp()
        for s in self._sfp_list:
            try:
                print("index {} tx disable {} dom {} calibration {} temp {} volt {} power (tx {} rx {})".format(s.index,
                    s.dom_tx_disable_supported,
                    s.dom_supported,
                    s.calibration,
                    s.dom_temp_supported,
                    s.dom_volt_supported,
                    s.dom_rx_power_supported,
                    s.dom_tx_power_supported
                    ))
            except:
                print("fail to retrieve capabilities for module index {}".format(s.index))

    ##############################################
    # THERMAL methods
    ##############################################

    def initialize_thermals(self):
        if not self._thermal_list:
            from .thermal import initialize_chassis_thermals
            # Initialize thermals
            self._thermal_list = initialize_chassis_thermals()

    def get_num_thermals(self):
        """
        Retrieves the number of thermals available on this chassis

        Returns:
            An integer, the number of thermals available on this chassis
        """
        self.initialize_thermals()
        return len(self._thermal_list)

    def get_all_thermals(self):
        """
        Retrieves all thermals available on this chassis

        Returns:
            A list of objects derived from ThermalBase representing all thermals
            available on this chassis
        """
        self.initialize_thermals()
        return self._thermal_list

    def get_thermal(self, index):
        """
        Retrieves thermal unit represented by (0-based) index <index>

        Args:
            index: An integer, the index (0-based) of the thermal to
            retrieve

        Returns:
            An object dervied from ThermalBase representing the specified thermal
        """
        self.initialize_thermals()
        return super(Chassis, self).get_thermal(index)

    ##############################################
    # EEPROM methods
    ##############################################

    def initialize_eeprom(self):
        if not self._eeprom:
            from .eeprom import Eeprom
            # Initialize EEPROM
            self._eeprom = Eeprom()

    def get_eeprom(self):
        """
        Retreives eeprom device on this chassis

        Returns:
            An object derived from WatchdogBase representing the hardware
            eeprom device
        """
        self.initialize_eeprom()
        return self._eeprom

    def get_name(self):
        """
        Retrieves the name of the device

        Returns:
            string: The name of the device
        """
        self.initialize_eeprom()
        return self._eeprom.get_product_name()

    def get_model(self):
        """
        Retrieves the model number (or part number) of the device

        Returns:
            string: Model/part number of device
        """
        self.initialize_eeprom()
        return self._eeprom.get_part_number()

    def get_base_mac(self):
        """
        Retrieves the base MAC address for the chassis

        Returns:
            A string containing the MAC address in the format
            'XX:XX:XX:XX:XX:XX'
        """
        self.initialize_eeprom()
        return self._eeprom.get_base_mac()

    def get_serial(self):
        """
        Retrieves the hardware serial number for the chassis

        Returns:
            A string containing the hardware serial number for this chassis.
        """
        self.initialize_eeprom()
        return self._eeprom.get_serial_number()

    def get_system_eeprom_info(self):
        """
        Retrieves the full content of system EEPROM information for the chassis

        Returns:
            A dictionary where keys are the type code defined in
            OCP ONIE TlvInfo EEPROM format and values are their corresponding
            values.
        """
        self.initialize_eeprom()
        return self._eeprom.get_system_eeprom_info()

    ##############################################
    # Component methods
    ##############################################

    def initialize_components(self):
        if not utils.is_host():
            return
        if not self._component_list:
            # Initialize component list
            from .component import ComponentONIE, ComponentSSD, ComponentBIOS, ComponentCPLD
            self._component_list.append(ComponentONIE())
            self._component_list.append(ComponentSSD())
            self._component_list.append(DeviceDataManager.get_bios_component())
            self._component_list.extend(DeviceDataManager.get_cpld_component_list())

    def get_num_components(self):
        """
        Retrieves the number of components available on this chassis

        Returns:
            An integer, the number of components available on this chassis
        """
        self.initialize_components()
        return len(self._component_list)

    def get_all_components(self):
        """
        Retrieves all components available on this chassis

        Returns:
            A list of objects derived from ComponentBase representing all components
            available on this chassis
        """
        self.initialize_components()
        return self._component_list

    def get_component(self, index):
        """
        Retrieves component represented by (0-based) index <index>

        Args:
            index: An integer, the index (0-based) of the component to retrieve

        Returns:
            An object dervied from ComponentBase representing the specified component
        """
        self.initialize_components()
        return super(Chassis, self).get_component(index)

    ##############################################
    # System LED methods
    ##############################################

    def initizalize_system_led(self):
        if not Chassis._led:
            from .led import SystemLed, \
                SystemUidLed
            Chassis._led = SystemLed()
            Chassis._led_uid = SystemUidLed()

    def set_status_led(self, color):
        """
        Sets the state of the system LED

        Args:
            color: A string representing the color with which to set the
                   system LED

        Returns:
            bool: True if system LED state is set successfully, False if not
        """
        self.initizalize_system_led()
        return False if not Chassis._led else Chassis._led.set_status(color)

    def get_status_led(self):
        """
        Gets the state of the system LED

        Returns:
            A string, one of the valid LED color strings which could be vendor
            specified.
        """
        self.initizalize_system_led()
        return None if not Chassis._led else Chassis._led.get_status()

    def set_uid_led(self, color):
        """
        Sets the state of the system UID LED

        Args:
            color: A string representing the color with which to set the
                   system UID LED

        Returns:
            bool: True if system LED state is set successfully, False if not
        """
        self.initizalize_system_led()
        return False if not Chassis._led_uid else Chassis._led_uid.set_status(color)

    def get_uid_led(self):
        """
        Gets the state of the system UID LED

        Returns:
            A string, one of the valid LED color strings which could be vendor
            specified.
        """
        self.initizalize_system_led()
        return None if not Chassis._led_uid else Chassis._led_uid.get_status()

    def get_watchdog(self):
        """
        Retrieves hardware watchdog device on this chassis

        Returns:
            An object derived from WatchdogBase representing the hardware
            watchdog device

        Note:
            We overload this method to ensure that watchdog is only initialized
            when it is referenced. Currently, only one daemon can open the watchdog.
            To initialize watchdog in the constructor causes multiple daemon
            try opening watchdog when loading and constructing a chassis object
            and fail. By doing so we can eliminate that risk.
        """
        try:
            if self._watchdog is None:
                from .watchdog import get_watchdog
                self._watchdog = get_watchdog()
        except Exception as e:
            logger.log_info("Fail to load watchdog due to {}".format(repr(e)))

        return self._watchdog


    def get_revision(self):
        """
        Retrieves the hardware revision of the device

        Returns:
            string: Revision value of device
        """
        if self.dmi_data is None:
            self.dmi_data = self._parse_dmi(DMI_FILE)

        return self.dmi_data.get(DMI_VERSION, "N/A")

    def _parse_dmi(self, filename):
        """
        Read DMI data chassis data and returns a dictionary of values

        Returns:
            A dictionary containing the dmi table of the switch chassis info
        """
        result = {}
        try:
            if not os.access(filename, os.R_OK):
                return result

            with open(filename, "rb") as fileobj:
                data = fileobj.read()

            body = data[DMI_HEADER_LEN:]
            records = body.split(b'\x00')

            for k, v in DMI_TABLE_MAP.items():
                result[k] = records[v].decode("utf-8")

        except Exception as e:
            logger.log_error("Fail to decode DMI {} due to {}".format(filename, repr(e)))

        return result

    def _verify_reboot_cause(self, filename):
        '''
        Open and read the reboot cause file in
        /var/run/hwmanagement/system (which is defined as REBOOT_CAUSE_ROOT)
        If a reboot cause file doesn't exists, returns '0'.
        '''
        return bool(utils.read_int_from_file(os.path.join(REBOOT_CAUSE_ROOT, filename), log_func=None))

    def initialize_reboot_cause(self):
        self.reboot_major_cause_dict = {
            'reset_main_pwr_fail'       :   self.REBOOT_CAUSE_POWER_LOSS,
            'reset_aux_pwr_or_ref'      :   self.REBOOT_CAUSE_POWER_LOSS,
            'reset_comex_pwr_fail'      :   self.REBOOT_CAUSE_POWER_LOSS,
            'reset_asic_thermal'        :   self.REBOOT_CAUSE_THERMAL_OVERLOAD_ASIC,
            'reset_comex_thermal'       :   self.REBOOT_CAUSE_THERMAL_OVERLOAD_CPU,
            'reset_hotswap_or_wd'       :   self.REBOOT_CAUSE_WATCHDOG,
            'reset_comex_wd'            :   self.REBOOT_CAUSE_WATCHDOG,
            'reset_swb_wd'              :   self.REBOOT_CAUSE_WATCHDOG,
            'reset_sff_wd'              :   self.REBOOT_CAUSE_WATCHDOG,
            'reset_hotswap_or_halt'     :   self.REBOOT_CAUSE_HARDWARE_OTHER,
            'reset_voltmon_upgrade_fail':   self.REBOOT_CAUSE_HARDWARE_OTHER,
            'reset_reload_bios'         :   self.REBOOT_CAUSE_HARDWARE_BIOS,
            'reset_fw_reset'            :   self.REBOOT_CAUSE_HARDWARE_RESET_FROM_ASIC,
            'reset_from_asic'           :   self.REBOOT_CAUSE_HARDWARE_RESET_FROM_ASIC,
            'reset_long_pb'             :   self.REBOOT_CAUSE_HARDWARE_BUTTON,
            'reset_short_pb'            :   self.REBOOT_CAUSE_HARDWARE_BUTTON
        }
        self.reboot_minor_cause_dict = {}
        self.reboot_by_software = 'reset_sw_reset'
        self.reboot_cause_initialized = True

    def _parse_warmfast_reboot_from_proc_cmdline(self):
        if os.path.isfile(REBOOT_TYPE_KEXEC_FILE):
            with open(REBOOT_TYPE_KEXEC_FILE) as cause_file:
                cause_file_kexec = cause_file.readline()
            m = re.search(REBOOT_TYPE_KEXEC_PATTERN_WARM, cause_file_kexec)
            if m and m.group(1):
                return 'warm-reboot'
            m = re.search(REBOOT_TYPE_KEXEC_PATTERN_FAST, cause_file_kexec)
            if m and m.group(1):
                return 'fast-reboot'
        return None

    def _wait_reboot_cause_ready(self):
        max_wait_time = REBOOT_CAUSE_MAX_WAIT_TIME
        while max_wait_time > 0:
            if utils.read_int_from_file(REBOOT_CAUSE_READY_FILE, log_func=None) == 1:
                return True
            time.sleep(REBOOT_CAUSE_CHECK_INTERVAL)
            max_wait_time -= REBOOT_CAUSE_CHECK_INTERVAL

        return False

    def get_reboot_cause(self):
        """
        Retrieves the cause of the previous reboot

        Returns:
            A tuple (string, string) where the first element is a string
            containing the cause of the previous reboot. This string must be
            one of the predefined strings in this class. If the first string
            is "REBOOT_CAUSE_HARDWARE_OTHER", the second string can be used
            to pass a description of the reboot cause.
        """
        #read reboot causes files in the following order

        # To avoid the leftover hardware reboot cause confusing the reboot cause determine service
        # Skip the hardware reboot cause check if warm/fast reboot cause found from cmdline
        if utils.is_host():
            reboot_cause = self._parse_warmfast_reboot_from_proc_cmdline()
            if reboot_cause:
                return self.REBOOT_CAUSE_NON_HARDWARE, ''

        if not self._wait_reboot_cause_ready():
            logger.log_error("Hardware reboot cause is not ready")
            return self.REBOOT_CAUSE_NON_HARDWARE, ''

        if not self.reboot_cause_initialized:
            self.initialize_reboot_cause()

        for reset_file, reset_cause in self.reboot_major_cause_dict.items():
            if self._verify_reboot_cause(reset_file):
                return reset_cause, ''

        for reset_file, reset_cause in self.reboot_minor_cause_dict.items():
            if self._verify_reboot_cause(reset_file):
                return self.REBOOT_CAUSE_HARDWARE_OTHER, reset_cause

        if self._verify_reboot_cause(self.reboot_by_software):
            logger.log_info("Hardware reboot cause: the system was rebooted due to software requesting")
        else:
            logger.log_info("Hardware reboot cause: no hardware reboot cause found")

        return self.REBOOT_CAUSE_NON_HARDWARE, ''

    def get_thermal_manager(self):
        from .thermal_manager import ThermalManager
        return ThermalManager

    def get_position_in_parent(self):
        """
		Retrieves 1-based relative physical position in parent device. If the agent cannot determine the parent-relative position
        for some reason, or if the associated value of entPhysicalContainedIn is '0', then the value '-1' is returned
		Returns:
		    integer: The 1-based relative physical position in parent device or -1 if cannot determine the position
		"""
        return -1

    def is_replaceable(self):
        """
        Indicate whether this device is replaceable.
        Returns:
            bool: True if it is replaceable.
        """
        return False


class ModularChassis(Chassis):
    def __init__(self):
        super(ModularChassis, self).__init__()
        self.module_initialized_count = 0

    def is_modular_chassis(self):
        """
        Retrieves whether the sonic instance is part of modular chassis

        Returns:
            A bool value, should return False by default or for fixed-platforms.
            Should return True for supervisor-cards, line-cards etc running as part
            of modular-chassis.
        """
        return True

    ##############################################
    # Module methods
    ##############################################
    def initialize_single_module(self, index):
        count = self.get_num_modules()
        if index < count:
            if not self._module_list:
                self._module_list = [None] * count

            if not self._module_list[index]:
                from .module import Module
                self._module_list[index] = Module(index + 1)
                self.module_initialized_count += 1

    def initialize_modules(self):
        if not self._module_list:
            from .module import Module
            count = self.get_num_modules()
            for index in range(1, count + 1):
                self._module_list.append(Module(index))
            self.module_initialized_count = count
        elif self.module_initialized_count != len(self._module_list):
            from .module import Module
            for index in range(len(self._module_list)):
                if self._module_list[index] is None:
                    self._module_list[index] = Module(index + 1)
            self.module_initialized_count = len(self._module_list)

    def get_num_modules(self):
        """
        Retrieves the number of modules available on this chassis

        Returns:
            An integer, the number of modules available on this chassis
        """
        return DeviceDataManager.get_linecard_count()

    def get_all_modules(self):
        """
        Retrieves all modules available on this chassis

        Returns:
            A list of objects derived from ModuleBase representing all
            modules available on this chassis
        """
        self.initialize_modules()
        return self._module_list

    def get_module(self, index):
        """
        Retrieves module represented by (0-based) index <index>

        Args:
            index: An integer, the index (0-based) of the module to
            retrieve

        Returns:
            An object dervied from ModuleBase representing the specified
            module
        """
        self.initialize_single_module(index)
        return super(ModularChassis, self).get_module(index)

    @utils.default_return(-1)
    def get_module_index(self, module_name):
        """
        Retrieves module index from the module name

        Args:
            module_name: A string, prefixed by SUPERVISOR, LINE-CARD or FABRIC-CARD
            Ex. SUPERVISOR0, LINE-CARD1, FABRIC-CARD5

        Returns:
            An integer, the index of the ModuleBase object in the module_list
        """
        return int(module_name[len('LINE-CARD')-1:])

    ##############################################
    # SFP methods
    ##############################################

    def get_num_sfps(self):
        """
        Retrieves the number of sfps available on this chassis

        Returns:
            An integer, the number of sfps available on this chassis
        """
        return reduce(lambda x, y: x + y, (x.get_num_sfps() for x in self.get_all_modules()))

    def get_all_sfps(self):
        """
        Retrieves all sfps available on this chassis

        Returns:
            A list of objects derived from SfpBase representing all sfps
            available on this chassis
        """
        return reduce(lambda x, y: x + y, (x.get_all_sfps() for x in self.get_all_modules()))

    def get_sfp(self, index):
        """
        Retrieves sfp represented by (1-based) index <index>

        Args:
            index: An integer, the index (1-based) of the sfp to retrieve.
                   The index should be the sequence of a physical port in a chassis,
                   starting from 1.
                   For example, 1 for Ethernet0, 2 for Ethernet4 and so on.

        Returns:
            An object dervied from SfpBase representing the specified sfp
        """
        sfp_index = index % DeviceDataManager.get_linecard_max_port_count() - 1
        slot_id = int((index - sfp_index - 1) / 16) + 1
        module = self.get_module(slot_id - 1)
        if not module:
            return None

        return module.get_sfp(sfp_index - 1)