"""This module is used to communicate with the shield.
Requirement: Make objects from binary stuff.
"""
from .hardware_messages import read_message_type, ConnectionClosed
from .states import WaitingForStart
from .cache import NeedlePositionCache
from threading import RLock, Thread
from itertools import chain
from time import sleep
[docs]class Communication(object):
"""This class comunicates with the AYAB shield."""
[docs] def __init__(self, file, get_needle_positions, machine,
on_message_received=(), left_end_needle=None,
right_end_needle=None):
"""Create a new Communication object.
:param file: a file-like object with read and write methods for the
communication with the Arduino. This could be a
:class:`serial.Serial` or a :meth:`socket.socket.makefile`.
:param get_needle_positions: a callable that takes an :class:`index
<int>` and returns :obj:`None` or an iterable over needle positions.
:param AYABInterface.machines.Machine machine: the machine to use for
knitting
:param list on_message_received: an iterable over callables that takes
a received :class:`message
<AYABInterface.communication.hardware_messages.Message>` whenever
it comes in. Since :attr:`state` changes only take place when a
message is received, this can be used as an state observer.
:param left_end_needle: A needle number on the machine.
Other needles that are on the left side of this needle are not used
for knitting. Their needle positions are not be set.
:param right_end_needle: A needle number on the machine.
Other needles that are on the right side of this needle are not used
for knitting. Their needle positions are not be set.
"""
self._file = file
self._on_message_received = on_message_received
self._machine = machine
self._state = WaitingForStart(self)
self._controller = None
self._last_requested_line_number = 0
self._needle_positions_cache = NeedlePositionCache(
get_needle_positions, self._machine)
self._left_end_needle = (
machine.left_end_needle
if left_end_needle is None else left_end_needle)
self._right_end_needle = (
machine.right_end_needle
if right_end_needle is None else right_end_needle)
self._lock = RLock()
self._thread = None
self._number_of_threads_receiving_messages = 0
self._on_message = []
@property
def needle_positions(self):
"""A cache for the needle positions.
:rtype: AYABInterface.communication.cache.NeedlePositionCache
"""
return self._needle_positions_cache
[docs] def start(self):
"""Start the communication about a content.
:param Content content: the content of the communication.
"""
with self.lock:
self._state.communication_started()
_read_message_type = staticmethod(read_message_type)
def _message_received(self, message):
"""Notify the observers about the received message."""
with self.lock:
self._state.receive_message(message)
for callable in chain(self._on_message_received, self._on_message):
callable(message)
[docs] def on_message(self, callable):
"""Add an observer to received messages.
:param callable: a callable that is called every time a
:class:`AYABInterface.communication.host_messages.Message` is sent or
a :class:`AYABInterface.communication.controller_messages.Message` is
received
"""
self._on_message.append(callable)
[docs] def receive_message(self):
"""Receive a message from the file."""
with self.lock:
assert self.can_receive_messages()
message_type = self._read_message_type(self._file)
message = message_type(self._file, self)
self._message_received(message)
[docs] def can_receive_messages(self):
"""Whether tihs communication is ready to receive messages.]
:rtype: bool
.. code:: python
assert not communication.can_receive_messages()
communication.start()
assert communication.can_receive_messages()
communication.stop()
assert not communication.can_receive_messages()
"""
with self.lock:
return not self._state.is_waiting_for_start() and \
not self._state.is_connection_closed()
[docs] def stop(self):
"""Stop the communication with the shield."""
with self.lock:
self._message_received(ConnectionClosed(self._file, self))
[docs] def api_version_is_supported(self, api_version):
"""Return whether an api version is supported by this class.
:rtype: bool
:return: if the :paramref:`api version <api_version>` is supported
:param int api_version: the api version
Currently supported api versions: ``4``
"""
return api_version == 4
[docs] def send(self, host_message_class, *args):
"""Send a host message.
:param type host_message_class: a subclass of
:class:`AYABImterface.communication.host_messages.Message`
:param args: additional arguments that shall be passed to the
:paramref:`host_message_class` as arguments
"""
message = host_message_class(self._file, self, *args)
with self.lock:
message.send()
for callable in self._on_message:
callable(message)
@property
def state(self):
"""The state this object is in.
:return: the state this communication object is in.
:rtype: AYABInterface.communication.states.State
.. note:: When calling :meth:`parallelize` the state can change while
you check it.
"""
return self._state
@property
def lock(self):
"""The lock of the communication.
In case you :meth:`parallelize` the communication, you may want to use
this :class:`lock <threading.RLock>` to make shure the parallelization
does not break your code.
"""
return self._lock
@state.setter
def state(self, new_state):
"""Set the state."""
with self.lock:
self._state.exit()
self._state = new_state
self._state.enter()
@property
def left_end_needle(self):
"""The left end needle of the needle positions.
:rtype: int
:return: the :attr:`left end needle of the machine
<AYABInterface.machine.Machine.left_end_needle>`
"""
return self._left_end_needle
@property
def right_end_needle(self):
"""The left end needle of the needle positions.
:rtype: int
:return: the :attr:`right end needle of the machine
<AYABInterface.machine.Machine.right_end_needle>`
"""
return self._right_end_needle
@property
def controller(self):
"""Information about the controller.
If no information about the controller is received, the return value
is :obj:`None`.
If information about the controller is known after :ref:`cnfinfo` was
received, you can access these values:
.. code:: python
>>> communication.controller.firmware_version
(5, 2)
>>> communication.controller.firmware_version.major
5
>>> communication.controller.firmware_version.minor
2
>>> communication.controller.api_version
4
"""
return self._controller
@controller.setter
def controller(self, value):
self._controller = value
@property
def last_requested_line_number(self):
"""The number of the last line that was requested.
:rtype: int
:return: the last requested line number or ``0``
"""
return self._last_requested_line_number
@last_requested_line_number.setter
def last_requested_line_number(self, line_number):
"""Set the last requested line number."""
self._last_requested_line_number = line_number
[docs] def parallelize(self, seconds_to_wait=2):
"""Start a parallel thread for receiving messages.
If :meth:`start` was no called before, start will be called in the
thread.
The thread calls :meth:`receive_message` until the :attr:`state`
:meth:`~AYABInterface.communication.states.State.is_connection_closed`.
:param float seconds_to_wait: A time in seconds to wait with the
parallel execution. This is useful to allow the controller time to
initialize.
.. seealso:: :attr:`lock`, :meth:`runs_in_parallel`
"""
with self.lock:
thread = Thread(target=self._parallel_receive_loop,
args=(seconds_to_wait,))
thread.deamon = True
thread.start()
self._thread = thread
def _parallel_receive_loop(self, seconds_to_wait):
"""Run the receiving in parallel."""
sleep(seconds_to_wait)
with self._lock:
self._number_of_threads_receiving_messages += 1
try:
with self._lock:
if self.state.is_waiting_for_start():
self.start()
while True:
with self.lock:
if self.state.is_connection_closed():
return
self.receive_message()
finally:
with self._lock:
self._number_of_threads_receiving_messages -= 1
[docs] def runs_in_parallel(self):
"""Whether the communication runs in parallel.
:rtype: bool
:return: whether :meth:`parallelize` was called and the communication
still receives messages and is not stopped
"""
return self._number_of_threads_receiving_messages != 0
__all__ = ["Communication"]