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
# 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 <>.

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): """ :param str algorithm_id: The unique ID of the algorithm :param list(AbstractInput) required_inputs: The inputs required by the algorithm :param list(AbstractInput) optional_inputs: The optional inputs for the algorithm, which will be provided when available :param list(Output) outputs: The output types of the algorithm :param list(Token) required_input_tokens: Tokens required to have been generated before this algorithm can start :param list(Token) optional_input_tokens: Tokens required to have been generated before this algorithm can start if and only if at least one algorithm generates the token :param list(Token) generated_output_tokens: Tokens generated by this algorithm :param list(str) command_line_arguments: The command line to call .. note: Each argument is passed through :py:meth:`str.format` with the available inputs as *keyword* arguments. """ # pylint: disable=too-many-arguments super().__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( 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))
[docs] @overrides(AbstractAlgorithm.write_provenance_header) def write_provenance_header(self, provenance_file): provenance_file.write("{}\n".format(self._algorithm_id)) provenance_file.write("\t{}\n".format(self._command_line_arguments))