Initial commit moving openolt adapter from voltha-go to the new repo.
This version works with ponsim rather than openolt, this is temporary.
It is currently being fixed to work with openolt.
Change-Id: I34a800c98f050140b367e2d474b7aa8b79f34b9a
Signed-off-by: William Kurkian <wkurkian@cisco.com>
diff --git a/python/adapters/extensions/omci/onu_device_entry.py b/python/adapters/extensions/omci/onu_device_entry.py
new file mode 100644
index 0000000..7a0c439
--- /dev/null
+++ b/python/adapters/extensions/omci/onu_device_entry.py
@@ -0,0 +1,635 @@
+#
+# Copyright 2018 the original author or authors.
+#
+# 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
+from copy import deepcopy
+from voltha.protos.device_pb2 import ImageDownload
+from voltha.extensions.omci.omci_defs import EntityOperations, ReasonCodes
+import voltha.extensions.omci.omci_entities as omci_entities
+from voltha.extensions.omci.omci_cc import OMCI_CC
+from common.event_bus import EventBusClient
+from voltha.extensions.omci.tasks.task_runner import TaskRunner
+from voltha.extensions.omci.onu_configuration import OnuConfiguration
+from voltha.extensions.omci.tasks.reboot_task import OmciRebootRequest, RebootFlags
+from voltha.extensions.omci.tasks.omci_modify_request import OmciModifyRequest
+from voltha.extensions.omci.omci_me import OntGFrame
+from voltha.extensions.omci.state_machines.image_agent import ImageAgent
+
+from twisted.internet import reactor, defer
+from enum import IntEnum
+
+OP = EntityOperations
+RC = ReasonCodes
+
+ACTIVE_KEY = 'active'
+IN_SYNC_KEY = 'in-sync'
+LAST_IN_SYNC_KEY = 'last-in-sync-time'
+SUPPORTED_MESSAGE_ENTITY_KEY = 'managed-entities'
+SUPPORTED_MESSAGE_TYPES_KEY = 'message-type'
+
+
+class OnuDeviceEvents(IntEnum):
+ # Events of interest to Device Adapters and OpenOMCI State Machines
+ DeviceStatusEvent = 0 # OnuDeviceEntry running status changed
+ MibDatabaseSyncEvent = 1 # MIB database sync changed
+ OmciCapabilitiesEvent = 2 # OMCI ME and message type capabilities
+ AlarmDatabaseSyncEvent = 3 # Alarm database sync changed
+
+ # TODO: Add other events here as needed
+
+
+class OnuDeviceEntry(object):
+ """
+ An ONU Device entry in the MIB
+ """
+ def __init__(self, omci_agent, device_id, adapter_agent, custom_me_map,
+ mib_db, alarm_db, support_classes, clock=None):
+ """
+ Class initializer
+
+ :param omci_agent: (OpenOMCIAgent) Reference to OpenOMCI Agent
+ :param device_id: (str) ONU Device ID
+ :param adapter_agent: (AdapterAgent) Adapter agent for ONU
+ :param custom_me_map: (dict) Additional/updated ME to add to class map
+ :param mib_db: (MibDbApi) MIB Database reference
+ :param alarm_db: (MibDbApi) Alarm Table/Database reference
+ :param support_classes: (dict) State machines and tasks for this ONU
+ """
+ self.log = structlog.get_logger(device_id=device_id)
+
+ self._started = False
+ self._omci_agent = omci_agent # OMCI AdapterAgent
+ self._device_id = device_id # ONU Device ID
+ self._adapter_agent = adapter_agent
+ self._runner = TaskRunner(device_id, clock=clock) # OMCI_CC Task runner
+ self._deferred = None
+ # self._img_download_deferred = None # deferred of image file download from server
+ self._omci_upgrade_deferred = None # deferred of ONU OMCI upgrading procedure
+ self._omci_activate_deferred = None # deferred of ONU OMCI Softwre Image Activate
+ self._img_deferred = None # deferred returned to caller of do_onu_software_download
+ self._first_in_sync = False
+ self._first_capabilities = False
+ self._timestamp = None
+ # self._image_download = None # (voltha_pb2.ImageDownload)
+ self.reactor = clock if clock is not None else reactor
+
+ # OMCI related databases are on a per-agent basis. State machines and tasks
+ # are per ONU Vendor
+ #
+ self._support_classes = support_classes
+ self._configuration = None
+
+ try:
+ # MIB Synchronization state machine
+ self._mib_db_in_sync = False
+ mib_synchronizer_info = support_classes.get('mib-synchronizer')
+ advertise = mib_synchronizer_info['advertise-events']
+ self._mib_sync_sm = mib_synchronizer_info['state-machine'](self._omci_agent,
+ device_id,
+ mib_synchronizer_info['tasks'],
+ mib_db,
+ advertise_events=advertise)
+ # ONU OMCI Capabilities state machine
+ capabilities_info = support_classes.get('omci-capabilities')
+ advertise = capabilities_info['advertise-events']
+ self._capabilities_sm = capabilities_info['state-machine'](self._omci_agent,
+ device_id,
+ capabilities_info['tasks'],
+ advertise_events=advertise)
+ # ONU Performance Monitoring Intervals state machine
+ interval_info = support_classes.get('performance-intervals')
+ advertise = interval_info['advertise-events']
+ self._pm_intervals_sm = interval_info['state-machine'](self._omci_agent, device_id,
+ interval_info['tasks'],
+ advertise_events=advertise)
+
+ # ONU ALARM Synchronization state machine
+ self._alarm_db_in_sync = False
+ alarm_synchronizer_info = support_classes.get('alarm-synchronizer')
+ advertise = alarm_synchronizer_info['advertise-events']
+ self._alarm_sync_sm = alarm_synchronizer_info['state-machine'](self._omci_agent,
+ device_id,
+ alarm_synchronizer_info['tasks'],
+ alarm_db,
+ advertise_events=advertise)
+ # State machine of downloading image file from server
+ downloader_info = support_classes.get('image_downloader')
+ image_upgrader_info = support_classes.get('image_upgrader')
+ # image_activate_info = support_classes.get('image_activator')
+ advertise = downloader_info['advertise-event']
+ # self._img_download_sm = downloader_info['state-machine'](self._omci_agent, device_id,
+ # downloader_info['tasks'],
+ # advertise_events=advertise)
+ self._image_agent = ImageAgent(self._omci_agent, device_id,
+ downloader_info['state-machine'], downloader_info['tasks'],
+ image_upgrader_info['state-machine'], image_upgrader_info['tasks'],
+ # image_activate_info['state-machine'],
+ advertise_events=advertise, clock=clock)
+
+ # self._omci_upgrade_sm = image_upgrader_info['state-machine'](device_id, advertise_events=advertise)
+
+ except Exception as e:
+ self.log.exception('state-machine-create-failed', e=e)
+ raise
+
+ # Put state machines in the order you wish to start them
+
+ self._state_machines = []
+ self._on_start_state_machines = [ # Run when 'start()' called
+ self._mib_sync_sm,
+ self._capabilities_sm,
+ ]
+ self._on_sync_state_machines = [ # Run after first in_sync event
+ self._alarm_sync_sm,
+ ]
+ self._on_capabilities_state_machines = [ # Run after first capabilities events
+ self._pm_intervals_sm
+ ]
+ self._custom_me_map = custom_me_map
+ self._me_map = omci_entities.entity_id_to_class_map.copy()
+
+ if custom_me_map is not None:
+ self._me_map.update(custom_me_map)
+
+ self.event_bus = EventBusClient()
+
+ # Create OMCI communications channel
+ self._omci_cc = OMCI_CC(adapter_agent, self.device_id, self._me_map, clock=clock)
+
+ @staticmethod
+ def event_bus_topic(device_id, event):
+ """
+ Get the topic name for a given event for this ONU Device
+ :param device_id: (str) ONU Device ID
+ :param event: (OnuDeviceEvents) Type of event
+ :return: (str) Topic string
+ """
+ assert event in OnuDeviceEvents, \
+ 'Event {} is not an ONU Device Event'.format(event.name)
+ return 'omci-device:{}:{}'.format(device_id, event.name)
+
+ @property
+ def device_id(self):
+ return self._device_id
+
+ @property
+ def omci_cc(self):
+ return self._omci_cc
+
+ @property
+ def adapter_agent(self):
+ return self._adapter_agent
+
+ @property
+ def task_runner(self):
+ return self._runner
+
+ @property
+ def mib_synchronizer(self):
+ """
+ Reference to the OpenOMCI MIB Synchronization state machine for this ONU
+ """
+ return self._mib_sync_sm
+
+ @property
+ def omci_capabilities(self):
+ """
+ Reference to the OpenOMCI OMCI Capabilities state machine for this ONU
+ """
+ return self._capabilities_sm
+
+ @property
+ def pm_intervals_state_machine(self):
+ """
+ Reference to the OpenOMCI PM Intervals state machine for this ONU
+ """
+ return self._pm_intervals_sm
+
+ def set_pm_config(self, pm_config):
+ """
+ Set PM interval configuration
+
+ :param pm_config: (OnuPmIntervalMetrics) PM Interval configuration
+ """
+ self._pm_intervals_sm.set_pm_config(pm_config)
+
+ @property
+ def timestamp(self):
+ """Pollable Metrics last collected timestamp"""
+ return self._timestamp
+
+ @timestamp.setter
+ def timestamp(self, value):
+ self._timestamp = value
+
+ @property
+ def alarm_synchronizer(self):
+ """
+ Reference to the OpenOMCI Alarm Synchronization state machine for this ONU
+ """
+ return self._alarm_sync_sm
+
+ @property
+ def active(self):
+ """
+ Is the ONU device currently active/running
+ """
+ return self._started
+
+ @property
+ def custom_me_map(self):
+ """ Vendor-specific Managed Entity Map for this vendor's device"""
+ return self._custom_me_map
+
+ @property
+ def me_map(self):
+ """ Combined ME and Vendor-specific Managed Entity Map for this device"""
+ return self._me_map
+
+ def _cancel_deferred(self):
+ d, self._deferred = self._deferred, None
+ try:
+ if d is not None and not d.called:
+ d.cancel()
+ except:
+ pass
+
+ @property
+ def mib_db_in_sync(self):
+ return self._mib_db_in_sync
+
+ @mib_db_in_sync.setter
+ def mib_db_in_sync(self, value):
+ if self._mib_db_in_sync != value:
+ # Save value
+ self._mib_db_in_sync = value
+
+ # Start up other state machines if needed
+ if self._first_in_sync:
+ self.first_in_sync_event()
+
+ # Notify any event listeners
+ topic = OnuDeviceEntry.event_bus_topic(self.device_id,
+ OnuDeviceEvents.MibDatabaseSyncEvent)
+ msg = {
+ IN_SYNC_KEY: self._mib_db_in_sync,
+ LAST_IN_SYNC_KEY: self.mib_synchronizer.last_mib_db_sync
+ }
+ self.event_bus.publish(topic=topic, msg=msg)
+
+ @property
+ def alarm_db_in_sync(self):
+ return self._alarm_db_in_sync
+
+ @alarm_db_in_sync.setter
+ def alarm_db_in_sync(self, value):
+ if self._alarm_db_in_sync != value:
+ # Save value
+ self._alarm_db_in_sync = value
+
+ # Start up other state machines if needed
+ if self._first_in_sync:
+ self.first_in_sync_event()
+
+ # Notify any event listeners
+ topic = OnuDeviceEntry.event_bus_topic(self.device_id,
+ OnuDeviceEvents.AlarmDatabaseSyncEvent)
+ msg = {
+ IN_SYNC_KEY: self._alarm_db_in_sync
+ }
+ self.event_bus.publish(topic=topic, msg=msg)
+
+ @property
+ def configuration(self):
+ """
+ Get the OMCI Configuration object for this ONU. This is a class that provides some
+ common database access functions for ONU capabilities and read-only configuration values.
+
+ :return: (OnuConfiguration)
+ """
+ return self._configuration
+
+ @property
+ def image_agent(self):
+ return self._image_agent
+
+ # @property
+ # def image_download(self):
+ # return self._image_download
+
+ def start(self):
+ """
+ Start the ONU Device Entry state machines
+ """
+ self.log.debug('OnuDeviceEntry.start', previous=self._started)
+ if self._started:
+ return
+
+ self._started = True
+ self._omci_cc.enabled = True
+ self._first_in_sync = True
+ self._first_capabilities = True
+ self._runner.start()
+ self._configuration = OnuConfiguration(self._omci_agent, self._device_id)
+
+ # Start MIB Sync and other state machines that can run before the first
+ # MIB Synchronization event occurs. Start 'later' so that any
+ # ONU Device, OMCI DB, OMCI Agent, and others are fully started before
+ # performing the start.
+
+ self._state_machines = []
+
+ def start_state_machines(machines):
+ for sm in machines:
+ self._state_machines.append(sm)
+ sm.start()
+
+ self._deferred = reactor.callLater(0, start_state_machines,
+ self._on_start_state_machines)
+ # Notify any event listeners
+ self._publish_device_status_event()
+
+ def stop(self):
+ """
+ Stop the ONU Device Entry state machines
+ """
+ if not self._started:
+ return
+
+ self._started = False
+ self._cancel_deferred()
+ self._omci_cc.enabled = False
+
+ # Halt MIB Sync and other state machines
+ for sm in self._state_machines:
+ sm.stop()
+
+ self._state_machines = []
+
+ # Stop task runner
+ self._runner.stop()
+
+ # Notify any event listeners
+ self._publish_device_status_event()
+
+ def first_in_sync_event(self):
+ """
+ This event is called on the first MIB synchronization event after
+ OpenOMCI has been started. It is responsible for starting any
+ other state machine and to initiate an ONU Capabilities report
+ """
+ if self._first_in_sync:
+ self._first_in_sync = False
+
+ # Start up the ONU Capabilities task
+ self._configuration.reset()
+
+ # Insure that the ONU-G Administrative lock is disabled
+ def failure(reason):
+ self.log.error('disable-admin-state-lock', reason=reason)
+
+ frame = OntGFrame(attributes={'administrative_state': 0}).set()
+ task = OmciModifyRequest(self._omci_agent, self.device_id, frame)
+ self.task_runner.queue_task(task).addErrback(failure)
+
+ # Start up any other remaining OpenOMCI state machines
+ def start_state_machines(machines):
+ for sm in machines:
+ self._state_machines.append(sm)
+ reactor.callLater(0, sm.start)
+
+ self._deferred = reactor.callLater(0, start_state_machines,
+ self._on_sync_state_machines)
+
+ # if an ongoing upgrading is not accomplished, restart it
+ if self._img_deferred is not None:
+ self._image_agent.onu_bootup()
+
+ def first_in_capabilities_event(self):
+ """
+ This event is called on the first capabilities event after
+ OpenOMCI has been started. It is responsible for starting any
+ other state machine. These are often state machines that have tasks
+ that are dependent upon knowing if various MEs are supported
+ """
+ if self._first_capabilities:
+ self._first_capabilities = False
+
+ # Start up any other remaining OpenOMCI state machines
+ def start_state_machines(machines):
+ for sm in machines:
+ self._state_machines.append(sm)
+ reactor.callLater(0, sm.start)
+
+ self._deferred = reactor.callLater(0, start_state_machines,
+ self._on_capabilities_state_machines)
+
+ # def __on_omci_download_success(self, image_download):
+ # self.log.debug("__on_omci_download_success", image=image_download)
+ # self._omci_upgrade_deferred = None
+ # # self._ret_deferred = None
+ # self._omci_activate_deferred = self._image_agent.activate_onu_image(image_download.name)
+ # self._omci_activate_deferred.addCallbacks(self.__on_omci_image_activate_success,
+ # self.__on_omci_image_activate_fail, errbackArgs=(image_name,))
+ # return image_name
+
+ # def __on_omci_download_fail(self, fail, image_name):
+ # self.log.debug("__on_omci_download_fail", failure=fail, image_name=image_name)
+ # self.reactor.callLater(0, self._img_deferred.errback, fail)
+ # self._omci_upgrade_deferred = None
+ # self._img_deferred = None
+
+ def __on_omci_image_activate_success(self, image_name):
+ self.log.debug("__on_omci_image_activate_success", image_name=image_name)
+ self._omci_activate_deferred = None
+ self._img_deferred.callback(image_name)
+ self._img_deferred = None
+ return image_name
+
+ def __on_omci_image_activate_fail(self, fail, image_name):
+ self.log.debug("__on_omci_image_activate_fail", faile=fail, image_name=image_name)
+ self._omci_activate_deferred = None
+ self._img_deferred.errback(fail)
+ self._img_deferred = None
+
+ def _publish_device_status_event(self):
+ """
+ Publish the ONU Device start/start status.
+ """
+ topic = OnuDeviceEntry.event_bus_topic(self.device_id,
+ OnuDeviceEvents.DeviceStatusEvent)
+ msg = {ACTIVE_KEY: self._started}
+ self.event_bus.publish(topic=topic, msg=msg)
+
+ def publish_omci_capabilities_event(self):
+ """
+ Publish the ONU Device start/start status.
+ """
+ if self.first_in_capabilities_event:
+ self.first_in_capabilities_event()
+
+ topic = OnuDeviceEntry.event_bus_topic(self.device_id,
+ OnuDeviceEvents.OmciCapabilitiesEvent)
+ msg = {
+ SUPPORTED_MESSAGE_ENTITY_KEY: self.omci_capabilities.supported_managed_entities,
+ SUPPORTED_MESSAGE_TYPES_KEY: self.omci_capabilities.supported_message_types
+ }
+ self.event_bus.publish(topic=topic, msg=msg)
+
+ def delete(self):
+ """
+ Stop the ONU Device's state machine and remove the ONU, and any related
+ OMCI state information from the OpenOMCI Framework
+ """
+ self.stop()
+ self.mib_synchronizer.delete()
+
+ # OpenOMCI cleanup
+ if self._omci_agent is not None:
+ self._omci_agent.remove_device(self._device_id, cleanup=True)
+
+ def query_mib(self, class_id=None, instance_id=None, attributes=None):
+ """
+ Get MIB database information.
+
+ This method can be used to request information from the database to the detailed
+ level requested
+
+ :param class_id: (int) Managed Entity class ID
+ :param instance_id: (int) Managed Entity instance
+ :param attributes: (list or str) Managed Entity instance's attributes
+
+ :return: (dict) The value(s) requested. If class/inst/attribute is
+ not found, an empty dictionary is returned
+ :raises DatabaseStateError: If the database is not enabled
+ """
+ self.log.debug('query', class_id=class_id, instance_id=instance_id,
+ attributes=attributes)
+
+ return self.mib_synchronizer.query_mib(class_id=class_id, instance_id=instance_id,
+ attributes=attributes)
+
+ def query_mib_single_attribute(self, class_id, instance_id, attribute):
+ """
+ Get MIB database information for a single specific attribute
+
+ This method can be used to request information from the database to the detailed
+ level requested
+
+ :param class_id: (int) Managed Entity class ID
+ :param instance_id: (int) Managed Entity instance
+ :param attribute: (str) Managed Entity instance's attribute
+
+ :return: (varies) The value requested. If class/inst/attribute is
+ not found, None is returned
+ :raises DatabaseStateError: If the database is not enabled
+ """
+ self.log.debug('query-single', class_id=class_id,
+ instance_id=instance_id, attributes=attribute)
+ assert isinstance(attribute, basestring), \
+ 'Only a single attribute value can be retrieved'
+
+ entry = self.mib_synchronizer.query_mib(class_id=class_id,
+ instance_id=instance_id,
+ attributes=attribute)
+
+ return entry[attribute] if attribute in entry else None
+
+ def query_alarm_table(self, class_id=None, instance_id=None):
+ """
+ Get Alarm information
+
+ This method can be used to request information from the alarm database to
+ the detailed level requested
+
+ :param class_id: (int) Managed Entity class ID
+ :param instance_id: (int) Managed Entity instance
+
+ :return: (dict) The value(s) requested. If class/inst/attribute is
+ not found, an empty dictionary is returned
+ :raises DatabaseStateError: If the database is not enabled
+ """
+ self.log.debug('query', class_id=class_id, instance_id=instance_id)
+
+ return self.alarm_synchronizer.query_mib(class_id=class_id, instance_id=instance_id)
+
+ def reboot(self,
+ flags=RebootFlags.Reboot_Unconditionally,
+ timeout=OmciRebootRequest.DEFAULT_REBOOT_TIMEOUT):
+ """
+ Request a reboot of the ONU
+
+ :param flags: (RebootFlags) Reboot condition
+ :param timeout: (int) Reboot task priority
+ :return: (deferred) Fires upon completion or error
+ """
+ assert self.active, 'This device is not active'
+
+ return self.task_runner.queue_task(OmciRebootRequest(self._omci_agent,
+ self.device_id,
+ flags=flags,
+ timeout=timeout))
+
+ # def get_imagefile(self, local_name, local_dir, remote_url=None):
+ # """
+ # Return a Deferred that will be triggered if the file is locally available
+ # or downloaded successfully
+ # """
+ # self.log.info('start download from {}'.format(remote_url))
+
+ # # for debug purpose, start runner here to queue downloading task
+ # # self._runner.start()
+
+ # return self._image_agent.get_image(self._image_download)
+
+ def do_onu_software_download(self, image_dnld):
+ """
+ image_dnld: (ImageDownload)
+ : Return a Deferred that will be triggered when upgrading results in success or failure
+ """
+ self.log.debug('do_onu_software_download')
+ image_download = deepcopy(image_dnld)
+ # self._img_download_deferred = self._image_agent.get_image(self._image_download)
+ # self._img_download_deferred.addCallbacks(self.__on_download_success, self.__on_download_fail, errbackArgs=(self._image_download,))
+ # self._ret_deferred = defer.Deferred()
+ # return self._ret_deferred
+ return self._image_agent.get_image(image_download)
+
+ # def do_onu_software_switch(self):
+ def do_onu_image_activate(self, image_dnld_name):
+ """
+ Return a Deferred that will be triggered when switching software image results in success or failure
+ """
+ if self._img_deferred is None:
+ self.log.debug('do_onu_image_activate')
+ self._img_deferred = defer.Deferred()
+ self._omci_upgrade_deferred = self._image_agent.onu_omci_download(image_dnld_name)
+ self._omci_upgrade_deferred.addCallbacks(self.__on_omci_image_activate_success,
+ self.__on_omci_image_activate_fail, errbackArgs=(image_dnld_name,))
+ return self._img_deferred
+
+ def cancel_onu_software_download(self, image_name):
+ self.log.debug('cancel_onu_software_download')
+ self._image_agent.cancel_download_image(image_name)
+ self._image_agent.cancel_upgrade_onu()
+ if self._img_deferred and not self._img_deferred.called:
+ self._img_deferred.cancel()
+ self._img_deferred = None
+ # self._image_download = None
+
+ def get_image_download_status(self, image_name):
+ return self._image_agent.get_image_status(image_name)
+