Source code for gemseo.core.monitoring

# Copyright 2021 IRT Saint Exupéry, https://www.irt-saintexupery.com
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License version 3 as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
# Contributors:
#    INITIAL AUTHORS - API and implementation and/or documentation
#        :author: Remi Lafage
#    OTHER AUTHORS   - MACROSCOPIC CHANGES
"""Monitoring mechanism to track |g| execution (update events)."""
from __future__ import annotations

from gemseo.core.scenario import Scenario
from gemseo.utils.singleton import SingleInstancePerAttributeId


[docs]class Monitoring(metaclass=SingleInstancePerAttributeId): """This class implements the observer pattern. It is a singleton, it is called by |g| core classes like MDODicipline whenever an event of interest like a status change occurs. Client objects register with add_observer and are notified whenever a discipline status change occurs. """ def __init__(self, scenario: Scenario) -> None: """ Args: scenario: The scenario to be monitored. """ # noqa: D205, D212, D415 self._observers = [] self.workflow = scenario.get_expected_workflow() self.workflow.set_observer(self) self.workflow.enable()
[docs] def add_observer(self, observer) -> None: """Register an observer object interested in observable update events. Args: observer: The object to be notified. """ if observer not in self._observers: self._observers.append(observer)
[docs] def remove_observer(self, observer) -> None: """Unsubscribe the given observer. Args: observer: The observer to be removed. """ if observer in self._observers: self._observers.remove(observer)
[docs] def remove_all_observers(self) -> None: """Unsubscribe all observers.""" self._observers = []
[docs] def update(self, atom) -> None: """Notify the observers that the corresponding observable object is updated. Observers have to know what to retrieve from the observable object. Args: atom: The updated object. """ for obs in self._observers: obs.update(atom)
[docs] def get_statuses(self) -> dict[str, str]: """Get the statuses of all disciplines. Returns: These statuses associated with the atom ids. """ return self.workflow.get_statuses()
def __str__(self) -> str: return str(self.workflow)