Source code for forml.runtime._pseudo

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

Special lightweight launcher for dummy execution.
import logging
import types
import typing

from forml import io
from forml import project as prjmod
from forml import setup
from import asset
from forml.provider.registry.filesystem import volatile

from ..pipeline import payload
from . import _pad

if typing.TYPE_CHECKING:
    from forml import flow, project, runtime  # pylint: disable=reimported
    from import dsl

LOGGER = logging.getLogger(__name__)

[docs]class Virtual: """Custom launcher allowing to execute the provided artifact using the default or an explicit runner in combination with a special pipeline sink to capture and return any output produced by the given action. All states produced while executing the launcher actions are (temporarily) persisted in an internal model registry implemented by the :class:`Volatile provider <forml.provider.registry.filesystem.volatile.Registry>`. Args: artifact: Project artifact to be launched. The available launcher actions are exposed using the following common triggers: Methods: train(lower=None, upper=None) -> runtime.Virtual.Trained: Trigger the *train* action. Returns: Accessor of the train-mode features/outcomes outputs. tune(lower=None, upper=None) -> None: Trigger the *tune* action. apply(lower=None, upper=None) -> flow.Features: Trigger the *apply* action. Returns: The apply-mode output. eval(lower=None, upper=None) -> float: Trigger the (train-test) *eval* action. Returns: Evaluation metric value. Furthermore, these triggers can be accessed using three different approaches: #. Directly on the launcher instance: >>> launcher_instance.eval() 0.83 #. Using a *getitem* syntax specifying an explicit *runner* provider: >>> launcher_instance['graphviz'].train() #. Using a *call* syntax specifying both an explicit *runner* and a list of source *feeds* providers: >>> launcher_instance('dask', ['openlake']).apply() [0.31, 0.63, 0.16, 0.87] """
[docs] class Trained: """Lazy accessor of the virtual train-mode features/outcomes segment outputs as returned by the :meth:`runtime.Virtual.train <forml.runtime.Virtual.train>` method.""" def __init__(self, future: payload.Sniff.Value.Future): self._future: payload.Sniff.Value.Future = future def __getattr__(self, item: typing.Any) -> typing.Any: return getattr(self._future.result(), item) @property def features(self) -> 'flow.Features': """Get the train-mode *features* segment values.""" return self._future.result()[0] @property def labels(self) -> 'flow.Labels': """Get the train-mode *labels* segment values.""" return self._future.result()[1]
class Handler: """Wrapper for selected launcher parameters.""" def __init__(self, launcher: _pad.Launcher, sniffer: payload.Sniff): self._launcher: _pad.Launcher = launcher self._sniffer: payload.Sniff = sniffer def _run( self, mode: 'runtime.Launcher.Mode', lower: typing.Optional['dsl.Native'] = None, upper: typing.Optional['dsl.Native'] = None, ) -> payload.Sniff.Value.Future: with self._sniffer as future: mode(lower, upper) return future def apply( self, lower: typing.Optional['dsl.Native'] = None, upper: typing.Optional['dsl.Native'] = None ) -> 'flow.Features': """Trigger the *apply* action. See Also: Full description in the Virtual class docstring. """ future = self._run(self._launcher.apply, lower, upper) try: return future.result() except payload.Sniff.Lost as err: LOGGER.warning(err) return None def train( self, lower: typing.Optional['dsl.Native'] = None, upper: typing.Optional['dsl.Native'] = None ) -> 'runtime.Virtual.Trained': """Trigger the *train* action. See Also: Full description in the Virtual class docstring. """ return Virtual.Trained(self._run(self._launcher.train_return, lower, upper)) def eval( self, lower: typing.Optional['dsl.Native'] = None, upper: typing.Optional['dsl.Native'] = None ) -> float: """Trigger the *eval* action. See Also: Full description in the Virtual class docstring. """ future = self._run(self._launcher.eval_traintest, lower, upper) try: result = future.result() return result.features if isinstance(result, payload.Sniff.Captor.Trainset) else result except payload.Sniff.Lost as err: LOGGER.warning(err) return float('nan') @property def tune(self) -> typing.Callable[[typing.Optional['dsl.Native'], typing.Optional['dsl.Native']], None]: """Trigger the *tune* action. See Also: Full description in the Virtual class docstring. """ return self._launcher.tune class Sink(io.Sink): """Sniffer sink.""" def __init__(self, sniffer: payload.Sniff): super().__init__() self._sniffer: payload.Sniff = sniffer def save(self, schema: typing.Optional['dsl.Source.Schema']) -> 'flow.Composable': return self._sniffer def __init__(self, artifact: 'project.Artifact'): class Manifest(types.ModuleType): """Fake manifest module.""" NAME = (artifact.package or setup.PRJNAME).replace('.', '-') VERSION = '0' PACKAGE = artifact.package MODULES = artifact.modules def __init__(self): super().__init__(prjmod.Manifest.MODULE) with setup.context(Manifest()): package = prjmod.Package(artifact.path or asset.mkdtemp(prefix='virtual-')) self._project: str = self._registry: asset.Registry = volatile.Registry() self._sniffer: payload.Sniff = payload.Sniff() asset.Directory(self._registry).get(self._project).put(package) def __call__( self, runner: typing.Optional[typing.Union[setup.Runner, str]] = None, feeds: typing.Optional[typing.Iterable[typing.Union[setup.Feed, str, io.Feed]]] = None, ) -> 'runtime.Virtual.Handler': launcher = _pad.Platform(runner, self._registry, feeds, self.Sink(self._sniffer)).launcher(self._project) return self.Handler(launcher, self._sniffer) def __getitem__(self, runner: typing.Union[setup.Runner, str]) -> 'runtime.Virtual.Handler': """Convenient shortcut for selecting a specific runner using the `launcher[name]` syntax. Args: runner: Runner alias/qualname to use. Returns: Launcher handler. """ return self(runner) def __getattr__( self, mode: str ) -> typing.Callable[[typing.Optional['dsl.Native'], typing.Optional['dsl.Native']], typing.Any]: """Convenient shortcut for accessing the particular launcher mode using the ``launcher.train()`` syntax. Args: mode: Launcher mode to execute. Returns: Callable launcher handler. """ return getattr(self(), mode)