Source code for AYABInterface.needle_positions

"""This module provides the interface to the AYAB shield."""

_NEEDLE_POSITION_ERROR_MESSAGE = \
    "Needle position in row {} at index {} is {} but one of {} was expected."
_ROW_LENGTH_ERROR_MESSAGE = "The length of row {} is {} but {} is expected."


[docs]class NeedlePositions(object): """An interface that just focusses on the needle positions."""
[docs] def __init__(self, rows, machine): """Create a needle interface. :param list rows: a list of lists of :attr:`needle positions <AYABInterface.machines.Machine.needle_positions>` :param AYABInterface.machines.Machine: the machine type to use :raises ValueError: if the arguments are not valid, see :meth:`check` """ self._rows = rows self._machine = machine self._completed_rows = [] self._on_row_completed = [] self.check()
[docs] def check(self): """Check for validity. :raises ValueError: - if not all lines are as long as the :attr:`number of needles <AYABInterface.machines.Machine.number_of_needles>` - if the contents of the rows are not :attr:`needle positions <AYABInterface.machines.Machine.needle_positions>` """ # TODO: This violates the law of demeter. # The architecture should be changed that this check is either # performed by the machine or by the unity of machine and # carriage. expected_positions = self._machine.needle_positions expected_row_length = self._machine.number_of_needles for row_index, row in enumerate(self._rows): if len(row) != expected_row_length: message = _ROW_LENGTH_ERROR_MESSAGE.format( row_index, len(row), expected_row_length) raise ValueError(message) for needle_index, needle_position in enumerate(row): if needle_position not in expected_positions: message = _NEEDLE_POSITION_ERROR_MESSAGE.format( row_index, needle_index, repr(needle_position), ", ".join(map(repr, expected_positions))) raise ValueError(message)
# the Content interface @property def machine(self): """The machine these positions are on.""" return self._machine
[docs] def get_row(self, index, default=None): """Return the row at the given index or the default value.""" if not isinstance(index, int) or index < 0 or index >= len(self._rows): return default return self._rows[index]
[docs] def row_completed(self, index): """Mark the row at index as completed. .. seealso:: :meth:`completed_row_indices` This method notifies the obsevrers from :meth:`on_row_completed`. """ self._completed_rows.append(index) for row_completed in self._on_row_completed: row_completed(index)
# end of the Content interface @property def completed_row_indices(self): """The indices of the completed rows. :rtype: list When a :meth:`row was completed <row_completed>`, the index of the row turns up here. The order is preserved, entries may occur duplicated. """ return self._completed_rows.copy()
[docs] def on_row_completed(self, callable): """Add an observer for completed rows. :param callable: a callable that is called with the row index as first argument When :meth:`row_completed` was called, this :paramref:`callable` is called with the row index as first argument. Call this method several times to register more observers. """ self._on_row_completed.append(callable)
__all__ = ["NeedlePositions"]