Source code for rastervision.pipeline.runner.inprocess_runner
from rastervision.pipeline.cli import _run_command
from rastervision.pipeline.runner.runner import Runner
INPROCESS = 'inprocess'
[docs]class InProcessRunner(Runner):
"""Runs each command sequentially within a single process.
Useful for testing and debugging.
"""
[docs] def run(self,
cfg_json_uri,
pipeline,
commands,
num_splits=1,
pipeline_run_name: str = 'raster-vision'):
for command in commands:
if command in pipeline.split_commands and num_splits > 1:
for split_ind in range(num_splits):
_run_command(cfg_json_uri, command, split_ind, num_splits)
else:
_run_command(cfg_json_uri, command, 0, 1)
[docs] def run_command(self, cmd: list[str]):
raise NotImplementedError(
'Use LocalRunner.run_command to run a command locally.')