Source code for gemseo_matlab.license_manager

# Copyright 2021 IRT Saint Exupéry,
# 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
# 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.
# Copyright (c) 2018 IRT-AESE.
# All rights reserved.
# Contributors:
#    INITIAL AUTHORS - API and implementation and/or documentation
#        :author: Arthur Piat
#        :author: Nicolas Roussouly: GEMSEO integration
"""Definition of the Matlab license manager.


This module contains the :class:`.LicenseManager`
which enables to check the presence of any toolbox licenses
of the Matlab installation.

from __future__ import annotations

import logging
from pathlib import Path
from typing import TYPE_CHECKING
from typing import ClassVar

from gemseo_matlab.engine import MatlabEngine

    from import Iterable

LOGGER = logging.getLogger(__name__)

[docs] class LicenseManager: """Manage Matlab License. The licenseManager was created to enable de-synchronised launch of optimization using matlab_discipline. The goal is to wait until all licenses that are needed are available in Matlab workspace. Parallel computing launch can be used with this tool. Examples: >>> # Build a new matlab engine >>> eng = get_matlab_engine() >>> # add a toolbox to the engine >>> eng.add_toolbox("signal_toolbox") >>> # build a license manager from the previous engine >>> lm = LicenseManager(eng) >>> # check licenses of the engine until all are available >>> lm.check_licenses() """ engine: MatlabEngine """The MatlabEngine instance.""" SIGNAL_TOOL: ClassVar[str] = "signal_toolbox" DISTRIB_COMP_TOOL: ClassVar[str] = "distrib_computing_toolbox" CURVE_FIT_TOOL: ClassVar[str] = "Curve_Fitting_Toolbox" def __init__(self, engine: MatlabEngine) -> None: """ Args: engine: The MatlabEngine instance. """ # noqa: D205, D212, D415 self.__engine = engine if self.__engine.is_closed: # If engine is not (re)started here, add_path # will raise an error. A closed engine could happen here # since engine is a singleton and could be closed in any # other location self.__engine.start_engine() self.__engine.add_path(str(Path(__file__).parent / "matlab_files")) self.licenses = self.__engine.get_toolboxes()
[docs] def check_licenses( self, licenses: Iterable[str] = (), pause_frac: float = 60, pause_const: float = 20, ) -> None: """Check that the Matlab licenses exist. The method fetches all the needed licenses thanks to the matlab function licenseControl and the class Logger. Note that the MATLAB function will be looping until the given toolboxes are available. Args: licenses: The matlab toolboxes. If empty, use the already existing engine licenses. pause_frac: The time used between each try to get licenses. pause_const: The time added in order to estimate the waiting time. The waiting time is estimated at each try with the following formula: ``Wt = pause_const + random([0,1])*pause_frac``. """ if licenses: self.licenses = licenses if not self.licenses:"No MATLAB license check will be performed.") return self.__engine.execute_function( "licenseControl", self.licenses, pause_frac, pause_const, nargout=0 )
[docs] def start_parallel_computing( self, n_parallel_workers: int = 4, cluster_name: MatlabEngine.ParallelType = MatlabEngine.ParallelType.LOCAL, ) -> None: """Start parallel computing in MatlabEngine. Args: n_parallel_workers: The number of "workers" to the parallel pool. Maximum number allowed is 12 cluster_name: The matlab parallel pool cluster name. """ self.check_licenses([self.DISTRIB_COMP_TOOL]) self.__engine.start_parallel_computing(n_parallel_workers, cluster_name)
[docs] def end_parallel_computing(self) -> None: """Stop parallel computing in matlab engine.""" self.__engine.end_parallel_computing()