diff --git a/.github/workflows/python_actions.yml b/.github/workflows/python_actions.yml
index d7fc27545..772c10fbd 100644
--- a/.github/workflows/python_actions.yml
+++ b/.github/workflows/python_actions.yml
@@ -26,7 +26,8 @@ jobs:
coverage-package: spalloc tests
flake8-packages: spalloc_client tests
pylint-packages: spalloc_client
- mypy-packages: spalloc_client tests
+ mypy-packages: tests
+ mypy-full_packages: spalloc_client
sphinx_directory: docs/source
secrets: inherit
diff --git a/mypy.bash b/mypy.bash
new file mode 100755
index 000000000..7eff6bc69
--- /dev/null
+++ b/mypy.bash
@@ -0,0 +1,27 @@
+#!/bin/bash
+
+# Copyright (c) 2024 The University of Manchester
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This bash assumes that other repositories are installed in paralled
+
+# requires the latest mypy
+# pip install --upgrade mypy
+
+utils="../SpiNNUtils/spinn_utilities"
+machine="../SpiNNMachine/spinn_machine"
+man="../SpiNNMan/spinnman"
+pacman="../PACMAN/pacman"
+
+mypy --python-version 3.8 $utils $machine $man $pacman spalloc_client tests
diff --git a/mypyd.bash b/mypyd.bash
new file mode 100755
index 000000000..19a185204
--- /dev/null
+++ b/mypyd.bash
@@ -0,0 +1,28 @@
+#!/bin/bash
+
+# Copyright (c) 2024 The University of Manchester
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This bash assumes that other repositories are installed in paralled
+
+# requires the latest mypy
+# pip install --upgrade mypy
+
+utils="../SpiNNUtils/spinn_utilities"
+machine="../SpiNNMachine/spinn_machine"
+man="../SpiNNMan/spinnman"
+pacman="../PACMAN/pacman"
+
+mypy --python-version 3.8 --disallow-untyped-defs $utils $machine $man $pacman spalloc_client
+
diff --git a/spalloc_client/__init__.py b/spalloc_client/__init__.py
index 2d08687a8..3afe7fd1e 100644
--- a/spalloc_client/__init__.py
+++ b/spalloc_client/__init__.py
@@ -23,5 +23,5 @@
__all__ = [
"Job", "JobDestroyedError", "JobState", "ProtocolClient",
- "ProtocolError", "ProtocolTimeoutError", "SpallocServerException",
- "StateChangeTimeoutError"]
+ "ProtocolError", "ProtocolTimeoutError",
+ "SpallocServerException", "StateChangeTimeoutError"]
diff --git a/spalloc_client/_keepalive_process.py b/spalloc_client/_keepalive_process.py
index 3ca225bc6..ae49c0199 100644
--- a/spalloc_client/_keepalive_process.py
+++ b/spalloc_client/_keepalive_process.py
@@ -17,10 +17,11 @@
"""
import sys
import threading
+from typing import List
from spalloc_client.protocol_client import ProtocolClient, ProtocolTimeoutError
-def wait_for_exit(stop_event):
+def wait_for_exit(stop_event: threading.Event) -> None:
""" Listens to stdin for a line equal to 'exit' or end-of-file and then\
notifies the given event (that it is time to stop keeping the Spalloc\
job alive).
@@ -34,8 +35,10 @@ def wait_for_exit(stop_event):
stop_event.set()
-def keep_job_alive(hostname, port, job_id, keepalive_period, timeout,
- reconnect_delay, stop_event):
+def keep_job_alive(
+ hostname: str, port: int, job_id: int, keepalive_period: float,
+ timeout: float, reconnect_delay: float,
+ stop_event: threading.Event) -> None:
""" Keeps a Spalloc job alive. Run as a separate process to the main\
Spalloc client.
@@ -78,7 +81,7 @@ def keep_job_alive(hostname, port, job_id, keepalive_period, timeout,
client.close()
-def _run(argv):
+def _run(argv: List[str]) -> None:
print("KEEPALIVE")
sys.stdout.flush()
hostname = argv[1]
diff --git a/spalloc_client/_utils.py b/spalloc_client/_utils.py
index 9f940e730..675a21506 100644
--- a/spalloc_client/_utils.py
+++ b/spalloc_client/_utils.py
@@ -14,9 +14,16 @@
from datetime import datetime
import time
+from typing import Optional
-def time_left(timestamp):
+def time_left_float(timestamp: float) -> float:
+ """ Convert a not None timestamp into how long to wait for it.
+ """
+ return max(0.0, timestamp - time.time())
+
+
+def time_left(timestamp: Optional[float]) -> Optional[float]:
""" Convert a timestamp into how long to wait for it.
"""
if timestamp is None:
@@ -24,7 +31,7 @@ def time_left(timestamp):
return max(0.0, timestamp - time.time())
-def timed_out(timestamp):
+def timed_out(timestamp: Optional[float]) -> bool:
""" Check if a timestamp has been reached.
"""
if timestamp is None:
@@ -32,7 +39,7 @@ def timed_out(timestamp):
return timestamp < time.time()
-def make_timeout(delay_seconds):
+def make_timeout(delay_seconds: Optional[float]) -> Optional[float]:
""" Convert a delay (in seconds) into a timestamp.
"""
if delay_seconds is None:
@@ -40,7 +47,7 @@ def make_timeout(delay_seconds):
return time.time() + delay_seconds
-def render_timestamp(timestamp) -> str:
+def render_timestamp(timestamp: float) -> str:
""" Convert a timestamp (Unix seconds) into a local human-readable\
timestamp string.
"""
diff --git a/spalloc_client/job.py b/spalloc_client/job.py
index 8a5b7acd7..be451dcbd 100644
--- a/spalloc_client/job.py
+++ b/spalloc_client/job.py
@@ -18,15 +18,21 @@
import logging
import subprocess
import time
+from types import TracebackType
+from typing import (cast, Dict, List, Optional, Tuple, Type, TypeVar, Union)
import sys
+from typing_extensions import Literal, Self, TypeAlias
+
+from spinn_utilities.typing.json import JsonArray
+
from spalloc_client.scripts.support import (
VERSION_RANGE_START, VERSION_RANGE_STOP)
from .protocol_client import ProtocolClient, ProtocolTimeoutError
-from .config import read_config, SEARCH_PATH
+from .spalloc_config import SpallocConfig, SEARCH_PATH
from .states import JobState
-from ._utils import time_left, timed_out, make_timeout
+from ._utils import time_left, time_left_float, timed_out, make_timeout
logger = logging.getLogger(__name__)
@@ -36,6 +42,44 @@
# https://docs.python.org/3.1/library/logging.html#configuring-logging-for-a-library
logger.addHandler(logging.StreamHandler())
+F = TypeVar('F', bound='float')
+_INT: TypeAlias = Union[int, None, Literal["USE_CONFIG"]]
+_FLOAT: TypeAlias = Union[float, None, Literal["USE_CONFIG"]]
+_LIST: TypeAlias = Union[List[str], None, Literal["USE_CONFIG"]]
+_BOOL: TypeAlias = Union[bool, None, Literal["USE_CONFIG"]]
+
+
+def pick_str(param: Optional[str], config: Optional[str]) -> Optional[str]:
+ """ Use the param unless it is the default value, otherwise use config"""
+ if param == "USE_CONFIG":
+ return config
+ return param
+
+
+def pick_list(param: _LIST,
+ config: Optional[List[str]]) -> Optional[List[str]]:
+ """ Use the param unless it is the default value, otherwise use config"""
+ if param == "USE_CONFIG":
+ return config
+ else:
+ return param
+
+
+def pick_num(param: Union[F, None, Literal["USE_CONFIG"]],
+ config: Optional[F]) -> Optional[F]:
+ """ Use the param unless it is the default value, otherwise use config"""
+ if param == "USE_CONFIG":
+ return config
+ return param
+
+
+def pick_bool(param: _BOOL, config: Optional[bool]) -> Optional[bool]:
+ """ Use the param if None or a bool, otherwise use config"""
+ if param is None or isinstance(param, bool):
+ return param
+ else:
+ return config
+
class Job(object):
""" A high-level interface for requesting and managing allocations of
@@ -121,7 +165,20 @@ class Job(object):
allocated.
"""
- def __init__(self, *args, **kwargs):
+ def __init__(self, *args: int, hostname: Optional[str] = "USE_CONFIG",
+ port: _INT = "USE_CONFIG",
+ reconnect_delay: _FLOAT = "USE_CONFIG",
+ timeout: _FLOAT = "USE_CONFIG",
+ config_filenames: _LIST = "USE_CONFIG",
+ resume_job_id: Optional[int] = None,
+ owner: Optional[str] = "USE_CONFIG",
+ keepalive: _FLOAT = "USE_CONFIG",
+ machine: Optional[str] = "USE_CONFIG",
+ tags: _LIST = "USE_CONFIG",
+ min_ratio: _FLOAT = "USE_CONFIG",
+ max_dead_boards: _INT = "USE_CONFIG",
+ max_dead_links: _INT = "USE_CONFIG",
+ require_torus: _BOOL = "USE_CONFIG"):
""" Request a SpiNNaker machine.
A :py:class:`.Job` is constructed in one of the following styles::
@@ -235,22 +292,26 @@ def __init__(self, *args, **kwargs):
specified.)
"""
# Read configuration
- config_filenames = kwargs.pop("config_filenames", SEARCH_PATH)
- config = read_config(config_filenames)
+ config_filenames = pick_list(config_filenames, SEARCH_PATH)
+ config = SpallocConfig(config_filenames)
# Get protocol client options
- hostname = kwargs.get("hostname", config["hostname"])
- owner = kwargs.get("owner", config["owner"])
- port = kwargs.get("port", config["port"])
- self._reconnect_delay = kwargs.get("reconnect_delay",
- config["reconnect_delay"])
- self._timeout = kwargs.get("timeout", config["timeout"])
+ hostname = pick_str(hostname, config.hostname)
+ owner = pick_str(owner, config.owner)
+ port = pick_num(port, config.port)
+ reconnect_delay = pick_num(reconnect_delay, config.reconnect_delay)
+ if reconnect_delay is None:
+ raise ValueError("A reconnect_delay must be specified.")
+ self._reconnect_delay = reconnect_delay
+ self._timeout = pick_num(timeout, config.timeout)
+
if hostname is None:
raise ValueError("A hostname must be specified.")
+ if port is None:
+ raise ValueError("A port must be specified.")
# Cached responses of _get_state and _get_machine_info
- self._last_state = None
- self._last_machine_info = None
+ self._last_machine_info: Optional["_JobMachineInfoTuple"] = None
# Connection to server (and associated lock)
self._client = ProtocolClient(hostname, port)
@@ -261,7 +322,6 @@ def __init__(self, *args, **kwargs):
self._assert_compatible_version()
# Resume/create the job
- resume_job_id = kwargs.get("resume_job_id", None)
if resume_job_id:
self.id = resume_job_id
@@ -285,45 +345,42 @@ def __init__(self, *args, **kwargs):
logger.info("Spalloc resumed job %d", self.id)
else:
# Get job creation arguments
- job_args = args
- job_kwargs = {
- "owner": owner,
- "keepalive": kwargs.get("keepalive", config["keepalive"]),
- "machine": kwargs.get("machine", config["machine"]),
- "tags": kwargs.get("tags", config["tags"]),
- "min_ratio": kwargs.get("min_ratio", config["min_ratio"]),
- "max_dead_boards":
- kwargs.get("max_dead_boards", config["max_dead_boards"]),
- "max_dead_links":
- kwargs.get("max_dead_links", config["max_dead_links"]),
- "require_torus":
- kwargs.get("require_torus", config["require_torus"]),
- "timeout": self._timeout,
- }
+ machine = pick_str(machine, config.machine)
+ tags = pick_list(tags, config.tags)
# Sanity check arguments
- if job_kwargs["owner"] is None:
+ if owner is None:
raise ValueError("An owner must be specified.")
- if (job_kwargs["tags"] is not None and
- job_kwargs["machine"] is not None):
+ if tags is not None and machine is not None:
raise ValueError(
"Only one of tags and machine may be specified.")
- self._keepalive = job_kwargs["keepalive"]
+ self._keepalive = pick_num(keepalive, config.keepalive)
# Create the job (failing fast if can't communicate)
- self.id = self._client.create_job(*job_args, **job_kwargs)
+ self.id = self._client.create_job(
+ self._timeout, *args, owner=owner,
+ keepalive=self._keepalive, machine=machine, tags=tags,
+ min_ratio=pick_num(min_ratio, config.min_ratio),
+ max_dead_boards=pick_num(
+ max_dead_boards, config.max_dead_boards),
+ max_dead_links=pick_num(
+ max_dead_links, config.max_dead_links),
+ require_torus=pick_bool(
+ require_torus, config.require_torus))
logger.info("Created spalloc job %d", self.id)
# Set-up and start background keepalive thread
- self._keepalive_process = subprocess.Popen(map(str, [
- sys.executable, "-m", "spalloc_client._keepalive_process",
- hostname, port, self.id, self._keepalive, self._timeout,
- self._reconnect_delay]), stdin=subprocess.PIPE,
- stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ self._keepalive_process = subprocess.Popen(
+ [sys.executable, "-m", "spalloc_client._keepalive_process",
+ str(hostname), str(port), str(self.id), str(self._keepalive),
+ str(self._timeout), str(self._reconnect_delay)],
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT)
# Wait for it to announce that it is working
stdout = self._keepalive_process.stdout
+ assert stdout is not None
while not stdout.closed:
line = stdout.readline().decode("utf-8").strip()
if line == "KEEPALIVE":
@@ -334,7 +391,7 @@ def __init__(self, *args, **kwargs):
if line:
raise ValueError(f"Keepalive process wrote odd line: {line}")
- def __enter__(self):
+ def __enter__(self) -> Self:
""" Convenience context manager for common case where a new job is to
be created and then destroyed once some code has executed.
@@ -356,11 +413,13 @@ def __enter__(self):
self.destroy()
raise
- def __exit__(self, _type, _value, _traceback):
+ def __exit__(self, exc_type: Optional[Type],
+ exc_value: Optional[BaseException],
+ exc_tb: Optional[TracebackType]) -> Literal[False]:
self.destroy()
return False
- def _assert_compatible_version(self):
+ def _assert_compatible_version(self) -> None:
""" Assert that the server version is compatible.
"""
v = self._client.version(timeout=self._timeout)
@@ -371,7 +430,7 @@ def _assert_compatible_version(self):
raise ValueError(
f"Server version {v} is not compatible with this client.")
- def _reconnect(self):
+ def _reconnect(self) -> None:
""" Reconnect to the server and check version.
If reconnection fails, the error is reported as a warning but no
@@ -388,7 +447,7 @@ def _reconnect(self):
"Spalloc server is unreachable (%s), will keep trying...", e)
self._client.close()
- def destroy(self, reason=None):
+ def destroy(self, reason: Optional[str] = None) -> None:
""" Destroy the job and disconnect from the server.
Parameters
@@ -407,7 +466,7 @@ def destroy(self, reason=None):
self.close()
- def close(self):
+ def close(self) -> None:
""" Disconnect from the server and stop keeping the job alive.
.. warning::
@@ -425,7 +484,7 @@ def close(self):
# Disconnect
self._client.close()
- def _get_state(self):
+ def _get_state(self) -> "_JobStateTuple":
""" Get the state of the job.
Returns
@@ -434,12 +493,12 @@ def _get_state(self):
"""
state = self._client.get_job_state(self.id, timeout=self._timeout)
return _JobStateTuple(
- state=JobState(state["state"]),
+ state=JobState(cast(int, state["state"])),
power=state["power"],
keepalive=state["keepalive"],
reason=state["reason"])
- def set_power(self, power):
+ def set_power(self, power: bool) -> None:
""" Turn the boards allocated to the job on or off.
Does nothing if the job has not yet been allocated any boards.
@@ -458,7 +517,7 @@ def set_power(self, power):
else:
self._client.power_off_job_boards(self.id, timeout=self._timeout)
- def reset(self):
+ def reset(self) -> None:
""" Reset (power-cycle) the boards allocated to the job.
Does nothing if the job has not been allocated.
@@ -468,7 +527,7 @@ def reset(self):
"""
self.set_power(True)
- def _get_machine_info(self):
+ def _get_machine_info(self) -> "_JobMachineInfoTuple":
""" Get information about the boards allocated to the job, e.g. the IPs
and system dimensions.
@@ -479,39 +538,37 @@ def _get_machine_info(self):
info = self._client.get_job_machine_info(
self.id, timeout=self._timeout)
+ info_connections = cast(list, info["connections"])
return _JobMachineInfoTuple(
width=info["width"],
height=info["height"],
connections=({(x, y): hostname
- for (x, y), hostname in info["connections"]}
- if info["connections"] is not None
+ for (x, y), hostname in info_connections}
+ if info_connections is not None
else None),
machine_name=info["machine_name"],
boards=info["boards"])
@property
- def state(self):
+ def state(self) -> JobState:
""" The current state of the job.
"""
- self._last_state = self._get_state()
- return self._last_state.state
+ return self._get_state().state
@property
- def power(self):
+ def power(self) -> bool:
""" Are the boards powered/powering on or off?
"""
- self._last_state = self._get_state()
- return self._last_state.power
+ return self._get_state().power
@property
- def reason(self):
+ def reason(self) -> str:
""" For what reason was the job destroyed (if any and if destroyed).
"""
- self._last_state = self._get_state()
- return self._last_state.reason
+ return self._get_state().reason
@property
- def connections(self):
+ def connections(self) -> Dict[Tuple[int, int], str]:
""" The list of Ethernet connected chips and their IPs.
Returns
@@ -527,13 +584,13 @@ def connections(self):
return self._last_machine_info.connections
@property
- def hostname(self):
+ def hostname(self) -> Optional[str]:
""" The hostname of chip 0, 0 (or None if not allocated yet).
"""
return self.connections[(0, 0)]
@property
- def width(self):
+ def width(self) -> int:
""" The width of the allocated machine in chips (or None).
"""
# Note that the dimensions of a job will never change once defined so
@@ -545,7 +602,7 @@ def width(self):
return self._last_machine_info.width
@property
- def height(self):
+ def height(self) -> int:
""" The height of the allocated machine in chips (or None).
"""
# Note that the dimensions of a job will never change once defined so
@@ -557,7 +614,7 @@ def height(self):
return self._last_machine_info.height
@property
- def machine_name(self):
+ def machine_name(self) -> str:
""" The name of the machine the job is allocated on (or None).
"""
# Note that the machine will never change once defined so only need to
@@ -569,7 +626,7 @@ def machine_name(self):
return self._last_machine_info.machine_name
@property
- def boards(self):
+ def boards(self) -> Optional[JsonArray]:
""" The coordinates of the boards allocated for the job (or None).
"""
# Note that the machine will never change once defined so only need to
@@ -580,7 +637,8 @@ def boards(self):
return self._last_machine_info.boards
- def wait_for_state_change(self, old_state, timeout=None):
+ def wait_for_state_change(self, old_state: JobState,
+ timeout: Optional[float] = None) -> JobState:
""" Block until the job's state changes from the supplied state.
Parameters
@@ -626,7 +684,7 @@ def wait_for_state_change(self, old_state, timeout=None):
# return the old state
return old_state
- def _do_wait_for_a_change(self, finish_time):
+ def _do_wait_for_a_change(self, finish_time: Optional[float]) -> bool:
""" Wait for a state change and keep the job alive.
"""
# Since we're about to block holding the client lock, we must be
@@ -640,12 +698,12 @@ def _do_wait_for_a_change(self, finish_time):
# user-specified timeout or half the keepalive interval.
if finish_time is not None and self._keepalive is not None:
wait_timeout = min(self._keepalive / 2.0,
- time_left(finish_time))
+ time_left_float(finish_time))
elif finish_time is None:
wait_timeout = None if self._keepalive is None \
else self._keepalive / 2.0
else:
- wait_timeout = time_left(finish_time)
+ wait_timeout = time_left_float(finish_time)
if wait_timeout is None or wait_timeout >= 0.0:
self._client.wait_for_notification(wait_timeout)
return True
@@ -656,19 +714,19 @@ def _do_wait_for_a_change(self, finish_time):
# The user's timeout expired while waiting for a state change
return False
- def _do_reconnect(self, finish_time):
+ def _do_reconnect(self, finish_time: Optional[float]) -> None:
""" Reconnect after the reconnection delay (or timeout, whichever
came first).
"""
self._client.close()
if finish_time is not None:
- delay = min(time_left(finish_time), self._reconnect_delay)
+ delay = min(time_left_float(finish_time), self._reconnect_delay)
else:
delay = self._reconnect_delay
time.sleep(max(0.0, delay))
self._reconnect()
- def wait_until_ready(self, timeout=None):
+ def wait_until_ready(self, timeout: Optional[float] = None) -> None:
""" Block until the job is allocated and ready.
Parameters
@@ -715,7 +773,8 @@ def wait_until_ready(self, timeout=None):
# Timed out!
raise StateChangeTimeoutError()
- def where_is_machine(self, chip_x, chip_y):
+ def where_is_machine(
+ self, chip_x: int, chip_y: int) -> Tuple[int, int, int]:
""" Locates and returns cabinet, frame, board for a given chip in a\
machine allocated to this job.
@@ -727,7 +786,8 @@ def where_is_machine(self, chip_x, chip_y):
job_id=self.id, chip_x=chip_x, chip_y=chip_y)
if result is None:
raise ValueError("received None instead of machine location")
- return result['physical']
+ [cabinet, frame, board] = cast(list, result['physical'])
+ return (cast(int, cabinet), cast(int, frame), cast(int, board))
class StateChangeTimeoutError(Exception):
diff --git a/spalloc_client/protocol_client.py b/spalloc_client/protocol_client.py
index 65bfc5ebd..473359c9c 100644
--- a/spalloc_client/protocol_client.py
+++ b/spalloc_client/protocol_client.py
@@ -18,10 +18,14 @@
import errno
import json
import socket
-from typing import Dict, List, Optional
-from threading import current_thread, RLock, local
+from types import TracebackType
+from typing import Any, cast, Dict, List, Literal, Optional, Type, Union
+from threading import current_thread, RLock, local, Thread
-from spinn_utilities.typing.json import JsonObject, JsonObjectArray
+from typing_extensions import Self
+
+from spinn_utilities.typing.json import (
+ JsonObject, JsonObjectArray, JsonValue)
from spalloc_client._utils import time_left, timed_out, make_timeout
@@ -47,10 +51,10 @@ class _ProtocolThreadLocal(local):
of our state in each thread.
"""
# See https://github.com/SpiNNakerManchester/spalloc/issues/12
- def __init__(self):
+ def __init__(self) -> None:
super().__init__()
self.buffer = b""
- self.sock = None
+ self.sock: Optional[socket.socket] = None
class ProtocolClient(object):
@@ -80,7 +84,8 @@ class ProtocolClient(object):
# Done!
"""
- def __init__(self, hostname, port=22244, timeout=None):
+ def __init__(self, hostname: str, port: int = 22244,
+ timeout: Optional[float] = None):
""" Define a new connection.
.. note::
@@ -91,32 +96,34 @@ def __init__(self, hostname, port=22244, timeout=None):
----------
hostname : str
The hostname of the server.
- port : str or int
+ port : int
The port to use (default: 22244).
"""
self._hostname = hostname
self._port = port
# Mapping from threads to sockets. Kept because we need to have way to
# shut down all sockets at once.
- self._socks = dict()
+ self._socks: Dict[Thread, socket.socket] = dict()
# Thread local variables
self._local = _ProtocolThreadLocal()
# A queue of unprocessed notifications
- self._notifications = deque()
+ self._notifications: deque = deque()
self._dead = True
self._socks_lock = RLock()
self._notifications_lock = RLock()
self._default_timeout = timeout
- def __enter__(self):
+ def __enter__(self) -> Self:
self.connect(self._default_timeout)
return self
- def __exit__(self, exc_type, exc_val, exc_tb):
+ def __exit__(self, exc_type: Optional[Type],
+ exc_value: Optional[BaseException],
+ exc_tb: Optional[TracebackType]) -> Literal[False]:
self.close()
return False
- def _get_connection(self, timeout: Optional[int]) -> socket.socket:
+ def _get_connection(self, timeout: Optional[float]) -> socket.socket:
if self._dead:
raise OSError(errno.ENOTCONN, "not connected")
connect_needed = False
@@ -141,7 +148,7 @@ def _get_connection(self, timeout: Optional[int]) -> socket.socket:
sock.settimeout(timeout)
return sock
- def _do_connect(self, sock: socket.socket):
+ def _do_connect(self, sock: socket.socket) -> bool:
success = False
try:
sock.connect((self._hostname, self._port))
@@ -154,7 +161,7 @@ def _do_connect(self, sock: socket.socket):
def _has_open_socket(self) -> bool:
return self._local.sock is not None
- def connect(self, timeout: Optional[int] = None):
+ def connect(self, timeout: Optional[float] = None) -> None:
"""(Re)connect to the server.
Raises
@@ -168,7 +175,7 @@ def connect(self, timeout: Optional[int] = None):
self._dead = False
self._connect(timeout)
- def _connect(self, timeout: Optional[int]) -> socket.socket:
+ def _connect(self, timeout: Optional[float]) -> socket.socket:
""" Try to (re)connect to the server.
"""
try:
@@ -179,7 +186,7 @@ def _connect(self, timeout: Optional[int]) -> socket.socket:
# Pass on the exception
raise
- def _close(self, key=None):
+ def _close(self, key: Optional[Thread] = None) -> None:
if key is None:
key = current_thread()
with self._socks_lock:
@@ -192,7 +199,7 @@ def _close(self, key=None):
self._local.buffer = b""
sock.close()
- def close(self):
+ def close(self) -> None:
""" Disconnect from the server.
"""
self._dead = True
@@ -202,7 +209,7 @@ def close(self):
self._close(key)
self._local = _ProtocolThreadLocal()
- def _recv_json(self, timeout=None) -> JsonObject:
+ def _recv_json(self, timeout: Optional[float] = None) -> JsonObject:
""" Receive a line of JSON from the server.
Parameters
@@ -242,7 +249,8 @@ def _recv_json(self, timeout=None) -> JsonObject:
line, _, self._local.buffer = self._local.buffer.partition(b"\n")
return json.loads(line.decode("utf-8"))
- def _send_json(self, obj, timeout=None):
+ def _send_json(
+ self, obj: JsonObject, timeout: Optional[float] = None) -> None:
""" Attempt to send a line of JSON to the server.
Parameters
@@ -271,7 +279,9 @@ def _send_json(self, obj, timeout=None):
except socket.timeout as e:
raise ProtocolTimeoutError("send timed out.") from e
- def call(self, name, *args, **kwargs):
+ def call(self, name: str, timeout: Optional[float],
+ *args: Union[int, str, None],
+ **kwargs: Any) -> JsonValue:
""" Send a command to the server and return the reply.
Parameters
@@ -295,11 +305,13 @@ def call(self, name, *args, **kwargs):
If the connection is unavailable or is closed.
"""
try:
- timeout = kwargs.pop("timeout", None)
finish_time = make_timeout(timeout)
# Construct the command message
- command = {"command": name, "args": args, "kwargs": kwargs}
+ command: JsonObject = {}
+ command["command"] = name
+ command["args"] = list(args)
+ command["kwargs"] = kwargs
self._send_json(command, timeout=timeout)
# Command sent! Attempt to receive the response...
@@ -313,10 +325,13 @@ def call(self, name, *args, **kwargs):
# Got a notification, keep trying...
with self._notifications_lock:
self._notifications.append(obj)
+
+ raise ProtocolTimeoutError(f"{timeout=} passed!")
except (IOError, OSError) as e:
raise ProtocolError(str(e)) from e
- def wait_for_notification(self, timeout=None):
+ def wait_for_notification(
+ self, timeout: Optional[float] = None) -> Optional[JsonObject]:
""" Return the next notification to arrive.
Parameters
@@ -361,98 +376,117 @@ def wait_for_notification(self, timeout=None):
# The bindings of the Spalloc protocol methods themselves; simplifies use
# from IDEs.
- def version(self, timeout: Optional[int] = None) -> str:
+ def version(self, timeout: Optional[float] = None) -> str:
""" Ask what version of spalloc is running. """
- return self.call("version", timeout=timeout)
-
- def create_job(self, *args: List[object],
- **kwargs: Dict[str, object]) -> JsonObject:
+ return cast(str, self.call("version", timeout))
+
+ def create_job(self, timeout: Optional[float], *args: int,
+ owner: Optional[str] = None,
+ keepalive: Optional[float] = None,
+ machine: Optional[str] = None,
+ tags: Optional[List[str]] = None,
+ min_ratio: Optional[float] = None,
+ max_dead_boards: Optional[int] = None,
+ max_dead_links: Optional[int] = None,
+ require_torus: Optional[bool] = None) -> int:
"""
Start a new job
"""
# If no owner, don't bother with the call
- if "owner" not in kwargs:
+ if owner is None:
raise SpallocServerException(
"owner must be specified for all jobs.")
- return self.call("create_job", *args, **kwargs)
+ if tags is not None and machine is not None:
+ raise SpallocServerException(
+ f"Unexpected {tags=} and {machine=} are both not None")
+ return cast(int, self.call(
+ "create_job", timeout, *args, owner=owner,
+ keepalive=keepalive, machine=machine, tags=tags,
+ min_ratio=min_ratio, max_dead_boards=max_dead_boards,
+ max_dead_links=max_dead_links, require_torus=require_torus))
def job_keepalive(self, job_id: int,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = 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)
+ return cast(dict, self.call("job_keepalive", timeout, job_id))
def get_job_state(self, job_id: int,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
"""Get the state for this job """
- return self.call("get_job_state", job_id, timeout=timeout)
+ return cast(dict, self.call("get_job_state", timeout, job_id))
def get_job_machine_info(self, job_id: int,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
""" Get info for this job. """
- return self.call("get_job_machine_info", job_id, timeout=timeout)
+ return cast(dict, self.call("get_job_machine_info", timeout, job_id))
def power_on_job_boards(self, job_id: int,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
""" Turn on the power on the jobs boards. """
- return self.call("power_on_job_boards", job_id, timeout=timeout)
+ return cast(dict, self.call("power_on_job_boards", timeout, job_id))
def power_off_job_boards(self, job_id: int,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
""" Turn off the power on the jobs boards. """
- return self.call("power_off_job_boards", job_id, timeout=timeout)
+ return cast(dict, self.call("power_off_job_boards", timeout, job_id))
def destroy_job(self, job_id: int, reason: Optional[str] = None,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
""" Destroy the job """
- return self.call("destroy_job", job_id, reason, timeout=timeout)
+ return cast(dict, self.call("destroy_job", timeout,
+ job_id, reason=reason))
def notify_job(self, job_id: Optional[int] = None,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
""" Turn on notification of job status changes. """
- return self.call("notify_job", job_id, timeout=timeout)
+ return cast(dict, self.call("notify_job", timeout, job_id))
def no_notify_job(self, job_id: Optional[int] = None,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
""" Turn off notification of job status changes. """
- return self.call("no_notify_job", job_id, timeout=timeout)
+ return cast(dict, self.call("no_notify_job", timeout,
+ job_id))
def notify_machine(self, machine_name: Optional[str] = None,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
""" Turn on notification of machine status changes. """
- return self.call("notify_machine", machine_name, timeout=timeout)
+ return cast(dict, self.call("notify_machine", timeout,
+ machine_name))
def no_notify_machine(self, machine_name: Optional[str] = None,
- timeout: Optional[int] = None) -> JsonObject:
+ timeout: Optional[float] = None) -> JsonObject:
""" Turn off notification of machine status changes. """
- return self.call("no_notify_machine", machine_name, timeout=timeout)
+ return cast(dict, self.call("no_notify_machine", timeout,
+ machine_name))
- def list_jobs(self, timeout: Optional[int] = None) -> JsonObjectArray:
+ def list_jobs(self, timeout: Optional[float] = None) -> JsonObjectArray:
""" Obtains a list of jobs currently running. """
- return self.call("list_jobs", timeout=timeout)
+ return cast(list, self.call("list_jobs", timeout))
def list_machines(self,
timeout: Optional[float] = None) -> JsonObjectArray:
""" Obtains a list of currently supported machines. """
- return self.call("list_machines", timeout=timeout)
+ return cast(list, self.call("list_machines", timeout))
- def get_board_position(self, machine_name: str, x: int, y: int, z: int,
- timeout: Optional[int] = None): # pragma: no cover
+ def get_board_position(
+ self, machine_name: str, x: int, y: int, z: int,
+ timeout: Optional[float] = None) -> JsonObject: # 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)
+ return cast(dict, self.call("get_board_position", timeout,
+ machine_name, x, y, z))
def get_board_at_position(self, machine_name: str, x: int, y: int, z: int,
- timeout: Optional[int] = None
+ timeout: Optional[float] = 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)
+ return cast(dict, self.call("get_board_at_position", timeout,
+ machine_name, x, y, z))
_acceptable_kwargs_for_where_is = frozenset([
frozenset("machine x y z".split()),
@@ -460,12 +494,9 @@ def get_board_at_position(self, machine_name: str, x: int, y: int, z: int,
frozenset("machine chip_x chip_y".split()),
frozenset("job_id chip_x chip_y".split())])
- def where_is(self, timeout: Optional[int] = None, **kwargs) -> JsonObject:
+ def where_is(self, job_id: int, chip_x: int, chip_y: int,
+ timeout: Optional[float] = None) -> 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:
- raise SpallocServerException(
- f"Invalid arguments: {', '.join(keywords)}")
- kwargs["timeout"] = timeout
- return self.call("where_is", **kwargs)
+ return cast(dict, self.call("where_is", timeout, job_id=job_id,
+ chip_x=chip_x, chip_y=chip_y))
diff --git a/spalloc_client/scripts/alloc.py b/spalloc_client/scripts/alloc.py
index ea8ae3434..f1228476e 100644
--- a/spalloc_client/scripts/alloc.py
+++ b/spalloc_client/scripts/alloc.py
@@ -113,11 +113,12 @@
import subprocess
import sys
import tempfile
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple, Union
from shlex import quote
from spalloc_client import (
- config, Job, JobState, __version__, ProtocolError, ProtocolTimeoutError,
+ Job, JobState, __version__, ProtocolError, ProtocolTimeoutError,
SpallocServerException)
+from spalloc_client.spalloc_config import SpallocConfig
from spalloc_client.term import Terminal, render_definitions
# pylint: disable=invalid-name
@@ -127,7 +128,7 @@
def write_ips_to_csv(connections: Dict[Tuple[int, int], str],
- ip_file_filename: str):
+ ip_file_filename: str) -> None:
""" Write the supplied IP addresses to a CSV file.
The produced CSV has three columns: x, y and hostname where x and y give
@@ -147,7 +148,7 @@ def write_ips_to_csv(connections: Dict[Tuple[int, int], str],
def print_info(machine_name: str, connections: Dict[Tuple[int, int], str],
- width: int, height: int, ip_file_filename: str):
+ width: int, height: int, ip_file_filename: str) -> None:
""" Print the current machine info in a human-readable form and wait for
the user to press enter.
@@ -186,7 +187,7 @@ def print_info(machine_name: str, connections: Dict[Tuple[int, int], str],
def run_command(
command: List[str], job_id: int, machine_name: str,
connections: Dict[Tuple[int, int], str], width: int, height: int,
- ip_file_filename: str):
+ ip_file_filename: str) -> int:
""" Run a user-specified command, substituting arguments for values taken
from the allocated board.
@@ -254,7 +255,7 @@ def run_command(
p.terminate()
-def info(msg: str):
+def info(msg: str) -> None:
"""
Writes a message to the terminal
"""
@@ -264,15 +265,15 @@ def info(msg: str):
t.stream.write(f"{msg}\n")
-def update(msg: str, colour: functools.partial, *args: List[object]):
+def update(msg: str, colour: functools.partial) -> None:
"""
Writes a message to the terminal in the schoosen colour.
"""
assert t is not None
- info(t.update(colour(msg.format(*args))))
+ info(t.update(colour(msg)))
-def wait_for_job_ready(job: Job):
+def wait_for_job_ready(job: Job) -> Tuple[int, Optional[str]]:
"""
Wait for it to become ready, keeping the user informed along the way
"""
@@ -284,11 +285,9 @@ def wait_for_job_ready(job: Job):
# Show debug info on state-change
if old_state != cur_state:
if cur_state == JobState.queued:
- update("Job {}: Waiting in queue...", t.yellow,
- job.id)
+ update(f"Job {job.id}: Waiting in queue...", t.yellow)
elif cur_state == JobState.power:
- update("Job {}: Waiting for power on...", t.yellow,
- job.id)
+ update(f"Job {job.id}: Waiting for power on...", t.yellow)
elif cur_state == JobState.ready:
# Here we go!
return 0, None
@@ -301,26 +300,24 @@ def wait_for_job_ready(job: Job):
pass
if reason is not None:
- update("Job {}: Destroyed: {}", t.red,
- job.id, reason)
+ update(f"Job {job.id}: Destroyed: {reason}", t.red)
else:
- update("Job {}: Destroyed.", t.red,
- job.id)
+ update(f"Job {job.id}: Destroyed.", t.red)
return 1, reason
elif cur_state == JobState.unknown:
- update("Job {}: Job not recognised by server.", t.red,
- job.id)
+ update(f"Job {job.id}: Job not recognised by server.",
+ t.red)
return 2, None
else:
- update("Job {}: Entered an unrecognised state {}.",
- t.red, job.id, cur_state)
+ update(f"Job {job.id}: Entered an unrecognised state "
+ f"{cur_state}.", t.red)
return 3, None
old_state, cur_state = \
cur_state, job.wait_for_state_change(cur_state)
except KeyboardInterrupt:
# Gracefully terminate from keyboard interrupt
- update("Job {}: Keyboard interrupt.", t.red, job.id)
+ update(f"Job {job.id}: Keyboard interrupt.", t.red)
return 4, "Keyboard interrupt."
@@ -329,7 +326,7 @@ def parse_argv(argv: Optional[List[str]]) -> Tuple[
"""
Parse the arguments.
"""
- cfg = config.read_config()
+ cfg = SpallocConfig()
parser = argparse.ArgumentParser(
description="Request (and allocate) a SpiNNaker machine.")
@@ -358,38 +355,38 @@ def parse_argv(argv: Optional[List[str]]) -> Tuple[
help="if given, resume keeping the specified job alive rather than "
"creating a new job (all allocation requirements will be ignored)")
allocation_args.add_argument(
- "--machine", "-m", nargs="?", default=cfg["machine"],
+ "--machine", "-m", nargs="?", default=cfg.machine,
help="only allocate boards which are part of a specific machine, or "
"any machine if no machine is given (default: %(default)s)")
allocation_args.add_argument(
"--tags", "-t", nargs="*", metavar="TAG",
- default=cfg["tags"] or ["default"],
+ default=cfg.tags or ["default"],
help="only allocate boards which have (at least) the specified flags "
- f"(default: {' '.join(cfg['tags'] or [])})")
+ f"(default: {' '.join(cfg.tags or [])})")
allocation_args.add_argument(
- "--min-ratio", type=float, metavar="RATIO", default=cfg["min_ratio"],
+ "--min-ratio", type=float, metavar="RATIO", default=cfg.min_ratio,
help="when allocating by number of boards, require that the "
"allocation be at least as square as this ratio (default: "
"%(default)s)")
allocation_args.add_argument(
"--max-dead-boards", type=int, metavar="NUM", default=(
- -1 if cfg["max_dead_boards"] is None else cfg["max_dead_boards"]),
+ -1 if cfg.max_dead_boards is None else cfg.max_dead_boards),
help="boards allowed to be dead in the allocation, or -1 to allow "
"any number of dead boards (default: %(default)s)")
allocation_args.add_argument(
"--max-dead-links", type=int, metavar="NUM", default=(
- -1 if cfg["max_dead_links"] is None else cfg["max_dead_links"]),
+ -1 if cfg.max_dead_links is None else cfg.max_dead_links),
help="inter-board links allowed to be dead in the allocation, or -1 "
"to allow any number of dead links (default: %(default)s)")
allocation_args.add_argument(
"--require-torus", "-w", action="store_true",
- default=cfg["require_torus"],
+ default=cfg.require_torus,
help="require that the allocation contain torus (a.k.a. wrap-around) "
- f"links {'(default)' if cfg['require_torus'] else ''}")
+ f"links {'(default)' if cfg.require_torus else ''}")
allocation_args.add_argument(
"--no-require-torus", "-W", action="store_false", dest="require_torus",
help="do not require that the allocation contain torus (a.k.a. "
- f"wrap-around) links {'' if cfg['require_torus'] else '(default)'}")
+ f"wrap-around) links {'' if cfg.require_torus else '(default)'}")
command_args = parser.add_argument_group("command wrapping arguments")
command_args.add_argument(
@@ -404,35 +401,36 @@ def parse_argv(argv: Optional[List[str]]) -> Tuple[
server_args = parser.add_argument_group("spalloc server arguments")
server_args.add_argument(
- "--owner", default=cfg["owner"],
+ "--owner", default=cfg.owner,
help="by convention, the email address of the owner of the job "
"(default: %(default)s)")
server_args.add_argument(
- "--hostname", "-H", default=cfg["hostname"],
+ "--hostname", "-H", default=cfg.hostname,
help="hostname or IP of the spalloc server (default: %(default)s)")
server_args.add_argument(
- "--port", "-P", default=cfg["port"], type=int,
+ "--port", "-P", default=cfg.port, type=int,
help="port number of the spalloc server (default: %(default)s)")
server_args.add_argument(
"--keepalive", type=int, metavar="SECONDS",
- default=(-1 if cfg["keepalive"] is None else cfg["keepalive"]),
+ default=(-1 if cfg.keepalive is None else cfg.keepalive),
help="the interval at which to require keepalive messages to be "
"sent to prevent the server cancelling the job, or -1 to not "
"require keepalive messages (default: %(default)s)")
server_args.add_argument(
- "--reconnect-delay", default=cfg["reconnect_delay"], type=float,
+ "--reconnect-delay", default=cfg.reconnect_delay, type=float,
metavar="SECONDS",
help="seconds to wait before reconnecting to the server if the "
"connection is lost (default: %(default)s)")
server_args.add_argument(
- "--timeout", default=cfg["timeout"], type=float, metavar="SECONDS",
+ "--timeout", default=cfg.timeout, type=float, metavar="SECONDS",
help="seconds to wait for a response from the server "
"(default: %(default)s)")
return parser, parser.parse_args(argv)
-def run_job(job_args: List[str], job_kwargs: Dict[str, str],
- ip_file_filename: str):
+def run_job(ip_file_filename: str,
+ job_args: List[int],
+ job_kwargs: Dict[str, Union[float, str, None]]) -> int:
"""
Run a job
"""
@@ -443,7 +441,7 @@ def run_job(job_args: List[str], job_kwargs: Dict[str, str],
# Create the job
try:
- job = Job(*job_args, **job_kwargs)
+ job = Job(*job_args, **job_kwargs) # type: ignore[arg-type]
except (OSError, IOError, ProtocolError, ProtocolTimeoutError) as e:
info(t.red(f"Could not connect to server: {e}"))
return 6
@@ -458,7 +456,7 @@ def run_job(job_args: List[str], job_kwargs: Dict[str, str],
# Machine is now ready
write_ips_to_csv(job.connections, ip_file_filename)
- update("Job {}: Ready!", t.green, job.id)
+ update(f"Job {job.id}: Ready!", t.green)
# Either run the user's application or just print the details.
if not arguments.command:
@@ -483,7 +481,7 @@ def _minzero(value: float) -> Optional[float]:
return value if value >= 0.0 else None
-def main(argv: Optional[List[str]] = None):
+def main(argv: Optional[List[str]] = None) -> int:
"""
The main method run
"""
@@ -501,7 +499,7 @@ def main(argv: Optional[List[str]] = None):
parser.error("--hostname of spalloc server must be specified")
# Set universal job arguments
- job_kwargs = {
+ job_kwargs: Dict[str, Union[float, str, None]] = {
"hostname": arguments.hostname,
"port": arguments.port,
"reconnect_delay": _minzero(arguments.reconnect_delay),
@@ -541,7 +539,7 @@ def main(argv: Optional[List[str]] = None):
_, ip_file_filename = tempfile.mkstemp(".csv", "spinnaker_ips_")
try:
- return run_job(job_args, job_kwargs, ip_file_filename)
+ return run_job(ip_file_filename, job_args, job_kwargs)
except SpallocServerException as e: # pragma: no cover
info(t.red(f"Error from server: {e}"))
return 6
diff --git a/spalloc_client/scripts/job.py b/spalloc_client/scripts/job.py
index d44dcde83..cdb8edfe6 100644
--- a/spalloc_client/scripts/job.py
+++ b/spalloc_client/scripts/job.py
@@ -76,23 +76,27 @@
"""
import argparse
import sys
-from typing import Any, Dict
+from typing import Any, cast, Dict, Optional
from spinn_utilities.overrides import overrides
+from spinn_utilities.typing.json import JsonObject
from spalloc_client import __version__, JobState
from spalloc_client.term import (
Terminal, render_definitions, render_boards, DEFAULT_BOARD_EDGES)
from spalloc_client import ProtocolClient
from spalloc_client._utils import render_timestamp
+from spalloc_client.spalloc_config import SpallocConfig
from spalloc_client.scripts.support import Terminate, Script
-def _state_name(mapping):
- return JobState(mapping["state"]).name # pylint: disable=no-member
+def _state_name(mapping: JsonObject) -> str:
+ state = JobState(cast(int, mapping["state"]))
+ return state.name # pylint: disable=no-member
-def show_job_info(t, client, timeout, job_id):
+def show_job_info(t: Terminal, client: ProtocolClient,
+ timeout: Optional[float], job_id: int) -> None:
""" Print a human-readable overview of a Job's attributes.
Parameters
@@ -106,28 +110,27 @@ def show_job_info(t, client, timeout, job_id):
job_id : int
The job ID of interest.
- Returns
+ Returnsisin
-------
int
An error code, 0 for success.
"""
# Get the complete job information (if the job is alive)
job_list = client.list_jobs(timeout=timeout)
- job = [job for job in job_list if job["job_id"] == job_id]
- info = dict()
+ jobs = [job for job in job_list if job["job_id"] == job_id]
+ info: Dict[str, Any] = dict()
info["Job ID"] = job_id
- if not job:
+ if not jobs:
# Job no longer exists, just print basic info
- job = client.get_job_state(job_id, timeout=timeout)
-
+ job = cast(dict, client.get_job_state(job_id, timeout=timeout))
info["State"] = _state_name(job)
if job["reason"] is not None:
info["Reason"] = job["reason"]
else:
# Job is enqueued, show all info
machine_info = client.get_job_machine_info(job_id, timeout=timeout)
- job = job[0]
+ job = cast(dict, jobs[0])
info["Owner"] = job["owner"]
info["State"] = _state_name(job)
@@ -137,8 +140,8 @@ def show_job_info(t, client, timeout, job_id):
if "keepalivehost" in job and job["keepalivehost"] is not None:
info["Owner host"] = job["keepalivehost"]
- args = job["args"]
- kwargs = job["kwargs"]
+ args = cast(list, job["args"])
+ kwargs = cast(dict, job["kwargs"])
info["Request"] = "Job({}{}{})".format(
", ".join(map(str, args)),
",\n " if args and kwargs else "",
@@ -151,10 +154,11 @@ def show_job_info(t, client, timeout, job_id):
t.dim(" . "),
tuple(map(t.dim, DEFAULT_BOARD_EDGES)),
tuple(map(t.bright, DEFAULT_BOARD_EDGES)),
- )])
+ )], [])
if machine_info["connections"] is not None:
- connections = sorted(machine_info["connections"])
+ connections = cast(list, machine_info["connections"])
+ connections.sort()
info["Hostname"] = connections[0][1]
if machine_info["width"] is not None:
info["Width"] = machine_info["width"]
@@ -170,7 +174,8 @@ def show_job_info(t, client, timeout, job_id):
print(render_definitions(info))
-def watch_job(t, client, timeout, job_id):
+def watch_job(t: Terminal, client: ProtocolClient, timeout: Optional[float],
+ job_id: int) -> int:
""" Re-print a job's information whenever the job changes.
Parameters
@@ -179,7 +184,7 @@ def watch_job(t, client, timeout, job_id):
An output styling object for stdout.
client : :py:class:`.ProtocolClient`
A connection to the server.
- timeout : float or None
+ timeout : int or None
The timeout for server responses.
job_id : int
The job ID of interest.
@@ -203,14 +208,15 @@ def watch_job(t, client, timeout, job_id):
print("")
-def power_job(client, timeout, job_id, power):
+def power_job(client: ProtocolClient, timeout: Optional[float],
+ job_id: int, power: bool) -> None:
""" Power a job's boards on/off and wait for the action to complete.
Parameters
----------
client : :py:class:`.ProtocolClient`
A connection to the server.
- timeout : float or None
+ timeout : int or None
The timeout for server responses.
job_id : int
The job ID of interest.
@@ -251,14 +257,15 @@ def power_job(client, timeout, job_id, power):
f"job {job_id} in state {_state_name(state)}"))
-def list_ips(client, timeout, job_id):
+def list_ips(client: ProtocolClient, timeout: Optional[float],
+ job_id: int) -> None:
""" Print a CSV of board hostnames for all boards allocated to a job.
Parameters
----------
client : :py:class:`.ProtocolClient`
A connection to the server.
- timeout : float or None
+ timeout : int or None
The timeout for server responses.
job_id : int
The job ID of interest.
@@ -270,22 +277,28 @@ def list_ips(client, timeout, job_id):
boards.
"""
info = client.get_job_machine_info(job_id, timeout=timeout)
- connections = info["connections"]
+ connections = cast(list, info["connections"])
if connections is None:
raise Terminate(9, f"Job {job_id} is queued or does not exist")
print("x,y,hostname")
- for ((x, y), hostname) in sorted(connections):
+ connections.sort()
+ for connection in connections:
+ assert isinstance(connection, list)
+ (xy, hostname) = connection
+ assert isinstance(xy, list)
+ (x, y) = xy
print(f"{x},{y},{hostname}")
-def destroy_job(client, timeout, job_id, reason=None):
+def destroy_job(client: ProtocolClient, timeout: Optional[float],
+ job_id: int, reason: Optional[str] = None) -> None:
""" Destroy a running job.
Parameters
----------
client : :py:class:`.ProtocolClient`
A connection to the server.
- timeout : float or None
+ timeout : int or None
The timeout for server responses.
job_id : int
The job ID of interest.
@@ -305,11 +318,12 @@ class ManageJobScript(Script):
A tool for running Job scripts.
"""
- def __init__(self):
+ def __init__(self) -> None:
super().__init__()
- self.parser = None
+ self.parser: Optional[argparse.ArgumentParser] = None
- def get_job_id(self, client: ProtocolClient, args: argparse.Namespace):
+ def get_job_id(self, client: ProtocolClient,
+ args: argparse.Namespace) -> int:
"""
get a job for the owner named in the args
"""
@@ -324,10 +338,10 @@ def get_job_id(self, client: ProtocolClient, args: argparse.Namespace):
msg = (f"Ambiguous: {args.owner} has {len(job_ids)} live jobs: "
f"{', '.join(map(str, job_ids))}")
raise Terminate(3, msg)
- return job_ids[0]
+ return cast(int, job_ids[0])
@overrides(Script.get_parser)
- def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
+ def get_parser(self, cfg: SpallocConfig) -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Manage running jobs.")
parser.add_argument(
@@ -337,7 +351,7 @@ def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
help="the job ID of interest, optional if the current owner only "
"has one job")
parser.add_argument(
- "--owner", "-o", default=cfg["owner"],
+ "--owner", "-o", default=cfg.owner,
help="if no job ID is provided and this owner has only one job, "
"this job is assumed (default: %(default)s)")
control_args = parser.add_mutually_exclusive_group()
@@ -363,12 +377,13 @@ def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
return parser
@overrides(Script.verify_arguments)
- def verify_arguments(self, args: argparse.Namespace):
+ def verify_arguments(self, args: argparse.Namespace) -> None:
if args.job_id is None and args.owner is None:
+ assert self.parser is not None
self.parser.error("job ID (or --owner) not specified")
@overrides(Script.body)
- def body(self, client: ProtocolClient, args: argparse.Namespace):
+ def body(self, client: ProtocolClient, args: argparse.Namespace) -> int:
jid = self.get_job_id(client, args)
# Do as the user asked
diff --git a/spalloc_client/scripts/machine.py b/spalloc_client/scripts/machine.py
index 31bb625b0..fa1deb76c 100644
--- a/spalloc_client/scripts/machine.py
+++ b/spalloc_client/scripts/machine.py
@@ -32,19 +32,20 @@
from collections import defaultdict
import argparse
import sys
-from typing import Any, Callable, cast, Dict, List
+from typing import Any, cast, Dict, Iterator, List, Optional
from spinn_utilities.overrides import overrides
-from spinn_utilities.typing.json import JsonObjectArray
+from spinn_utilities.typing.json import JsonObject, JsonObjectArray
from spalloc_client import __version__, ProtocolClient
+from spalloc_client.spalloc_config import SpallocConfig
from spalloc_client.term import (
Terminal, render_table, render_definitions, render_boards, render_cells,
DEFAULT_BOARD_EDGES, TableRow, TableType)
from spalloc_client.scripts.support import Terminate, Script
-def generate_keys(alphabet="ABCDEFGHIJKLMNOPQRSTUVWXYZ"):
+def generate_keys(alphabet: str = "ABCDEFGHIJKLMNOPQRSTUVWXYZ") -> Iterator:
""" Generate ascending values in spreadsheet-column-name style.
For example, A, B, C, ..., Y, Z, AA, AB, AC...
@@ -58,7 +59,7 @@ def generate_keys(alphabet="ABCDEFGHIJKLMNOPQRSTUVWXYZ"):
def list_machines(t: Terminal, machines: JsonObjectArray,
- jobs: JsonObjectArray):
+ jobs: JsonObjectArray) -> None:
""" Display a table summarising the available machines and their load.
Parameters
@@ -101,7 +102,7 @@ def list_machines(t: Terminal, machines: JsonObjectArray,
print(render_table(table))
-def _get_machine(machines, machine_name):
+def _get_machine(machines: JsonObjectArray, machine_name: str) -> JsonObject:
for machine in machines:
if machine["name"] == machine_name:
return machine
@@ -110,7 +111,7 @@ def _get_machine(machines, machine_name):
def show_machine(t: Terminal, machines: JsonObjectArray, jobs: JsonObjectArray,
- machine_name: str, compact: bool = False):
+ machine_name: str, compact: bool = False) -> None:
""" Display a more detailed overview of an individual machine.
Parameters
@@ -149,28 +150,28 @@ def show_machine(t: Terminal, machines: JsonObjectArray, jobs: JsonObjectArray,
if job["allocated_machine_name"] == machine_name:
displayed_jobs.append(job)
job["key"] = next(job_key_generator)
- job["colour"] = job_colours[
- cast(int, job["job_id"]) % len(job_colours)]
# Calculate machine stats
- num_boards = ((machine["width"] * machine["height"] * 3) -
- len(machine["dead_boards"]))
+ num_boards = ((cast(int, machine["width"]) *
+ cast(int, machine["height"]) * 3) -
+ len(cast(list, machine["dead_boards"])))
num_in_use = sum(map(len, (cast(list, job["boards"])
for job in displayed_jobs)))
# Show general machine information
info = dict()
info["Name"] = machine["name"]
- info["Tags"] = ", ".join(machine["tags"])
+ info["Tags"] = ", ".join(cast(list, machine["tags"]))
info["In-use"] = f"{num_in_use} of {num_boards}"
info["Jobs"] = len(displayed_jobs)
print(render_definitions(info))
# Draw diagram of machine
- dead_boards = set((x, y, z) for x, y, z in machine["dead_boards"])
+ dead_boards = set((x, y, z) for x, y, z in cast(
+ list, machine["dead_boards"]))
board_groups = [(list([(x, y, z)
- for x in range(machine["width"])
- for y in range(machine["height"])
+ for x in range(cast(int, machine["width"]))
+ for y in range(cast(int, machine["height"]))
for z in range(3)
if (x, y, z) not in dead_boards]),
t.dim(" . "), # Label
@@ -184,7 +185,8 @@ def show_machine(t: Terminal, machines: JsonObjectArray, jobs: JsonObjectArray,
assert isinstance(board, list)
(x, y, z) = board
boards.append((cast(int, x), cast(int, y), cast(int, z)))
- colour_func = cast(Callable, job["colour"])
+ colour_func = job_colours[
+ cast(int, job["job_id"]) % len(job_colours)]
board_groups.append((
boards,
colour_func(cast(str, job["key"]).center(3)), # Label
@@ -192,7 +194,7 @@ def show_machine(t: Terminal, machines: JsonObjectArray, jobs: JsonObjectArray,
tuple(map(t.bright, DEFAULT_BOARD_EDGES)) # Outer
))
print("")
- print(render_boards(board_groups, machine["dead_links"],
+ print(render_boards(board_groups, cast(list, machine["dead_links"]),
tuple(map(t.red, DEFAULT_BOARD_EDGES))))
# Produce table showing jobs on machine
if compact:
@@ -201,8 +203,10 @@ def show_machine(t: Terminal, machines: JsonObjectArray, jobs: JsonObjectArray,
for job in displayed_jobs:
key = cast(str, job["key"])
job_id = str(job["job_id"])
+ colour_func = job_colours[
+ cast(int, job["job_id"]) % len(job_colours)]
cells.append((len(key) + len(job_id) + 1,
- f"{cast(Callable, job['colour'])(key)}:{job_id}"))
+ f"{colour_func(key)}:{job_id}"))
print("")
print(render_cells(cells))
else:
@@ -217,8 +221,10 @@ def show_machine(t: Terminal, machines: JsonObjectArray, jobs: JsonObjectArray,
owner = str(job["owner"])
if "keepalivehost" in job and job["keepalivehost"] is not None:
owner += f" {job['keepalivehost']}"
+ colour_func = job_colours[
+ cast(int, job["job_id"]) % len(job_colours)]
table_row: TableRow = [
- (cast(Callable, job["colour"]), cast(str, job["key"])),
+ (colour_func, cast(str, job["key"])),
cast(int, job["job_id"]),
len(cast(list, job["boards"])),
owner,
@@ -233,12 +239,13 @@ class ListMachinesScript(Script):
A Script object to get information from a spalloc machine.
"""
- def __init__(self):
+ def __init__(self) -> None:
super().__init__()
- self.parser = None
+ self.parser: Optional[argparse.ArgumentParser] = None
- def get_and_display_machine_info(self, client: ProtocolClient,
- args: argparse.Namespace, t: Terminal):
+ def get_and_display_machine_info(
+ self, client: ProtocolClient,
+ args: argparse.Namespace, t: Terminal) -> None:
""" Gets and displays info for the machine(s) """
# Get all information
machines = client.list_machines(timeout=args.timeout)
@@ -251,7 +258,7 @@ def get_and_display_machine_info(self, client: ProtocolClient,
show_machine(t, machines, jobs, args.machine, not args.detailed)
@overrides(Script.get_parser)
- def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
+ def get_parser(self, cfg: SpallocConfig) -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Get the state of individual machines.")
parser.add_argument(
@@ -269,13 +276,15 @@ def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
return parser
@overrides(Script.verify_arguments)
- def verify_arguments(self, args: argparse.Namespace):
+ def verify_arguments(self, args: argparse.Namespace) -> None:
# Fail if --detailed used without specifying machine
if args.machine is None and args.detailed:
+ assert self.parser is not None
self.parser.error(
"--detailed only works when a specific machine is specified")
- def one_shot(self, client: ProtocolClient, args: argparse.Namespace):
+ def one_shot(self, client: ProtocolClient,
+ args: argparse.Namespace) -> None:
"""
Display the machine info once
"""
@@ -283,7 +292,8 @@ def one_shot(self, client: ProtocolClient, args: argparse.Namespace):
# Get all information and display accordingly
self.get_and_display_machine_info(client, args, t)
- def recurring(self, client: ProtocolClient, args: argparse.Namespace):
+ def recurring(self, client: ProtocolClient,
+ args: argparse.Namespace) -> None:
"""
Repeatedly display the machine info
"""
@@ -307,7 +317,7 @@ def recurring(self, client: ProtocolClient, args: argparse.Namespace):
print("")
@overrides(Script.body)
- def body(self, client: ProtocolClient, args: argparse.Namespace):
+ def body(self, client: ProtocolClient, args: argparse.Namespace) -> int:
if args.watch:
self.recurring(client, args)
else:
diff --git a/spalloc_client/scripts/ps.py b/spalloc_client/scripts/ps.py
index 0f679460b..f29391730 100644
--- a/spalloc_client/scripts/ps.py
+++ b/spalloc_client/scripts/ps.py
@@ -27,19 +27,20 @@
import argparse
from collections.abc import Sized
import sys
-from typing import Any, cast, Dict, Union
+from typing import cast, Union
from spinn_utilities.overrides import overrides
from spinn_utilities.typing.json import JsonObjectArray
from spalloc_client import __version__, JobState, ProtocolClient
+from spalloc_client.spalloc_config import SpallocConfig
from spalloc_client.term import Terminal, render_table, TableColumn, TableType
from spalloc_client._utils import render_timestamp
from .support import Script
def render_job_list(t: Terminal, jobs: JsonObjectArray,
- args: argparse.Namespace):
+ args: argparse.Namespace) -> str:
""" Return a human-readable process listing.
Parameters
@@ -99,7 +100,7 @@ def render_job_list(t: Terminal, jobs: JsonObjectArray,
else:
num_boards = ""
# Format start time
- timestamp = render_timestamp(job["start_time"])
+ timestamp = render_timestamp(cast(float, job["start_time"]))
if job["allocated_machine_name"] is not None:
machine_name = str(job["allocated_machine_name"])
@@ -128,7 +129,7 @@ class ProcessListScript(Script):
An object form Job scripts.
"""
@overrides(Script.get_parser)
- def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
+ def get_parser(self, cfg: SpallocConfig) -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="List all active jobs.")
parser.add_argument(
"--version", "-V", action="version", version=__version__)
@@ -143,13 +144,15 @@ def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
help="list only jobs belonging to a particular owner")
return parser
- def one_shot(self, client: ProtocolClient, args: argparse.Namespace):
+ def one_shot(self, client: ProtocolClient,
+ args: argparse.Namespace) -> None:
""" 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: ProtocolClient, args: argparse.Namespace):
+ def recurring(
+ self, client: ProtocolClient, args: argparse.Namespace) -> None:
""" Repeatedly gets info on the job list. """
client.notify_job(timeout=args.timeout)
t = Terminal(stream=sys.stderr)
@@ -168,13 +171,14 @@ def recurring(self, client: ProtocolClient, args: argparse.Namespace):
print("")
@overrides(Script.body)
- def body(self, client: ProtocolClient, args: argparse.Namespace):
+ def body(self, client: ProtocolClient, args: argparse.Namespace) -> int:
if args.watch:
self.recurring(client, args)
else:
self.one_shot(client, args)
+ return 0
- def verify_arguments(self, args: argparse.Namespace):
+ def verify_arguments(self, args: argparse.Namespace) -> None:
pass
diff --git a/spalloc_client/scripts/support.py b/spalloc_client/scripts/support.py
index b31cbd942..c93008db2 100644
--- a/spalloc_client/scripts/support.py
+++ b/spalloc_client/scripts/support.py
@@ -14,11 +14,12 @@
from argparse import ArgumentParser, Namespace
import sys
-from typing import Any, Dict, Optional
+from typing import Any, Optional
from spinn_utilities.abstract_base import AbstractBase, abstractmethod
from spalloc_client import (
- config, ProtocolClient, ProtocolError, ProtocolTimeoutError,
+ ProtocolClient, ProtocolError, ProtocolTimeoutError,
SpallocServerException)
+from spalloc_client.spalloc_config import SpallocConfig
# The acceptable range of server version numbers
VERSION_RANGE_START = (0, 1, 0)
@@ -33,14 +34,14 @@ def __init__(self, code: int, message: Optional[str] = None):
self._code = code
self._msg = message
- def exit(self):
+ def exit(self) -> None:
""" Exit the program after printing an error msg. """
if self._msg is not None:
sys.stderr.write(self._msg + "\n")
sys.exit(self._code)
-def version_verify(client: ProtocolClient, timeout: Optional[int]):
+def version_verify(client: ProtocolClient, timeout: Optional[float]) -> None:
"""
Verify that the current version of the client is compatible
"""
@@ -52,50 +53,51 @@ def version_verify(client: ProtocolClient, timeout: Optional[int]):
class Script(object, metaclass=AbstractBase):
""" Base class of various Script Objects. """
- def __init__(self):
+ def __init__(self) -> None:
self.client_factory = ProtocolClient
- def get_parser(self, cfg: Dict[str, Any]) -> ArgumentParser:
+ def get_parser(self, cfg: SpallocConfig) -> ArgumentParser:
""" Return a set-up instance of :py:class:`argparse.ArgumentParser`
"""
raise NotImplementedError
@abstractmethod
- def verify_arguments(self, args: Namespace):
+ def verify_arguments(self, args: Namespace) -> None:
""" Check the arguments for sanity and do any second-stage parsing\
required.
"""
@abstractmethod
- def body(self, client: ProtocolClient, args: Namespace):
+ def body(self, client: ProtocolClient, args: Namespace) -> int:
""" How to do the processing of the script once a client has been\
obtained and verified to be compatible.
"""
+ raise NotImplementedError
def build_server_arg_group(self, server_args: Any,
- cfg: Dict[str, object]):
+ cfg: SpallocConfig) -> None:
"""
Adds a few more arguments
:param argparse._ArguementGroup server_args:
"""
server_args.add_argument(
- "--hostname", "-H", default=cfg["hostname"],
+ "--hostname", "-H", default=cfg.hostname,
help="hostname or IP of the spalloc server (default: %(default)s)")
server_args.add_argument(
- "--port", "-P", default=cfg["port"], type=int,
+ "--port", "-P", default=cfg.port, type=int,
help="port number of the spalloc server (default: %(default)s)")
server_args.add_argument(
- "--timeout", default=cfg["timeout"], type=float, metavar="SECONDS",
+ "--timeout", default=cfg.timeout, type=float, metavar="SECONDS",
help="seconds to wait for a response from the server (default: "
"%(default)s)")
server_args.add_argument(
- "--ignore_version", default=cfg["ignore_version"], type=bool,
+ "--ignore_version", default=cfg.ignore_version, type=bool,
help="Ignore the server version (WARNING: could result in errors) "
"default: %(default)s)")
- def __call__(self, argv=None):
- cfg = config.read_config()
+ def __call__(self, argv: Optional[str] = None) -> int:
+ cfg = SpallocConfig()
parser = self.get_parser(cfg)
server_args = parser.add_argument_group("spalloc server arguments")
self.build_server_arg_group(server_args, cfg)
@@ -120,3 +122,4 @@ def __call__(self, argv=None):
return 1
except Terminate as t:
t.exit()
+ return 1
diff --git a/spalloc_client/scripts/where_is.py b/spalloc_client/scripts/where_is.py
index eaac72c18..3c4885cf6 100644
--- a/spalloc_client/scripts/where_is.py
+++ b/spalloc_client/scripts/where_is.py
@@ -66,13 +66,14 @@
"""
import argparse
import sys
-from typing import Any, cast, Dict
+from typing import Any, cast, Dict, Optional
from spinn_utilities.overrides import overrides
from spalloc_client import __version__, ProtocolClient
from spalloc_client.term import render_definitions
from spalloc_client.scripts.support import Terminate, Script
+from spalloc_client.spalloc_config import SpallocConfig
class WhereIsScript(Script):
@@ -80,14 +81,14 @@ class WhereIsScript(Script):
An script object to find where a board is
"""
- def __init__(self):
+ def __init__(self) -> None:
super().__init__()
- self.parser = None
- self.where_is_kwargs = None
- self.show_board_chip = None
+ self.parser: Optional[argparse.ArgumentParser] = None
+ self.where_is_kwargs: Optional[dict] = None
+ self.show_board_chip = False
@overrides(Script.get_parser)
- def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
+ def get_parser(self, cfg: SpallocConfig) -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Find out the location (physical or logical) of a "
"chip or board.")
@@ -114,7 +115,7 @@ def get_parser(self, cfg: Dict[str, Any]) -> argparse.ArgumentParser:
return parser
@overrides(Script.verify_arguments)
- def verify_arguments(self, args: argparse.Namespace):
+ def verify_arguments(self, args: argparse.Namespace) -> None:
try:
if args.board:
machine, x, y, z = args.board
@@ -151,17 +152,19 @@ def verify_arguments(self, args: argparse.Namespace):
}
self.show_board_chip = True
except ValueError as e:
+ assert self.parser is not None
self.parser.error(f"Error: {e}")
@overrides(Script.body)
- def body(self, client: ProtocolClient, args: argparse.Namespace):
+ def body(self, client: ProtocolClient, args: argparse.Namespace) -> int:
# Ask the server
+ assert self.where_is_kwargs is not None
location = client.where_is(**self.where_is_kwargs)
if location is None:
raise Terminate(4, "No boards at the specified location")
out: Dict[str, Any] = dict()
- out["Machine"] = location["machine"]
+ out["Machine"] = cast(str, location["machine"])
cabinet, frame, board = cast(list, location["physical"])
out["Physical location"] = (
f"Cabinet {cabinet}, Frame {frame}, Board {board}")
@@ -175,6 +178,7 @@ def body(self, client: ProtocolClient, args: argparse.Namespace):
out["Coordinates within job"] = tuple(
cast(list, location["job_chip"]))
print(render_definitions(out))
+ return 0
main = WhereIsScript()
diff --git a/spalloc_client/config.py b/spalloc_client/spalloc_config.py
similarity index 52%
rename from spalloc_client/config.py
rename to spalloc_client/spalloc_config.py
index 445097317..bd2c51e39 100644
--- a/spalloc_client/config.py
+++ b/spalloc_client/spalloc_config.py
@@ -81,7 +81,7 @@
"""
import configparser
import os.path
-from typing import Any, Dict, List, Optional
+from typing import List, Optional
import appdirs
@@ -117,82 +117,153 @@
"ignore_version": "False"}
-def _read_none_or_float(parser, option):
+def _read_none_or_float(
+ parser: configparser.ConfigParser, option: str) -> Optional[float]:
if parser.get(SECTION, option) == "None":
return None
return parser.getfloat(SECTION, option)
-def _read_none_or_int(parser, option):
+def _read_none_or_int(
+ parser: configparser.ConfigParser, option: str) -> Optional[int]:
if parser.get(SECTION, option) == "None":
return None
return parser.getint(SECTION, option)
-def _read_any_str(parser, option):
+def _read_any_str(
+ parser: configparser.ConfigParser, option: str) -> Optional[str]:
try:
return parser.get(SECTION, option)
except configparser.NoOptionError:
return None
-def _read_none_or_str(parser, option):
+def _read_none_or_str(
+ parser: configparser.ConfigParser, option: str) -> Optional[str]:
if parser.get(SECTION, option) == "None":
return None
return parser.get(SECTION, option)
-def read_config(filenames: Optional[List[str]] = None) -> Dict[str, Any]:
- """ Attempt to read local configuration files to determine spalloc client
- settings.
-
- Parameters
- ----------
- filenames : [str, ...]
- Filenames to attempt to read. Later config file have higher priority.
-
- Returns
- -------
- dict
- The configuration loaded.
- """
- if filenames is None: # pragma: no cover
- filenames = SEARCH_PATH
- parser = configparser.ConfigParser()
-
- # Set default config values (NB: No read_dict in Python 2.7)
- parser.add_section(SECTION)
- for key, value in DEFAULT_CONFIG.items():
- parser.set(SECTION, key, value)
-
- # Attempt to read from each possible file location in turn
- for filename in filenames:
- try:
- with open(filename, "r", encoding="utf-8") as f:
- parser.read_file(f, filename)
- except (IOError, OSError):
- # File did not exist, keep trying
- pass
-
- cfg = {
- "hostname": _read_any_str(parser, "hostname"),
- "owner": _read_any_str(parser, "owner"),
- "port": parser.getint(SECTION, "port"),
- "keepalive": _read_none_or_float(parser, "keepalive"),
- "reconnect_delay": parser.getfloat(SECTION, "reconnect_delay"),
- "timeout": _read_none_or_float(parser, "timeout"),
- "machine": _read_none_or_str(parser, "machine"),
- "min_ratio": parser.getfloat(SECTION, "min_ratio"),
- "max_dead_boards": _read_none_or_int(parser, "max_dead_boards"),
- "max_dead_links": _read_none_or_int(parser, "max_dead_links"),
- "require_torus": parser.getboolean(SECTION, "require_torus"),
- "ignore_version": parser.getboolean(SECTION, "ignore_version")}
-
- tags = _read_none_or_str(parser, "tags")
- cfg["tags"] = None if tags is None else list(
- map(str.strip, tags.split(",")))
-
- return cfg
+class SpallocConfig(object):
+ """ Typed configs """
+
+ __slots__ = ("_hostname", "_ignore_version", "_keepalive", "_machine",
+ "_max_dead_boards", "_max_dead_links", "_min_ratio",
+ "_owner", "_port", "_reconnect_delay", "_require_torus",
+ "_tags", "_timeout")
+
+ def __init__(self, filenames: Optional[List[str]] = None):
+ """ Attempt to read local configuration files
+ to determine spalloc client settings.
+
+ Parameters
+ ----------
+ filenames : [str, ...]
+ Filenames to attempt to read.
+ Later config file have higher priority.
+
+ """
+ if filenames is None: # pragma: no cover
+ filenames = SEARCH_PATH
+ parser = configparser.ConfigParser()
+
+ # Set default config values (NB: No read_dict in Python 2.7)
+ parser.add_section(SECTION)
+ for key, value in DEFAULT_CONFIG.items():
+ parser.set(SECTION, key, value)
+
+ # Attempt to read from each possible file location in turn
+ for filename in filenames:
+ try:
+ with open(filename, "r", encoding="utf-8") as f:
+ parser.read_file(f, filename)
+ except (IOError, OSError):
+ # File did not exist, keep trying
+ pass
+
+ self._hostname = _read_any_str(parser, "hostname")
+ self._owner = _read_any_str(parser, "owner")
+ self._port = parser.getint(SECTION, "port")
+ self._keepalive = _read_none_or_float(parser, "keepalive")
+ self._reconnect_delay = parser.getfloat(SECTION, "reconnect_delay")
+ self._timeout = _read_none_or_float(parser, "timeout")
+ self._machine = _read_none_or_str(parser, "machine")
+ self._min_ratio = parser.getfloat(SECTION, "min_ratio")
+ self._max_dead_boards = _read_none_or_int(parser, "max_dead_boards")
+ self._max_dead_links = _read_none_or_int(parser, "max_dead_links")
+ self._require_torus = parser.getboolean(SECTION, "require_torus")
+ self._ignore_version = parser.getboolean(SECTION, "ignore_version")
+
+ tags = _read_none_or_str(parser, "tags")
+ self._tags = None if tags is None else list(
+ map(str.strip, tags.split(",")))
+
+ @property
+ def hostname(self) -> Optional[str]:
+ """ Name of the spalloc server if specified """
+ return self._hostname
+
+ @property
+ def ignore_version(self) -> bool:
+ """ Flag to say version can be ignored """
+ return self._ignore_version
+
+ @property
+ def keepalive(self) -> Optional[float]:
+ """ Time to keep job allive """
+ return self._keepalive
+
+ @property
+ def machine(self) -> Optional[str]:
+ """ Name of the spalloc machine to use"""
+ return self._machine
+
+ @property
+ def max_dead_boards(self) -> Optional[int]:
+ """ How many dead boards are allowed in the job"""
+ return self._max_dead_boards
+
+ @property
+ def max_dead_links(self) -> Optional[int]:
+ """ How many dead links are allowed in the Job"""
+ return self._max_dead_links
+
+ @property
+ def min_ratio(self) -> float:
+ """ Min ratio"""
+ return self._min_ratio
+
+ @property
+ def owner(self) -> Optional[str]:
+ """ Owner to assign job to """
+ return self._owner
+
+ @property
+ def port(self) -> int:
+ """ Spalloc server port"""
+ return self._port
+
+ @property
+ def reconnect_delay(self) -> float:
+ """ Reconnect delay """
+ return self._reconnect_delay
+
+ @property
+ def require_torus(self) -> bool:
+ """ Flag to say a torus is required """
+ return self._require_torus
+
+ @property
+ def tags(self) -> Optional[List[str]]:
+ """ List of tags """
+ return self._tags
+
+ @property
+ def timeout(self) -> Optional[float]:
+ """ Time before a command should timeout """
+ return self._timeout
if __name__ == "__main__": # pragma: no cover
diff --git a/spalloc_client/term.py b/spalloc_client/term.py
index 64a1ed0fe..e9285cb9c 100644
--- a/spalloc_client/term.py
+++ b/spalloc_client/term.py
@@ -21,7 +21,8 @@
from collections import defaultdict
from enum import IntEnum
from functools import partial
-from typing import Callable, Dict, Iterable, List, Tuple, Union
+from typing import (
+ Callable, Dict, Iterable, List, Optional, TextIO, Tuple, Union)
from typing_extensions import TypeAlias
# pylint: disable=wrong-spelling-in-docstring
@@ -100,7 +101,8 @@ class Terminal(object):
Is colour enabled?
"""
- def __init__(self, stream=None, force=None):
+ def __init__(self, stream: Optional[TextIO] = None,
+ force: Optional[bool] = None):
"""
Parameters
----------
@@ -120,18 +122,18 @@ def __init__(self, stream=None, force=None):
self._location_saved = False
- def __call__(self, string):
+ def __call__(self, string: str) -> str:
""" If enabled, passes through the given value, otherwise passes\
through an empty string.
"""
return string if self.enabled else ""
- def clear_screen(self):
+ def clear_screen(self) -> str:
""" Clear the screen and reset cursor to top-left corner.
"""
return self("\033[2J\033[;H")
- def update(self, string: str = "", start_again: bool = False):
+ def update(self, string: str = "", start_again: bool = False) -> str:
""" Print before a line and it will replace the previous line prefixed\
with :py:meth:`.update`.
@@ -153,7 +155,7 @@ def update(self, string: str = "", start_again: bool = False):
# Restore to previous location and clear line.
return "".join((self("\0338\033[K"), str(string)))
- def set_attrs(self, attrs=tuple()):
+ def set_attrs(self, attrs: List) -> str:
""" Construct an ANSI control sequence which sets the given attribute\
numbers.
"""
@@ -161,7 +163,8 @@ def set_attrs(self, attrs=tuple()):
return ""
return self(f"\033[{';'.join(str(attr) for attr in attrs)}m")
- def wrap(self, string=None, pre="", post=""):
+ def wrap(self, string: Optional[str] = None,
+ pre: str = "", post: str = "") -> str:
""" Wrap a string in the suppled pre and post strings or just print\
the pre string if no string given.
"""
@@ -169,7 +172,7 @@ def wrap(self, string=None, pre="", post=""):
return pre
return "".join((pre, str(string), post))
- def __getattr__(self, name):
+ def __getattr__(self, name: str) -> partial:
""" Implements all the 'magic' style methods.
"""
attrs = []
@@ -188,7 +191,7 @@ def __getattr__(self, name):
post=self("\033[0m"))
-def render_table(table: TableType, column_sep: str = " "):
+def render_table(table: TableType, column_sep: str = " ") -> str:
""" Render an ASCII table with optional ANSI escape codes.
An example table::
@@ -264,7 +267,7 @@ def render_table(table: TableType, column_sep: str = " "):
return "\n".join(column_sep.join(row).rstrip() for row in out)
-def render_definitions(definitions, separator=": "):
+def render_definitions(definitions: Dict, separator: str = ": ") -> str:
""" Render a definition list.
Such a list looks like this::
@@ -294,7 +297,7 @@ def render_definitions(definitions, separator=": "):
for key, value in definitions.items())
-def _board_to_cartesian(x, y, z):
+def _board_to_cartesian(x: int, y: int, z: int) -> Tuple[int, int]:
r""" Translate from logical board coordinates (x, y, z) into Cartesian
coordinates for printing hexagons.
@@ -374,9 +377,13 @@ def _board_to_cartesian(x, y, z):
"""
-def render_boards(board_groups, dead_links=frozenset(),
- dead_edge=("XXX", "X", "X"),
- blank_label=" ", blank_edge=(" ", " ", " ")):
+def render_boards(
+ board_groups: List[Tuple[List[Tuple[int, int, int]], str,
+ Tuple[str, str, str], Tuple[str, str, str]]],
+ dead_links: List,
+ dead_edge: Tuple[str, str, str] = ("XXX", "X", "X"),
+ blank_label: str = " ",
+ blank_edge: Tuple[str, str, str] = (" ", " ", " ")) -> str:
r""" Render an ASCII art diagram of a set of boards with sets of boards.
For example::
@@ -396,7 +403,7 @@ def render_boards(board_groups, dead_links=frozenset(),
which are to be used for the inner and outer board edges respectively.
Board groups are drawn sequentially with later board groups obscuring
earlier ones when their edges or boards overlap.
- dead_links : set([(x, y, z, link), ...])
+ dead_links : list([(x, y, z, link), ...])
Enumeration of all dead links. These links are re-drawn in the style
defined by the dead_edge argument after all board groups have been
drawn.
@@ -420,9 +427,9 @@ def render_boards(board_groups, dead_links=frozenset(),
# non-existent boards
all_boards = set()
- for boards, label, edge_inner, edge_outer in board_groups:
+ for _boards, label, edge_inner, edge_outer in board_groups:
# Convert to Cartesian coordinates
- boards = set(_board_to_cartesian(x, y, z) for x, y, z in boards)
+ boards = set(_board_to_cartesian(x, y, z) for x, y, z in _boards)
all_boards.update(boards)
# Set board labels and basic edge style
@@ -486,7 +493,8 @@ def render_boards(board_groups, dead_links=frozenset(),
return "\n".join(filter(None, map(str.rstrip, out)))
-def render_cells(cells, width=80, col_spacing=2):
+def render_cells(cells: List[Tuple[int, str]], width: int = 80,
+ col_spacing: int = 2) -> str:
""" Given a list of short (~10 char) strings, display these aligned in\
columns.
diff --git a/tests/conftest.py b/tests/conftest.py
index e809e22b9..59b95d22e 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -19,7 +19,7 @@
import pytest
from mock import Mock # type: ignore[import]
from spalloc_client import ProtocolClient
-from spalloc_client.config import SEARCH_PATH
+from spalloc_client.spalloc_config import SEARCH_PATH
from .common import MockServer
diff --git a/tests/scripts/test_job_script.py b/tests/scripts/test_job_script.py
index 4407f55b8..bf7ac83ba 100644
--- a/tests/scripts/test_job_script.py
+++ b/tests/scripts/test_job_script.py
@@ -16,7 +16,7 @@
import pytest
from mock import Mock, MagicMock # type: ignore[import]
from spalloc_client import JobState, ProtocolError
-from spalloc_client.config import TIMEOUT
+from spalloc_client.spalloc_config import TIMEOUT
from spalloc_client.term import Terminal
from spalloc_client.scripts.job import (
show_job_info, watch_job, power_job, list_ips, destroy_job, main)
diff --git a/tests/test_config.py b/tests/test_config.py
index 77c2e48d3..fa44e6b25 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -16,7 +16,7 @@
import shutil
import os.path
import pytest
-from spalloc_client.config import read_config, TIMEOUT
+from spalloc_client.spalloc_config import SpallocConfig
@pytest.yield_fixture
@@ -41,59 +41,9 @@ def test_priority(tempdir):
with open(f2, "w") as f:
f.write("[spalloc]\nport=321\ntags=qux")
- cfg = read_config([f1, f2])
+ cfg = SpallocConfig([f1, f2])
- assert cfg["port"] == 321
- assert cfg["reconnect_delay"] == 5.0
- assert cfg["hostname"] == "bar"
- assert cfg["tags"] == ["qux"]
-
-
-@pytest.mark.parametrize(
- "option_name,config_value,value",
- [("hostname", None, None),
- ("hostname", "foo", "foo"),
- ("port", None, 22244),
- ("port", "1234", 1234),
- ("owner", None, None),
- ("owner", "foo", "foo"),
- ("keepalive", None, 60.0),
- ("keepalive", "None", None),
- ("keepalive", "3.0", 3.0),
- ("reconnect_delay", None, 5.0),
- ("reconnect_delay", "3.0", 3.0),
- ("timeout", None, TIMEOUT),
- ("timeout", "None", None),
- ("timeout", "3.0", 3.0),
- ("machine", None, None),
- ("machine", "None", None),
- ("machine", "foo", "foo"),
- ("tags", None, None),
- ("tags", "None", None),
- ("tags", "foo", ["foo"]),
- ("tags", "one, two , three", ["one", "two", "three"]),
- ("min_ratio", None, 0.333),
- ("min_ratio", "1.0", 1.0),
- ("max_dead_boards", None, 0),
- ("max_dead_boards", "None", None),
- ("max_dead_boards", "3", 3),
- ("max_dead_links", None, None),
- ("max_dead_links", "None", None),
- ("max_dead_links", "3", 3),
- ("require_torus", None, False),
- ("require_torus", "False", False),
- ("require_torus", "True", True)])
-def test_options(filename, option_name, config_value, value):
- # Test all config options.
-
- # Write config file (omitting the config value if None, e.g. to test
- # default value)
- with open(filename, "w") as f:
- f.write("[spalloc]\n")
- if config_value is not None:
- f.write("{}={}".format(option_name, config_value))
-
- cfg = read_config([filename])
-
- assert option_name in cfg
- assert cfg[option_name] == value
+ assert cfg.port == 321
+ assert cfg.reconnect_delay == 5.0
+ assert cfg.hostname == "bar"
+ assert cfg.tags == ["qux"]
diff --git a/tests/test_job.py b/tests/test_job.py
index 974bd375d..7220cf0ff 100644
--- a/tests/test_job.py
+++ b/tests/test_job.py
@@ -75,10 +75,11 @@ def test_args_from_config(self, basic_config_file, client,
basic_job_kwargs.pop("hostname")
basic_job_kwargs.pop("port")
basic_job_kwargs.pop("reconnect_delay")
+ timeout = basic_job_kwargs.pop("timeout")
assert len(client.create_job.mock_calls) == 1
args = client.create_job.mock_calls[0][1]
kwargs = client.create_job.mock_calls[0][2]
- assert args == tuple()
+ assert args == (timeout,)
assert kwargs == basic_job_kwargs
finally:
j.close()
@@ -107,9 +108,8 @@ def test_override_config(self, basic_config_file, basic_job_kwargs,
assert len(client.create_job.mock_calls) == 1
args = client.create_job.mock_calls[0][1]
kwargs = client.create_job.mock_calls[0][2]
- assert args == tuple()
- assert kwargs == dict(timeout=0.2,
- owner="mossblaser",
+ assert args == (0.2,)
+ assert kwargs == dict(owner="mossblaser",
keepalive=0.3,
machine=None,
tags=["baz", "quz"],
diff --git a/tests/test_protocol_client.py b/tests/test_protocol_client.py
index 478b7e538..e1ad3c1a1 100644
--- a/tests/test_protocol_client.py
+++ b/tests/test_protocol_client.py
@@ -19,8 +19,7 @@
import pytest
from mock import Mock # type: ignore[import]
from spalloc_client import (
- ProtocolClient, SpallocServerException, ProtocolTimeoutError,
- ProtocolError)
+ ProtocolClient, SpallocServerException, ProtocolTimeoutError)
from .common import MockServer
logging.basicConfig(level=logging.DEBUG)
@@ -159,25 +158,27 @@ def test_send_json_fails(c):
def test_call(c, s, bg_accept):
c.connect()
bg_accept.join()
+ no_timeout = None
# Basic calls should work
s.send({"return": "Woo"})
- assert c.call("foo", 1, bar=2) == "Woo"
+ assert c.call("foo", no_timeout, 1, bar=2) == "Woo"
assert s.recv() == {"command": "foo", "args": [1], "kwargs": {"bar": 2}}
# Should be able to cope with notifications arriving before return value
s.send({"notification": 1})
s.send({"notification": 2})
s.send({"return": "Woo"})
- assert c.call("foo", 1, bar=2) == "Woo"
+ assert c.call("foo", no_timeout, 1, bar=2) == "Woo"
assert s.recv() == {"command": "foo", "args": [1], "kwargs": {"bar": 2}}
assert list(c._notifications) == [{"notification": 1}, {"notification": 2}]
c._notifications.clear()
# Should be able to timeout immediately
before = time.time()
+ timeout = 0.1
with pytest.raises(ProtocolTimeoutError):
- c.call("foo", 1, bar=2, timeout=0.1)
+ c.call("foo", timeout, 1, bar=2)
after = time.time()
assert s.recv() == {"command": "foo", "args": [1], "kwargs": {"bar": 2}}
assert 0.1 < after - before < 0.2
@@ -185,8 +186,9 @@ def test_call(c, s, bg_accept):
# Should be able to timeout after getting a notification
s.send({"notification": 3})
before = time.time()
+ timeout = 0.1
with pytest.raises(ProtocolTimeoutError):
- c.call("foo", 1, bar=2, timeout=0.1)
+ c.call("foo", timeout, 1, bar=2)
after = time.time()
assert s.recv() == {"command": "foo", "args": [1], "kwargs": {"bar": 2}}
assert 0.1 < after - before < 0.2
@@ -195,13 +197,14 @@ def test_call(c, s, bg_accept):
# Exceptions should transfer
s.send({"exception": "something informative"})
with pytest.raises(SpallocServerException) as e:
- c.call("foo")
+ c.call("foo", no_timeout)
assert "something informative" in str(e.value)
def test_wait_for_notification(c, s, bg_accept):
c.connect()
bg_accept.join()
+ no_timeout = None
# Should be able to timeout
with pytest.raises(ProtocolTimeoutError):
@@ -214,7 +217,7 @@ def test_wait_for_notification(c, s, bg_accept):
s.send({"notification": 1})
s.send({"notification": 2})
s.send({"return": "Woo"})
- assert c.call("foo", 1, bar=2) == "Woo"
+ assert c.call("foo", no_timeout, 1, bar=2) == "Woo"
assert s.recv() == {"command": "foo", "args": [1], "kwargs": {"bar": 2}}
assert c.wait_for_notification() == {"notification": 1}
assert c.wait_for_notification() == {"notification": 2}
@@ -229,10 +232,14 @@ def test_commands_as_methods(c, s, bg_accept):
bg_accept.join()
s.send({"return": "Woo"})
- assert c.create_job(1, bar=2, owner="dummy") == "Woo"
- assert s.recv() == {
+ no_timeout = None
+ assert c.create_job(no_timeout, 1, keepalive=2, owner="dummy") == "Woo"
+ commands = s.recv()
+ commands["kwargs"] = {k: v for k, v in commands["kwargs"].items()
+ if v is not None}
+ assert commands == {
"command": "create_job", "args": [1], "kwargs": {
- "bar": 2, "owner": "dummy"}}
+ "keepalive": 2, "owner": "dummy"}}
# Should fail for arbitrary internal method names
with pytest.raises(AttributeError):
@@ -240,16 +247,3 @@ def test_commands_as_methods(c, s, bg_accept):
# Should fail for arbitrary external method names
with pytest.raises(AttributeError):
c.bar()
-
-
-def test_where_is_sanity(c):
- with pytest.raises(SpallocServerException):
- c.where_is(foo=1, bar=2)
- with pytest.raises(SpallocServerException):
- c.where_is(machine=1, x=2, y=3, z=4, foo=5)
- with pytest.raises(ProtocolError):
- c.where_is(machine=1, x=2, y=3, z=4)
- with pytest.raises(ProtocolError):
- c.where_is(machine=1, x=2, y=3, z=4, timeout=5)
- with pytest.raises(ProtocolError):
- c.where_is(machine=1, x=2, y=3, z=4, timeout=None)
diff --git a/tests/test_term.py b/tests/test_term.py
index b03628c57..ae07dba7f 100644
--- a/tests/test_term.py
+++ b/tests/test_term.py
@@ -77,7 +77,7 @@ def test_set_attr():
t = Terminal(force=True)
# Empty list
- assert t.set_attrs() == ""
+ assert t.set_attrs([]) == ""
# Single item
assert t.set_attrs([1]) == "\033[1m"
@@ -87,7 +87,7 @@ def test_set_attr():
# When disabled should do nothing
t.enabled = False
- assert t.set_attrs() == ""
+ assert t.set_attrs([]) == ""
assert t.set_attrs([1]) == ""
assert t.set_attrs([1, 2, 3]) == ""
@@ -215,12 +215,12 @@ def test_render_definitions():
class TestRenderBoards(object):
def test_empty(self):
- assert render_boards([]) == ""
+ assert render_boards([], []) == ""
def test_single(self):
out = render_boards([
([(0, 0, 0)], "ABC", INNER_BOARD_EDGES, OUTER_BOARD_EDGES),
- ])
+ ], [])
assert out == (r" ___."
r"/ABC\."
r"\___/".replace(".", "\n"))
@@ -229,7 +229,7 @@ def test_three_boards(self):
out = render_boards([
([(0, 0, z) for z in range(3)], "ABC",
INNER_BOARD_EDGES, OUTER_BOARD_EDGES),
- ])
+ ], [])
assert out == (r" ___."
r"/ABC\___."
r"\===,ABC\."
@@ -244,7 +244,7 @@ def test_many_boards(self):
for y in range(2)
for z in range(3)], "ABC",
INNER_BOARD_EDGES, OUTER_BOARD_EDGES),
- ])
+ ], [])
assert out == (r" ___ ___."
r"/ABC\___/ABC\___."
r"\===,ABC`===,ABC\."
@@ -262,12 +262,12 @@ def test_dead_links(self):
for y in range(2)
for z in range(3)], "ABC",
INNER_BOARD_EDGES, OUTER_BOARD_EDGES),
- ], dead_links=set([
+ ], dead_links=[
(0, 0, 0, 0), # 0, 0, East
(0, 0, 0, 2), # 0, 0, North
(0, 0, 0, 1), # 0, 0, North East
(0, 0, 1, 4), # 1, 1, South West
- ]))
+ ])
assert out == (r" ___ ___."
r"/ABC\___/ABC\___."
r"\===,ABC`===,ABC\."
@@ -286,7 +286,7 @@ def test_multiple_board_groups(self):
for x in range(2) for y in range(2) for z in range(3)
if (x, y) != (0, 0)], "ABC",
INNER_BOARD_EDGES, OUTER_BOARD_EDGES),
- ])
+ ], [])
assert out == (r" ___ ___."
r"/ABC\___/ABC\___."
r"\===,ABC`===,ABC\."