devices/sercomm.py

# SPDX-FileCopyrightText: 2020 The Magma Authors.
# SPDX-FileCopyrightText: 2022 Open Networking Foundation <support@opennetworking.org>
#
# SPDX-License-Identifier: BSD-3-Clause

import logging
from typing import Any, Callable, Dict, List, Optional, Type

from common.service import MagmaService
from data_models import transform_for_magma
from data_models.data_model import (
    DataModel,
    InvalidTrParamPath,
    TrParam,
)
from data_models.data_model_parameters import (
    ParameterName,
    TrParameterType,
)
from device_config.configuration_init import build_desired_config
from device_config.enodeb_config_postprocessor import EnodebConfigurationPostProcessor
from device_config.enodeb_configuration import EnodebConfiguration
from devices.device_utils import EnodebDeviceName
from logger import EnodebdLogger
from state_machines.acs_state_utils import (
    get_all_objects_to_add,
    get_all_objects_to_delete,
    get_all_param_values_to_set,
    get_params_to_get,
    parse_get_parameter_values_response,
)
from state_machines.enb_acs import EnodebAcsStateMachine
from state_machines.enb_acs_impl import BasicEnodebAcsStateMachine
from state_machines.enb_acs_states import (
    AcsMsgAndTransition,
    AcsReadMsgResult,
    AddObjectsState,
    DeleteObjectsState,
    EnbSendRebootState,
    EndSessionState,
    EnodebAcsState,
    ErrorState,
    DownloadState,
    WaitDownloadResponseState,
    WaitInformTransferCompleteState,
    GetParametersState,
    SetParameterValuesState,
    WaitGetParametersState,
    WaitInformMRebootState,
    WaitInformState,
    WaitRebootResponseState,
    WaitSetParameterValuesState,
    CheckStatusState,
)
from tr069 import models


FAPSERVICE_PATH = "Device.Services.FAPService.1."
FAP_CONTROL = FAPSERVICE_PATH + "FAPControl."


class SercommHandler(BasicEnodebAcsStateMachine):
    def __init__(self, service: MagmaService,) -> None:
        self._state_map = {}
        super().__init__(service=service, use_param_key=True)

    def reboot_asap(self) -> None:
        self.transition("reboot")

    def is_enodeb_connected(self) -> bool:
        return not isinstance(self.state, WaitInformState)

    def _init_state_map(self) -> None:
        self._state_map = {
            # Inform comes in -> Respond with InformResponse
            "wait_inform": WaitInformState(self, when_done="get_rpc_methods"),
            # If first inform after boot -> GetRpc request comes in, if not
            # empty request comes in => Transition to get_transient_params
            "get_rpc_methods": SercommGetInitState(
                self, when_done="get_transient_params",
            ),
            # Read transient readonly params.
            "get_transient_params": SercommSendGetTransientParametersState(
                self, when_upgrade="firmware_upgrade", when_done="get_params",
            ),
            "firmware_upgrade": DownloadState(self, when_done="wait_download_response"),
            "wait_download_response": WaitDownloadResponseState(
                self, when_done="wait_transfer_complete"
            ),
            "wait_transfer_complete": WaitInformTransferCompleteState(
                self,
                when_done="get_params",
                when_periodic="wait_transfer_complete",
                when_timeout="end_session",
            ),
            "get_params": SercommGetObjectParametersState(
                self,
                when_delete="delete_objs",
                when_add="add_objs",
                when_set="set_params",
                when_skip="check_status",
            ),
            "delete_objs": DeleteObjectsState(
                self, when_add="add_objs", when_skip="set_params",
            ),
            "add_objs": AddObjectsState(self, when_done="set_params"),
            "set_params": SetParameterValuesState(self, when_done="wait_set_params",),
            "wait_set_params": WaitSetParameterValuesState(
                self,
                when_done="check_get_params",
                when_apply_invasive="check_get_params",
                status_non_zero_allowed=True,
            ),
            "check_get_params": GetParametersState(
                self, when_done="check_wait_get_params", request_all_params=True,
            ),
            "check_wait_get_params": WaitGetParametersState(
                self, when_done="check_status",
            ),
            "check_status": CheckStatusState(self, when_done="check_status"),
            "end_session": EndSessionState(self),
            # These states are only entered through manual user intervention
            "reboot": EnbSendRebootState(self, when_done="wait_reboot"),
            "wait_reboot": WaitRebootResponseState(
                self, when_done="wait_post_reboot_inform",
            ),
            "wait_post_reboot_inform": WaitInformMRebootState(
                self, when_done="wait_empty", when_timeout="wait_inform",
            ),
            # The states below are entered when an unexpected message type is
            # received
            "unexpected_fault": ErrorState(
                self, inform_transition_target="wait_inform",
            ),
        }

    @property
    def device_name(self) -> str:
        return EnodebDeviceName.SERCOMM

    @property
    def data_model_class(self) -> Type[DataModel]:
        return SercommTrDataModel

    @property
    def config_postprocessor(self) -> EnodebConfigurationPostProcessor:
        return SercommConfigurationInitializer(self)

    @property
    def state_map(self) -> Dict[str, EnodebAcsState]:
        return self._state_map

    @property
    def disconnected_state_name(self) -> str:
        return "wait_inform"

    @property
    def unexpected_fault_state_name(self) -> str:
        return "unexpected_fault"


class SASParameters:
    """ Class modeling the SAS parameters and their TR path"""

    # For CBRS radios we set this to the limit and the SAS can reduce the
    # power if needed.

    SAS_PARAMETERS = {
        ParameterName.SAS_ENABLE: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.Enable",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.SAS_SERVER_URL: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.Server",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_UID: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.UserContactInformation",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_CATEGORY: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.Category",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_CHANNEL_TYPE: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.ProtectionLevel",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_CERT_SUBJECT: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.CertSubject",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        # SAS_IC_GROUP_ID: TrParam(
        #     FAP_CONTROL + 'LTE.X_SCM_SAS.ICGGroupId', is_invasive=False,
        #     type=TrParameterType.STRING, False),
        ParameterName.SAS_LOCATION: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.Location",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_HEIGHT_TYPE: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.HeightType",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_CPI_ENABLE: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.CPIEnable",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.SAS_CPI_IPE: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.CPIInstallParamSuppliedEnable",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.SAS_FCCID: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.FCCIdentificationNumber",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_MEAS_CAPS: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.MeasCapability",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_MANU_ENABLE: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.ManufacturerPrefixEnable",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.SAS_CPI_NAME: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.CPIName",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_CPI_ID: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.CPIId",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SAS_ANTA_AZIMUTH: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.AntennaAzimuth",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.SAS_ANTA_DOWNTILT: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.AntennaDowntilt",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.SAS_ANTA_GAIN: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.AntennaGain",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.SAS_ANTA_BEAMWIDTH: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.AntennaBeamwidth",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.SAS_CPI_DATA: TrParam(
            FAP_CONTROL + "LTE.X_SCM_SAS.CPISignatureData",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
    }


class StatusParameters:
    """
    Stateful class that converts eNB status to Magma understood status.
    Sercomm has many status params that interact with each other.
    This class maintains the business logic of parsing these status params
    and converting it to Magma understood fields.
    """

    # Status parameters
    DEFAULT_GW = "defaultGW"
    SYNC_STATUS = "syncStatus"
    SAS_STATUS = "sasStatus"
    ENB_STATUS = "enbStatus"
    GPS_SCAN_STATUS = "gpsScanStatus"

    STATUS_PARAMETERS = {
        # Status nodes
        DEFAULT_GW: TrParam(
            "Device.X_SCM_DeviceFeature.X_SCM_NEStatus.X_SCM_DEFGW_Status",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        SAS_STATUS: TrParam(
            "Device.Services.FAPService.1.FAPControl.LTE.X_SCM_SAS.State",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        GPS_SCAN_STATUS: TrParam(
            "Device.FAP.GPS.ScanStatus",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.GPS_LAT: TrParam(
            "Device.FAP.GPS.LockedLatitude",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.GPS_LONG: TrParam(
            "Device.FAP.GPS.LockedLongitude",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SW_VERSION: TrParam(
            "Device.DeviceInfo.SoftwareVersion",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SERIAL_NUMBER: TrParam(
            "Device.DeviceInfo.SerialNumber",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
    }

    # Derived status params that don't have tr69 representation.
    DERIVED_STATUS_PARAMETERS = {
        ParameterName.RF_TX_STATUS: TrParam(
            "Device.X_SCM_DeviceFeature.SystemParams.RunControl.Service.4.Status",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.GPS_STATUS: TrParam(
            "Device.FAP.GPS.ScanStatus",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.PTP_STATUS: TrParam(
            InvalidTrParamPath,
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.MME_STATUS: TrParam(
            InvalidTrParamPath,  # FAP_CONTROL + "LTE.X_SCM_S1State",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.OP_STATE: TrParam(
            "Device.X_SCM_DeviceFeature.SystemParams.RunControl.Service.4.Status",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
    }

    @classmethod
    def set_magma_device_cfg(
        cls, name_to_val: Dict, device_cfg: EnodebConfiguration,
    ):
        """
        Convert Sercomm name_to_val representation to magma device_cfg
        """
        success_str = "SUCCESS"  # String constant returned by radio
        insync_str = "INSYNC"

        if (
            name_to_val.get(cls.DEFAULT_GW)
            and name_to_val[cls.DEFAULT_GW].upper() != success_str
        ):
            # Nothing will proceed if the eNB doesn't have an IP on the WAN
            serial_num = "unknown"
            if device_cfg.has_parameter(ParameterName.SERIAL_NUMBER):
                serial_num = device_cfg.get_parameter(ParameterName.SERIAL_NUMBER,)
            EnodebdLogger.error(
                "Radio with serial number %s doesn't have IP address " "on WAN",
                serial_num,
            )
            device_cfg.set_parameter(
                param_name=ParameterName.RF_TX_STATUS, value=False,
            )
            device_cfg.set_parameter(
                param_name=ParameterName.GPS_STATUS, value=False,
            )
            device_cfg.set_parameter(
                param_name=ParameterName.PTP_STATUS, value=False,
            )
            device_cfg.set_parameter(
                param_name=ParameterName.MME_STATUS, value=False,
            )
            device_cfg.set_parameter(
                param_name=ParameterName.OP_STATE, value=False,
            )
            return

        if (
            name_to_val.get(cls.SAS_STATUS)
            and name_to_val[cls.SAS_STATUS].upper() == success_str
        ):
            device_cfg.set_parameter(
                param_name=ParameterName.RF_TX_STATUS, value=True,
            )
        else:
            # No sas grant so not transmitting. There is no explicit node for Tx
            # in Sercomm
            device_cfg.set_parameter(
                param_name=ParameterName.RF_TX_STATUS, value=False,
            )

        if (
            name_to_val.get(cls.GPS_SCAN_STATUS)
            and name_to_val[cls.GPS_SCAN_STATUS].upper() == success_str
        ):
            device_cfg.set_parameter(
                param_name=ParameterName.GPS_STATUS, value=True,
            )
            # Time comes through GPS so can only be insync with GPS is
            # in sync, we use PTP_STATUS field to overload timer is in Sync.
            if (
                name_to_val.get(cls.SYNC_STATUS)
                and name_to_val[cls.SYNC_STATUS].upper() == insync_str
            ):
                device_cfg.set_parameter(
                    param_name=ParameterName.PTP_STATUS, value=True,
                )
            else:
                device_cfg.set_parameter(
                    param_name=ParameterName.PTP_STATUS, value=False,
                )
        else:
            device_cfg.set_parameter(
                param_name=ParameterName.GPS_STATUS, value=False,
            )
            device_cfg.set_parameter(
                param_name=ParameterName.PTP_STATUS, value=False,
            )

        if (
            name_to_val.get(cls.DEFAULT_GW)
            and name_to_val[cls.DEFAULT_GW].upper() == success_str
        ):
            device_cfg.set_parameter(
                param_name=ParameterName.MME_STATUS, value=True,
            )
        else:
            device_cfg.set_parameter(
                param_name=ParameterName.MME_STATUS, value=False,
            )

        if (
            name_to_val.get(cls.ENB_STATUS)
            and name_to_val[cls.ENB_STATUS].upper() == success_str
        ):
            device_cfg.set_parameter(
                param_name=ParameterName.OP_STATE, value=True,
            )
        else:
            device_cfg.set_parameter(
                param_name=ParameterName.OP_STATE, value=False,
            )

        pass_through_params = [ParameterName.GPS_LAT, ParameterName.GPS_LONG]

        for name in pass_through_params:
            device_cfg.set_parameter(name, name_to_val[name])


class SercommMiscParameters:
    """
    Default set of parameters that enable carrier aggregation and other
    miscellaneous properties
    """

    MISC_PARAMETERS = {
        ParameterName.CARRIER_AGG_ENABLE: TrParam(
            FAP_CONTROL + "LTE.X_SCM_RRMConfig.X_SCM_CA_Enable",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.CARRIER_NUMBER: TrParam(
            FAP_CONTROL + "LTE.X_SCM_RRMConfig.X_SCM_Cell_Number",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.CONTIGUOUS_CC: TrParam(
            FAP_CONTROL + "LTE.X_SCM_RRMConfig.X_SCM_CELL_Freq_Contiguous",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.PRIM_SOURCE: TrParam(
            FAPSERVICE_PATH + "REM.X_SCM_tfcsManagerConfig.primSrc",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.CWMP_ENABLE: TrParam(
            "Device.ManagementServer.EnableCWMP",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
    }


class SercommTrDataModel(DataModel):
    """
    Class to represent relevant data model parameters from TR-196/TR-098.
    This class is effectively read-only.

    These models have these idiosyncrasies (on account of running TR098):

    - Parameter content root is different (InternetGatewayDevice)
    - GetParameter queries with a wildcard e.g. InternetGatewayDevice. do
      not respond with the full tree (we have to query all parameters)
    - MME status is not exposed - we assume the MME is connected if
      the eNodeB is transmitting (OpState=true)
    - Parameters such as band capability/duplex config
      are rooted under `boardconf.` and not the device config root
    - Num PLMNs is not reported by these units
    """

    PARAMETERS = {
        # Top-level objects
        ParameterName.DEVICE: TrParam(
            "Device.",
            is_invasive=False,
            type=TrParameterType.OBJECT,
            is_optional=False,
        ),
        ParameterName.FAP_SERVICE: TrParam(
            FAP_CONTROL,
            is_invasive=False,
            type=TrParameterType.OBJECT,
            is_optional=False,
        ),
        # Device info
        ParameterName.IP_ADDRESS: TrParam(
            "Device.IP.Interface.1.IPv4Address.1.IPAddress",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        # RF-related parameters
        ParameterName.FREQ_BAND_1: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.FreqBandIndicator",
            is_invasive=False,
            type=TrParameterType.UNSIGNED_INT,
            is_optional=False,
        ),
        ParameterName.FREQ_BAND_2: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.X_SCM_FreqBandIndicator2",
            is_invasive=False,
            type=TrParameterType.UNSIGNED_INT,
            is_optional=False,
        ),
        ParameterName.FREQ_BAND_LIST: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.X_SCM_FreqBandIndicatorConfigList",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.EARFCNDL: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.EARFCNDL",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.EARFCNUL: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.EARFCNUL",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.EARFCNDL2: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.X_SCM_EARFCNDL2",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.EARFCNUL2: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.X_SCM_EARFCNUL2",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.EARFCNDL_LIST: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.X_SCM_EARFCNDLConfigList",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.EARFCNUL_LIST: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.X_SCM_EARFCNULConfigList",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.DL_BANDWIDTH: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.DLBandwidth",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.UL_BANDWIDTH: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.ULBandwidth",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.PCI_LIST: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.PhyCellID",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.SUBFRAME_ASSIGNMENT: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.PHY.TDDFrame.SubFrameAssignment",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.SPECIAL_SUBFRAME_PATTERN: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.PHY.TDDFrame.SpecialSubframePatterns",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.CELL_ENABLE64QAM: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.PHY.PUSCH.Enable64QAM",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.CELL_ID: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.Common.CellIdentity",
            is_invasive=False,
            type=TrParameterType.UNSIGNED_INT,
            is_optional=False,
        ),
        # Readonly LTE state
        ParameterName.ADMIN_STATE: TrParam(
            FAP_CONTROL + "LTE.AdminState",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.GPS_ENABLE: TrParam(
            "Device.FAP.GPS.ScanOnBoot",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        # Core network parameters
        ParameterName.MME_ADDRESS: TrParam(
            FAP_CONTROL + "LTE.Gateway.S1SigLinkServerList",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.MME_PORT: TrParam(
            FAP_CONTROL + "LTE.Gateway.S1SigLinkPort",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.TAC1: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.EPC.TAC",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.TAC2: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.EPC.X_SCM_TAC2",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        # Management server parameters
        ParameterName.PERIODIC_INFORM_ENABLE: TrParam(
            "Device.ManagementServer.PeriodicInformEnable",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.PERIODIC_INFORM_INTERVAL: TrParam(
            "Device.ManagementServer.PeriodicInformInterval",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        # Performance management parameters
        ParameterName.PERF_MGMT_ENABLE: TrParam(
            "Device.FAP.PerfMgmt.Config.1.Enable",
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        ),
        ParameterName.PERF_MGMT_UPLOAD_INTERVAL: TrParam(
            "Device.FAP.PerfMgmt.Config.1.PeriodicUploadInterval",
            is_invasive=False,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.PERF_MGMT_UPLOAD_URL: TrParam(
            "Device.FAP.PerfMgmt.Config.1.URL",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
        ParameterName.TX_POWER: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.RAN.RF.X_SCM_TxPowerConfig",
            is_invasive=True,
            type=TrParameterType.INT,
            is_optional=False,
        ),
        ParameterName.TUNNEL_TYPE: TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.Tunnel.1.TunnelRef",
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        ),
    }
    TRANSFORMS_FOR_ENB = {}
    NUM_PLMNS_IN_CONFIG = 1
    for i in range(1, NUM_PLMNS_IN_CONFIG + 1):
        PARAMETERS[ParameterName.PLMN_N % i] = TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.EPC.PLMNList.%d." % i,
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        )
        PARAMETERS[ParameterName.PLMN_N_CELL_RESERVED % i] = TrParam(
            FAPSERVICE_PATH
            + "CellConfig.LTE.EPC.PLMNList.%d.CellReservedForOperatorUse" % i,
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        )
        PARAMETERS[ParameterName.PLMN_N_ENABLE % i] = TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.EPC.PLMNList.%d.Enable" % i,
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        )
        PARAMETERS[ParameterName.PLMN_N_PRIMARY % i] = TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.EPC.PLMNList.%d.IsPrimary" % i,
            is_invasive=False,
            type=TrParameterType.BOOLEAN,
            is_optional=False,
        )
        PARAMETERS[ParameterName.PLMN_N_PLMNID % i] = TrParam(
            FAPSERVICE_PATH + "CellConfig.LTE.EPC.PLMNList.%d.PLMNID" % i,
            is_invasive=False,
            type=TrParameterType.STRING,
            is_optional=False,
        )

    PARAMETERS.update(SASParameters.SAS_PARAMETERS)
    PARAMETERS.update(SercommMiscParameters.MISC_PARAMETERS)
    PARAMETERS.update(StatusParameters.STATUS_PARAMETERS)
    PARAMETERS.update(StatusParameters.DERIVED_STATUS_PARAMETERS)

    TRANSFORMS_FOR_MAGMA = {
        # We don't set these parameters
        ParameterName.BAND_CAPABILITY: transform_for_magma.band_capability,
        ParameterName.DUPLEX_MODE_CAPABILITY: transform_for_magma.duplex_mode,
    }

    @classmethod
    def get_parameter(cls, param_name: ParameterName) -> Optional[TrParam]:
        return cls.PARAMETERS.get(param_name)

    @classmethod
    def _get_magma_transforms(cls,) -> Dict[ParameterName, Callable[[Any], Any]]:
        return cls.TRANSFORMS_FOR_MAGMA

    @classmethod
    def _get_enb_transforms(cls) -> Dict[ParameterName, Callable[[Any], Any]]:
        return cls.TRANSFORMS_FOR_ENB

    @classmethod
    def get_load_parameters(cls) -> List[ParameterName]:
        """
        Load all the parameters instead of a subset.
        """
        return list(cls.PARAMETERS.keys())

    @classmethod
    def get_num_plmns(cls) -> int:
        return cls.NUM_PLMNS_IN_CONFIG

    @classmethod
    def get_parameter_names(cls) -> List[ParameterName]:
        excluded_params = [
            str(ParameterName.DEVICE),
            str(ParameterName.FAP_SERVICE),
        ]
        names = list(
            filter(
                lambda x: (not str(x).startswith("PLMN"))
                and (str(x) not in excluded_params),
                cls.PARAMETERS.keys(),
            ),
        )
        return names

    @classmethod
    def get_numbered_param_names(cls,) -> Dict[ParameterName, List[ParameterName]]:
        names = {}
        for i in range(1, cls.NUM_PLMNS_IN_CONFIG + 1):
            params = [
                ParameterName.PLMN_N_CELL_RESERVED % i,
                ParameterName.PLMN_N_ENABLE % i,
                ParameterName.PLMN_N_PRIMARY % i,
                ParameterName.PLMN_N_PLMNID % i,
            ]
            names[ParameterName.PLMN_N % i] = params

        return names

    @classmethod
    def get_sas_param_names(cls) -> List[ParameterName]:
        return SASParameters.SAS_PARAMETERS.keys()


class SercommConfigurationInitializer(EnodebConfigurationPostProcessor):
    """
    Class to add the sas related parameters to the desired config.
    """

    SAS_KEY = "sas"

    def __init__(self, acs: EnodebAcsStateMachine):
        super().__init__()
        self.acs = acs

    def postprocess(self, desired_cfg: EnodebConfiguration) -> None:
        pass


class SercommSendGetTransientParametersState(EnodebAcsState):
    """
    Periodically read eNodeB status. Note: keep frequency low to avoid
    backing up large numbers of read operations if enodebd is busy.
    Some eNB parameters are read only and updated by the eNB itself.
    """

    def __init__(self, acs: EnodebAcsStateMachine, when_upgrade: str, when_done: str):
        super().__init__()
        self.acs = acs
        self.upgrade_transition = when_upgrade
        self.done_transition = when_done

    def get_msg(self, message: Any) -> AcsMsgAndTransition:

        request = models.GetParameterValues()
        request.ParameterNames = models.ParameterNames()
        request.ParameterNames.string = []

        for _, tr_param in StatusParameters.STATUS_PARAMETERS.items():
            path = tr_param.path
            request.ParameterNames.string.append(path)

        request.ParameterNames.string.append(
            "Device.X_SCM_DeviceFeature.SystemParams.RunControl.Service.4.Status",
        )

        request.ParameterNames.arrayType = "xsd:string[%d]" % len(
            request.ParameterNames.string
        )

        return AcsMsgAndTransition(msg=request, next_state=None)

    def read_msg(self, message: Any) -> AcsReadMsgResult:

        name_to_val = parse_get_parameter_values_response(self.acs.data_model, message,)

        # Update device configuration
        StatusParameters.set_magma_device_cfg(
            name_to_val, self.acs.device_cfg,
        )

        if name_to_val["SW version"] != "RC3923@2202151120":
            print("Get into Firmware Upgrade state")
            return AcsReadMsgResult(
                msg_handled=True, next_state=self.upgrade_transition
            )

        print("Skip firmware upgrade, configure the enb")
        return AcsReadMsgResult(msg_handled=True, next_state=self.done_transition)

    def state_description(self) -> str:
        return "Getting transient read-only parameters"


class SercommGetInitState(EnodebAcsState):
    """
    After the first Inform message the following can happen:
    1 - eNB can try to learn the RPC method of the ACS, reply back with the
    RPC response (happens right after boot)
    2 - eNB can send an empty message -> This means that the eNB is already
    provisioned so transition to next state. Only transition to next state
    after this message.
    3 - Some other method call that we don't care about so ignore.
    expected that the eNB -> This is an unhandled state so unlikely
    """

    def __init__(self, acs: EnodebAcsStateMachine, when_done):
        super().__init__()
        self.acs = acs
        self.done_transition = when_done
        self._is_rpc_request = False

    def get_msg(self, message: Any) -> AcsMsgAndTransition:
        """
        Return empty message response if care about this
        message type otherwise return empty RPC methods response.
        """
        if not self._is_rpc_request:
            resp = models.DummyInput()
            return AcsMsgAndTransition(msg=resp, next_state=None)

        resp = models.GetRPCMethodsResponse()
        resp.MethodList = models.MethodList()
        RPC_METHODS = ["Inform", "GetRPCMethods", "TransferComplete"]
        resp.MethodList.arrayType = "xsd:string[%d]" % len(RPC_METHODS)
        resp.MethodList.string = RPC_METHODS
        # Don't transition to next state wait for the empty HTTP post
        return AcsMsgAndTransition(msg=resp, next_state=None)

    def read_msg(self, message: Any) -> AcsReadMsgResult:
        # If this is a regular Inform, not after a reboot we'll get an empty
        # message, in this case transition to the next state. We consider
        # this phase as "initialized"
        if isinstance(message, models.DummyInput):
            return AcsReadMsgResult(msg_handled=True, next_state=self.done_transition,)
        if not isinstance(message, models.GetRPCMethods):
            # Unexpected, just don't die, ignore message.
            logging.error("Ignoring message %s", str(type(message)))
            # Set this so get_msg will return an empty message
            self._is_rpc_request = False
        else:
            # Return a valid RPC response
            self._is_rpc_request = True
        return AcsReadMsgResult(msg_handled=True, next_state=None)

    def state_description(self) -> str:
        return "Initializing the post boot sequence for eNB"


class SercommGetObjectParametersState(EnodebAcsState):
    """
    Get information on parameters belonging to objects that can be added or
    removed from the configuration.

    Englewood will report a parameter value as None if it does not exist
    in the data model, rather than replying with a Fault message like most
    eNB devices.
    """

    def __init__(
        self,
        acs: EnodebAcsStateMachine,
        when_delete: str,
        when_add: str,
        when_set: str,
        when_skip: str,
    ):
        super().__init__()
        self.acs = acs
        self.rm_obj_transition = when_delete
        self.add_obj_transition = when_add
        self.set_params_transition = when_set
        self.skip_transition = when_skip

    def get_params_to_get(self, data_model: DataModel,) -> List[ParameterName]:
        names = []

        # First get base params
        names = get_params_to_get(
            self.acs.device_cfg, self.acs.data_model, request_all_params=True,
        )
        # Add object params.
        num_plmns = data_model.get_num_plmns()
        obj_to_params = data_model.get_numbered_param_names()
        for i in range(1, num_plmns + 1):
            obj_name = ParameterName.PLMN_N % i
            desired = obj_to_params[obj_name]
            names += desired

        print(obj_to_params)
        return names

    def get_msg(self, message: Any) -> AcsMsgAndTransition:
        """ Respond with GetParameterValuesRequest """
        names = self.get_params_to_get(self.acs.data_model,)

        # Generate the request
        request = models.GetParameterValues()
        request.ParameterNames = models.ParameterNames()
        request.ParameterNames.arrayType = "xsd:string[%d]" % len(names)
        request.ParameterNames.string = []
        for name in names:
            path = self.acs.data_model.get_parameter(name).path
            if path is not InvalidTrParamPath:
                # Only get data elements backed by tr69 path
                request.ParameterNames.string.append(path)

        return AcsMsgAndTransition(msg=request, next_state=None)

    def read_msg(self, message: Any) -> AcsReadMsgResult:
        """
        Process GetParameterValuesResponse
        """
        if not isinstance(message, models.GetParameterValuesResponse):
            return AcsReadMsgResult(msg_handled=False, next_state=None)

        path_to_val = {}
        for param_value_struct in message.ParameterList.ParameterValueStruct:
            path_to_val[param_value_struct.Name] = param_value_struct.Value.Data

        EnodebdLogger.debug("Received object parameters: %s", str(path_to_val))

        # Parse simple params
        param_name_list = self.acs.data_model.get_parameter_names()

        for name in param_name_list:
            path = self.acs.data_model.get_parameter(name).path
            if path in path_to_val:
                value = path_to_val.get(path)
                magma_val = self.acs.data_model.transform_for_magma(name, value,)
                self.acs.device_cfg.set_parameter(name, magma_val)

        # Parse object params
        num_plmns = self.acs.data_model.get_num_plmns()
        for i in range(1, num_plmns + 1):
            obj_name = ParameterName.PLMN_N % i
            obj_to_params = self.acs.data_model.get_numbered_param_names()
            param_name_list = obj_to_params[obj_name]
            for name in param_name_list:
                path = self.acs.data_model.get_parameter(name).path
                if path in path_to_val:
                    value = path_to_val.get(path)
                    if value is None:
                        continue
                    if obj_name not in self.acs.device_cfg.get_object_names():
                        self.acs.device_cfg.add_object(obj_name)
                    magma_value = self.acs.data_model.transform_for_magma(name, value)
                    self.acs.device_cfg.set_parameter_for_object(
                        name, magma_value, obj_name,
                    )

        # Now we have enough information to build the desired configuration
        if self.acs.desired_cfg is None:
            self.acs.desired_cfg = build_desired_config(
                self.acs.device_cfg, self.acs.data_model, self.acs.config_postprocessor,
            )

        if (
            len(get_all_objects_to_delete(self.acs.desired_cfg, self.acs.device_cfg,),)
            > 0
        ):
            return AcsReadMsgResult(
                msg_handled=True, next_state=self.rm_obj_transition,
            )
        elif (
            len(get_all_objects_to_add(self.acs.desired_cfg, self.acs.device_cfg,),) > 0
        ):
            return AcsReadMsgResult(
                msg_handled=True, next_state=self.add_obj_transition,
            )
        elif (
            len(
                get_all_param_values_to_set(
                    self.acs.desired_cfg, self.acs.device_cfg, self.acs.data_model,
                ),
            )
            > 0
        ):
            return AcsReadMsgResult(
                msg_handled=True, next_state=self.set_params_transition,
            )
        return AcsReadMsgResult(msg_handled=True, next_state=self.skip_transition,)

    def state_description(self) -> str:
        return "Getting well known parameters"