Source code for pacman.executor.algorithm_classes.external_algorithm
# Copyright (c) 2017-2019 The University of Manchester
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import subprocess
from spinn_utilities.overrides import overrides
from spinn_utilities.progress_bar import ProgressBar
from pacman.exceptions import PacmanExternalAlgorithmFailedToCompleteException
from .abstract_algorithm import AbstractAlgorithm
[docs]class ExternalAlgorithm(AbstractAlgorithm):
""" An algorithm which is external to the SpiNNaker software, or rather\
its wrapper into PACMAN.
"""
__slots__ = [
# The command line to call
"_command_line_arguments"
]
def __init__(
self, algorithm_id, required_inputs, optional_inputs, outputs,
required_input_tokens, optional_input_tokens,
generated_output_tokens, command_line_arguments):
# pylint: disable=too-many-arguments
super(ExternalAlgorithm, self).__init__(
algorithm_id, required_inputs, optional_inputs, outputs,
required_input_tokens, optional_input_tokens,
generated_output_tokens)
self._command_line_arguments = command_line_arguments
[docs] @overrides(AbstractAlgorithm.call)
def call(self, inputs):
# Get the inputs to pass as the arguments
arg_inputs = self._get_inputs(inputs)
# Convert the arguments using the inputs
args = [
arg.format(**arg_inputs) for arg in self._command_line_arguments
]
algorithm_progress_bar = ProgressBar(
1, "Running external algorithm {}".format(self._algorithm_id))
# Run the external command
child = subprocess.Popen(
args, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
stdin=subprocess.PIPE)
child.wait()
algorithm_progress_bar.update(1)
algorithm_progress_bar.end()
# Detect any errors
if child.returncode != 0:
stdout, stderr = child.communicate()
raise PacmanExternalAlgorithmFailedToCompleteException(
"Algorithm {} returned a non-zero error code {}\n"
" Inputs: {}\n"
" Output: {}\n"
" Error: {}\n".format(
self._algorithm_id, child.returncode,
inputs.keys(), stdout, stderr))
# Return the results processed into a dict
# Use None here as the results don't actually exist, and are expected
# to be obtained from a file, whose name is in inputs
return self._get_outputs(inputs, [None] * len(self._outputs))
def __repr__(self):
return (
"ExternalAlgorithm(algorithm_id={},"
" required_inputs={}, optional_inputs={}, outputs={}"
" command_line_arguments={})".format(
self._algorithm_id, self._required_inputs,
self._optional_inputs, self._outputs,
self._command_line_arguments))