Source code for rastervision.pipeline.pipeline_config
from os.path import join
from typing import TYPE_CHECKING, Optional, Dict
from rastervision.pipeline.config import Config, Field
from rastervision.pipeline.config import register_config
if TYPE_CHECKING:
from rastervision.pipeline.pipeline import Pipeline # noqa
[docs]@register_config('pipeline')
class PipelineConfig(Config):
"""Base class for configuring :class:`Pipelines <.Pipeline>`.
This should be subclassed to configure new Pipelines.
"""
root_uri: str = Field(
None, description='The root URI for output generated by the pipeline')
rv_config: dict = Field(
None,
description='Used to store serialized RVConfig so pipeline can '
'run in remote environment with the local RVConfig. This should '
'not be set explicitly by users -- it is only used by the runner '
'when running a remote pipeline.')
plugin_versions: Optional[Dict[str, int]] = Field(
None,
description=
('Used to store a mapping of plugin module paths to the latest '
'version number. This should not be set explicitly by users -- it is set '
'automatically when serializing and saving the config to disk.'))
[docs] def get_config_uri(self) -> str:
"""Get URI of serialized version of this PipelineConfig."""
return join(self.root_uri, 'pipeline-config.json')
[docs] def build(self, tmp_dir: str) -> 'Pipeline':
"""Return a pipeline based on this configuration.
Subclasses should override this to return an instance of the
corresponding subclass of Pipeline.
Args:
tmp_dir: root of any temporary directory to pass to pipeline
"""
from rastervision.pipeline.pipeline import Pipeline # noqa
return Pipeline(self, tmp_dir)