From b89c3f444146e21dc4b318a84942a9337ed5edac Mon Sep 17 00:00:00 2001 From: "Christian Y. Brenninkmeijer" Date: Tue, 14 May 2024 10:26:08 +0100 Subject: [PATCH] add typing more typing flake8 typing fix copyright --- manual_scripts/get_triad.py | 58 ++++++++++++++++++++++++++++++ spinnman/spalloc/spalloc_client.py | 16 +++++---- spinnman/spalloc/spalloc_job.py | 1 - 3 files changed, 67 insertions(+), 8 deletions(-) create mode 100644 manual_scripts/get_triad.py diff --git a/manual_scripts/get_triad.py b/manual_scripts/get_triad.py new file mode 100644 index 000000000..27d1d6f8d --- /dev/null +++ b/manual_scripts/get_triad.py @@ -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))) \ No newline at end of file diff --git a/spinnman/spalloc/spalloc_client.py b/spinnman/spalloc/spalloc_client.py index 3503987cb..7e588004b 100644 --- a/spinnman/spalloc/spalloc_client.py +++ b/spinnman/spalloc/spalloc_client.py @@ -70,6 +70,7 @@ KEEP_ALIVE_PERIOND = 30 + def fix_url(url: Any) -> str: """ Makes sure the url is the correct format. @@ -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" @@ -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), @@ -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 @@ -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: @@ -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) @@ -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" diff --git a/spinnman/spalloc/spalloc_job.py b/spinnman/spalloc/spalloc_job.py index d7b9a907d..04ffa8715 100644 --- a/spinnman/spalloc/spalloc_job.py +++ b/spinnman/spalloc/spalloc_job.py @@ -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