docs

spalloc_client/protocol_client.py

@@ -18,8 +18,11 @@
 import errno
 import json
 import socket
-from typing import Dict, Optional
+from typing import Dict, List, Optional
 from threading import current_thread, RLock, local
+
+from spinn_utilities.typing.json import JsonObject
+
 from spalloc_client._utils import time_left, timed_out, make_timeout
 
 
@@ -199,7 +202,7 @@ def close(self):
             self._close(key)
         self._local = _ProtocolThreadLocal()
 
-    def _recv_json(self, timeout=None):
+    def _recv_json(self, timeout=None) -> JsonObject:
         """ Receive a line of JSON from the server.
 
         Parameters
@@ -358,63 +361,93 @@ def wait_for_notification(self, timeout=None):
     # The bindings of the Spalloc protocol methods themselves; simplifies use
     # from IDEs.
 
-    def version(self, timeout=None):  # pragma: no cover
+    def version(self, timeout: Optional[int] = None) -> JsonObject:
+        """ Ask what version of spalloc is running. """
         return self.call("version", timeout=timeout)
 
-    def create_job(self, *args, **kwargs):  # pragma: no cover
+    def create_job(self, *args: List[object],
+                   **kwargs: Dict[str, object]) -> JsonObject:
+        """
+        Start a new job
+        """
         # If no owner, don't bother with the call
         if "owner" not in kwargs:
             raise SpallocServerException(
                 "owner must be specified for all jobs.")
         return self.call("create_job", *args, **kwargs)
 
-    def job_keepalive(self, job_id, timeout=None):  # pragma: no cover
+    def job_keepalive(self, job_id: int,
+                      timeout: Optional[int] = None) -> JsonObject:
+        """
+        Send s message to keep the job alive.
+
+        Without these the job will be killed after a while.
+        """
         return self.call("job_keepalive", job_id, timeout=timeout)
 
-    def get_job_state(self, job_id, timeout=None):  # pragma: no cover
+    def get_job_state(self, job_id: int,
+                      timeout: Optional[int] = None) -> JsonObject:
+        """Get the state for this job """
         return self.call("get_job_state", job_id, timeout=timeout)
 
-    def get_job_machine_info(self, job_id, timeout=None):  # pragma: no cover
+    def get_job_machine_info(self, job_id: int,
+                             timeout: Optional[int] = None) -> JsonObject:
+        """ Get info for this job. """
         return self.call("get_job_machine_info", job_id, timeout=timeout)
 
-    def power_on_job_boards(self, job_id, timeout=None):  # pragma: no cover
+    def power_on_job_boards(self, job_id: int,
+                             timeout: Optional[int] = None) -> JsonObject:
+        """ Turn on the power on the jobs baords. """
         return self.call("power_on_job_boards", job_id, timeout=timeout)
 
-    def power_off_job_boards(self, job_id, timeout=None):  # pragma: no cover
+    def power_off_job_boards(self, job_id: int,
+                             timeout: Optional[int] = None) -> JsonObject:
+        """ Turn off the power on the jobs baords. """
         return self.call("power_off_job_boards", job_id, timeout=timeout)
 
-    def destroy_job(self, job_id, reason=None,
-                    timeout=None):  # pragma: no cover
+    def destroy_job(self, job_id: int, reason: Optional[str] = None,
+                    timeout: Optional[int] = None) -> JsonObject:
+        """ Destroy the job """
         return self.call("destroy_job", job_id, reason, timeout=timeout)
 
-    def notify_job(self, job_id=None, timeout=None):  # pragma: no cover
+    def notify_job(self, job_id: Optional[int] = None,
+                      timeout: Optional[int]=None) -> JsonObject:
+        """ Turn on notification of job status changes. """
         return self.call("notify_job", job_id, timeout=timeout)
 
-    def no_notify_job(self, job_id=None, timeout=None):  # pragma: no cover
+    def no_notify_job(self, job_id: Optional[int] = None,
+                      timeout: Optional[int]=None) -> JsonObject:
+        """ Turn off notification of job status changes. """
         return self.call("no_notify_job", job_id, timeout=timeout)
 
-    def notify_machine(self, machine_name=None,
-                       timeout=None):  # pragma: no cover
+    def notify_machine(self, machine_name: Optional[str] = None,
+                       timeout: Optional[int] = None) -> JsonObject:
+        """ Turn on notification of machine status changes. """
         return self.call("notify_machine", machine_name, timeout=timeout)
 
-    def no_notify_machine(self, machine_name=None,
-                          timeout=None):  # pragma: no cover
+    def no_notify_machine(self, machine_name: Optional[str] = None,
+                          timeout: Optional[int] = None) -> JsonObject:
+        """ Turn off notification of machine status changes. """
         return self.call("no_notify_machine", machine_name, timeout=timeout)
 
-    def list_jobs(self, timeout=None):  # pragma: no cover
+    def list_jobs(self, timeout: Optional[int] = None) -> JsonObject:
+        """ Obtains a list of jobs currently running. """
         return self.call("list_jobs", timeout=timeout)
 
-    def list_machines(self, timeout=None):  # pragma: no cover
+    def list_machines(self, timeout: Optional[float] = None) -> JsonObject:
+        """ Obtains a list of currently supported machines. """
         return self.call("list_machines", timeout=timeout)
 
-    def get_board_position(self, machine_name, x, y, z,
-                           timeout=None):  # pragma: no cover
+    def get_board_position(self, machine_name: str, x: int, y: int, z: int,
+                           timeout: Optional[int] = None):  # pragma: no cover
+        """ Gets the position of board x, y, z on the given machine. """
         # pylint: disable=too-many-arguments
         return self.call("get_board_position", machine_name, x, y, z,
                          timeout=timeout)
 
-    def get_board_at_position(self, machine_name, x, y, z,
-                              timeout=None):  # pragma: no cover
+    def get_board_at_position(self, machine_name: str, x: int, y: int, z: int,
+                              timeout: Optional[int] = None) -> JsonObject:  # pragma: no cover
+        """ Gets the board x, y, z on the requested machine. """
         # pylint: disable=too-many-arguments
         return self.call("get_board_at_position", machine_name, x, y, z,
                          timeout=timeout)
@@ -426,7 +459,8 @@ def get_board_at_position(self, machine_name, x, y, z,
         frozenset("job_id chip_x chip_y".split())])
 
     def where_is(self, timeout: Optional[int] = None,
-                 **kwargs: Dict[str, object]):
+                 **kwargs: Dict[str, object]) -> JsonObject:
+        """ Reports where ion the Machine a job is running """
         # Test for whether sane arguments are passed.
         keywords = frozenset(kwargs)
         if keywords not in ProtocolClient._acceptable_kwargs_for_where_is:

spalloc_client/scripts/machine.py

@@ -33,7 +33,9 @@
 from collections import defaultdict
 import argparse
 import sys
-from spalloc_client import __version__
+from typing import List
+
+from spalloc_client import __version__, ProtocolClient
 from spalloc_client.term import (
     Terminal, render_table, render_definitions, render_boards, render_cells,
     DEFAULT_BOARD_EDGES)
@@ -211,6 +213,9 @@ def show_machine(t, machines, jobs, machine_name, compact=False):
 
 
 class ListMachinesScript(Script):
+    """
+    A Script object to get information from a spalloc machine.
+    """
 
     def __init__(self):
         super().__init__()
@@ -250,12 +255,18 @@ def verify_arguments(self, args):
             self.parser.error(
                 "--detailed only works when a specific machine is specified")
 
-    def one_shot(self, client, args):
+    def one_shot(self,  client: ProtocolClient, args: List[object]):
+        """
+        Display the machine info once
+        """
         t = Terminal()
         # Get all information and display accordingly
         self.get_and_display_machine_info(client, args, t)
 
-    def recurring(self, client, args):
+    def recurring(self, client: ProtocolClient, args: List[object]):
+        """
+        Repeatedly display the machine info
+        """
         t = Terminal()
         while True:
             client.notify_machine(args.machine, timeout=args.timeout)

spalloc_client/scripts/ps.py

@@ -27,7 +27,9 @@
 """
 import argparse
 import sys
-from spalloc_client import __version__, JobState
+from typing import List
+
+from spalloc_client import __version__, JobState, ProtocolClient
 from spalloc_client.term import Terminal, render_table
 from spalloc_client._utils import render_timestamp
 from .support import Script
@@ -113,6 +115,9 @@ def render_job_list(t, jobs, args):
 
 
 class ProcessListScript(Script):
+    """
+    An object form Job scripts.
+    """
     def get_parser(self, cfg):
         parser = argparse.ArgumentParser(description="List all active jobs.")
         parser.add_argument(
@@ -128,12 +133,14 @@ def get_parser(self, cfg):
             help="list only jobs belonging to a particular owner")
         return parser
 
-    def one_shot(self, client, args):
+    def one_shot(self, client: ProtocolClient, args: List[object]):
+        """ Gets info on the job list once. """
         t = Terminal(stream=sys.stderr)
         jobs = client.list_jobs(timeout=args.timeout)
         print(render_job_list(t, jobs, args))
 
-    def recurring(self, client, args):
+    def recurring(self, client: ProtocolClient, args: List[object]):
+        """ Repeatedly gets info on the job list. """
         client.notify_job(timeout=args.timeout)
         t = Terminal(stream=sys.stderr)
         while True:

spalloc_client/scripts/support.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 import sys
-from typing import List
+from typing import Any, Dict, List, Optional, Tuple
 from spalloc_client import (
     config, ProtocolClient, ProtocolError, ProtocolTimeoutError,
     SpallocServerException)
@@ -25,7 +25,10 @@
 
 
 class Terminate(Exception):
-    def __init__(self, code, *args):
+    def __init__(self, code: int, *args: Tuple[object]):
+        """
+        An Exception that can be used to exit the program.
+        """
         super().__init__()
         self._code = code
         args = list(args)
@@ -38,19 +41,24 @@ def __init__(self, code, *args):
             self._msg = message
 
     def exit(self):
+        """ Exit the program after priintg and erro msg. """
         if self._msg is not None:
             sys.stderr.write(self._msg + "\n")
         sys.exit(self._code)
 
 
-def version_verify(client, timeout):
+def version_verify(client: ProtocolClient, timeout: Optional[int]):
+    """
+    Verify that the current version of the client is compatible
+    """
     version = tuple(map(int, client.version(timeout=timeout).split(".")))
     if not (VERSION_RANGE_START <= version < VERSION_RANGE_STOP):
         raise Terminate(2, "Incompatible server version ({})",
                         ".".join(map(str, version)))
 
 
 class Script(object):
+    """ Base class of various Scopt Objects. """
     def __init__(self):
         self.client_factory = ProtocolClient
 
@@ -69,7 +77,13 @@ def body(self, client: ProtocolClient, args: List[str]):
             obtained and verified to be compatible.
         """
 
-    def build_server_arg_group(self, server_args, cfg):
+    def build_server_arg_group(self, server_args: Any,
+                               cfg: Dict[str, object]):
+        """
+        Adds a few more arguements
+
+        :param argparse._ArguementGroup server_args:
+        """
         server_args.add_argument(
             "--hostname", "-H", default=cfg["hostname"],
             help="hostname or IP of the spalloc server (default: %(default)s)")

spalloc_client/scripts/where_is.py

@@ -72,6 +72,9 @@
 
 
 class WhereIsScript(Script):
+    """
+    An script object to find where a board is
+    """
 
     def __init__(self):
         super().__init__()