Skip to content

Commit

Permalink
add typing
Browse files Browse the repository at this point in the history
more typing

flake8

typing fix

copyright
  • Loading branch information
Christian-B committed May 14, 2024
1 parent 0f8dfcf commit b89c3f4
Show file tree
Hide file tree
Showing 3 changed files with 67 additions and 8 deletions.
58 changes: 58 additions & 0 deletions manual_scripts/get_triad.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
# Copyright (c) 2014 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.

from spinn_utilities.config_holder import set_config
from spinnman.spalloc import SpallocClient
from spinnman.config_setup import unittest_setup


SPALLOC_URL = "https://spinnaker.cs.man.ac.uk/spalloc"
SPALLOC_USERNAME = ""
SPALLOC_PASSWORD = ""

SPALLOC_MACHINE = "SpiNNaker1M"

x = 0
y = 3
b = 0 # Must be 0 if requesting a rect
RECT = True
WIDTH = 1 # In triads!
HEIGHT = 1 # In triads!

unittest_setup()
set_config("Machine", "version",5)
client = SpallocClient(SPALLOC_URL, SPALLOC_USERNAME, SPALLOC_PASSWORD)
if RECT:
job = client.create_job_rect_at_board(
WIDTH, HEIGHT, triad=(x, y, b), machine_name=SPALLOC_MACHINE,
max_dead_boards=1)
else:
job = client.create_job_board(
triad=(x, y, b), machine_name=SPALLOC_MACHINE)
print(job)
print("Waiting until ready...")
with job:
job.wait_until_ready()
print(job.get_connections())

txrx = job.create_transceiver()
# This call is for testing and can be changed without notice!
dims = txrx._get_machine_dimensions()
print(f"{dims.height=}, {dims.width=}")

machine = txrx.get_machine_details()
print(machine)

input("Press Enter to release...")
client.close()#print(2)#print(2^(1/(2^1)))
16 changes: 9 additions & 7 deletions spinnman/spalloc/spalloc_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@

KEEP_ALIVE_PERIOND = 30


def fix_url(url: Any) -> str:
"""
Makes sure the url is the correct format.
Expand Down Expand Up @@ -199,7 +200,7 @@ def _create(self, create: Mapping[str, JsonValue],
def create_job(
self, num_boards: int = 1,
machine_name: Optional[str] = None,
keepalive = KEEP_ALIVE_PERIOND) -> SpallocJob:
keepalive: int = KEEP_ALIVE_PERIOND) -> SpallocJob:
return self._create({
"num-boards": int(num_boards),
"keepalive-interval": f"PT{int(keepalive)}S"
Expand All @@ -209,7 +210,7 @@ def create_job(
def create_job_rect(
self, width: int, height: int,
machine_name: Optional[str] = None,
keepalive = KEEP_ALIVE_PERIOND) -> SpallocJob:
keepalive: int = KEEP_ALIVE_PERIOND) -> SpallocJob:
return self._create({
"dimensions": {
"width": int(width),
Expand All @@ -224,7 +225,7 @@ def create_job_board(
physical: Optional[Tuple[int, int, int]] = None,
ip_address: Optional[str] = None,
machine_name: Optional[str] = None,
keepalive = KEEP_ALIVE_PERIOND) -> SpallocJob:
keepalive: int = KEEP_ALIVE_PERIOND) -> SpallocJob:
board: JsonObject
if triad:
x, y, z = triad
Expand All @@ -248,7 +249,8 @@ def create_job_rect_at_board(
triad: Optional[Tuple[int, int, int]] = None,
physical: Optional[Tuple[int, int, int]] = None,
ip_address: Optional[str] = None,
machine_name: Optional[str] = None, keepalive = KEEP_ALIVE_PERIOND,
machine_name: Optional[str] = None,
keepalive: int = KEEP_ALIVE_PERIOND,
max_dead_boards: int = 0) -> SpallocJob:
board: JsonObject
if triad:
Expand Down Expand Up @@ -498,11 +500,12 @@ def __init__(self, session: Session, job_handle: str):
logger.info("established job at {}", job_handle)
self.__machine_url = self._url + "machine"
self.__chip_url = self._url + "chip"
self._keepalive_url = self._url + "keepalive"
self._keepalive_url: Optional[str] = self._url + "keepalive"
self.__proxy_handle: Optional[WebSocket] = None
self.__proxy_thread: Optional[_ProxyReceiver] = None
self.__proxy_ping: Optional[_ProxyPing] = None
keep_alive = threading.Thread(target=self.__start_keepalive, daemon=True)
keep_alive = threading.Thread(
target=self.__start_keepalive, daemon=True)
keep_alive.start()

@overrides(SpallocJob.get_session_credentials_for_db)
Expand Down Expand Up @@ -649,7 +652,6 @@ def __keepalive(self) -> bool:
:rtype: bool
"""
if self._keepalive_url is None:
print("False")
return False
cookies, headers = self._session_credentials
headers["Content-Type"] = "text/plain; charset=UTF-8"
Expand Down
1 change: 0 additions & 1 deletion spinnman/spalloc/spalloc_job.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from contextlib import AbstractContextManager
from typing import Dict, Mapping, Optional, Tuple
from spinn_utilities.abstract_base import AbstractBase, abstractmethod
from spinnman.constants import SCP_SCAMP_PORT
Expand Down

0 comments on commit b89c3f4

Please sign in to comment.