6574804013

Committed 19 Oct 2023 12:47PM UTC coverage: 51.937% (+1.2%) from 50.777%

Build # 6574804013

Build Type

Pull #327

github

Committed by

Christian-B

Commit Message

typing changes

Pull Request Pull Request #327: Type Annotations and Checking

Run Details

105 of 1288 branches covered (0.0%)

Branch coverage included in aggregate %.

2375 of 2375 new or added lines in 180 files covered. (100.0%)

4775 of 8108 relevant lines covered (58.89%)

0.59 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

18.6

/spinnman/get_cores_in_run_state.py

# Copyright (c) 2017 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 is a script used to check the state of a SpiNNaker machine.
"""

import sys
import argparse
from spinn_utilities.config_holder import set_config
from spinnman.transceiver import create_transceiver_from_hostname
from spinn_machine import CoreSubsets, CoreSubset
from spinnman.board_test_configuration import BoardTestConfiguration
from spinnman.config_setup import unittest_setup
from spinnman.model.enums import CPUState
SCAMP_ID = 0
IGNORED_IDS = {SCAMP_ID, 16}  # WHY 16?


def get_cores_in_run_state(txrx, app_id, print_all_chips):
    """
    :param Transceiver txrx:
    :param int app_id:
    :param bool print_all_chips:
    """
    count_finished = txrx.get_core_state_count(app_id, CPUState.FINISHED)
    count_run = txrx.get_core_state_count(app_id, CPUState.RUNNING)
    print(f'running: {count_run} finished: {count_finished}')

    machine = txrx.get_machine_details()
    print(f'machine width: {machine.width} height: {machine.height}')
    if print_all_chips:
        print(f'machine chips: {list(machine.chips)}')

    all_cores = []
    for chip in machine.chips:
        all_cores.append(CoreSubset(chip.x, chip.y, range(1, 17)))

    all_cores = CoreSubsets(core_subsets=all_cores)

    cpu_infos = txrx.get_cpu_infos(
        all_cores,
        [CPUState.FINISHED, CPUState.RUNNING, CPUState.WATCHDOG], True)
    cores_finished = cpu_infos.infos_for_state(CPUState.FINISHED)
    cores_running = cpu_infos.infos_for_state(CPUState.RUNNING)
    cores_watchdog = cpu_infos.infos_for_state(CPUState.WATCHDOG)

    for (x, y, p), _ in cores_running:
        if p not in IGNORED_IDS:
            print(f'run core: {x} {y} {p}')

    for (x, y, p), _ in cores_finished:
        print(f'finished core: {x} {y} {p}')

    for (x, y, p), _ in cores_watchdog:
        print(f'watchdog core: {x} {y} {p}')


def _make_transceiver(host, version, bmp_names):
    """
    :param host:
        Host to use or `None` to use test configuration for all parameters
    :type host: str or None
    :param version: Board version to use (`None` defaults to 5 unless host is
        192.168.240.253 (spin 3)
    :type version: int or None
    :param bmp: bmp connection or `None` to auto detect (if applicable)
    :type bmp: str or None
    :rtype: Transceiver
    """
    if host is None:
        config = BoardTestConfiguration()
        config.set_up_remote_board()
        host = config.remotehost
        bmp_names = None
        auto_detect_bmp = config.auto_detect_bmp
    else:
        if version is None:
            if host == "192.168.240.253":
                version = 3
            else:
                version = 5
        auto_detect_bmp = False
        set_config("Machine", "version", version)

    print(f"talking to SpiNNaker system at {host}")
    return create_transceiver_from_hostname(
        host, bmp_connection_data=bmp_names,
        auto_detect_bmp=auto_detect_bmp)


def main(args):
    """
    Runs the script.
    """
    unittest_setup()
    ap = argparse.ArgumentParser(
        description="Check the state of a SpiNNaker machine.")
    ap.add_argument(
        "-a", "--appid", help="the application ID to check", type=int,
        default=17)
    ap.add_argument(
        "-v", "--version", help="the version of your boards", type=int,
        default=None)
    ap.add_argument(
        "-b", "--bmp_names",
        help="the hostname or IP address of the BMP of the SpiNNaker machine "
             "to inspects",
        type=str, default=None)
    ap.add_argument(
        "-n", "--noprintchips", action="store_true", default=False,
        help=("don't print all the chips out; avoids a great deal of "
              "output for large machines"))
    ap.add_argument(
        "host", default=None, nargs='?',
        help="the hostname or IP address of the SpiNNaker machine to inspect")
    args = ap.parse_args(args)
    # These ought to be parsed from command line arguments
    app_id = args.appid
    print_chips = not args.noprintchips

    transceiver = _make_transceiver(args.host, args.version, args.bmp_names)
    try:
        get_cores_in_run_state(transceiver, app_id, print_chips)
    finally:
        transceiver.close()


if __name__ == "__main__":  # pragma: no cover
    main(sys.argv[1:])

1	# Copyright (c) 2017 The University of Manchester
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# https://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15	"""	1✔
16	This is a script used to check the state of a SpiNNaker machine.
17	"""
18
19	import sys	1✔
20	import argparse	1✔
21	from spinn_utilities.config_holder import set_config	1✔
22	from spinnman.transceiver import create_transceiver_from_hostname	1✔
23	from spinn_machine import CoreSubsets, CoreSubset	1✔
24	from spinnman.board_test_configuration import BoardTestConfiguration	1✔
25	from spinnman.config_setup import unittest_setup	1✔
26	from spinnman.model.enums import CPUState	1✔
27	SCAMP_ID = 0	1✔
28	IGNORED_IDS = {SCAMP_ID, 16} # WHY 16?	1✔
29
30
31	def get_cores_in_run_state(txrx, app_id, print_all_chips):	1✔
32	"""
33	:param Transceiver txrx:
34	:param int app_id:
35	:param bool print_all_chips:
36	"""
37	count_finished = txrx.get_core_state_count(app_id, CPUState.FINISHED)	×
38	count_run = txrx.get_core_state_count(app_id, CPUState.RUNNING)	×
39	print(f'running: {count_run} finished: {count_finished}')	×
40
41	machine = txrx.get_machine_details()	×
42	print(f'machine width: {machine.width} height: {machine.height}')	×
43	if print_all_chips:	×
44	print(f'machine chips: {list(machine.chips)}')	×
45
46	all_cores = []	×
47	for chip in machine.chips:	×
48	all_cores.append(CoreSubset(chip.x, chip.y, range(1, 17)))	×
49
50	all_cores = CoreSubsets(core_subsets=all_cores)	×
51
52	cpu_infos = txrx.get_cpu_infos(	×
53	all_cores,
54	[CPUState.FINISHED, CPUState.RUNNING, CPUState.WATCHDOG], True)
55	cores_finished = cpu_infos.infos_for_state(CPUState.FINISHED)	×
56	cores_running = cpu_infos.infos_for_state(CPUState.RUNNING)	×
57	cores_watchdog = cpu_infos.infos_for_state(CPUState.WATCHDOG)	×
58
59	for (x, y, p), _ in cores_running:	×
60	if p not in IGNORED_IDS:	×
61	print(f'run core: {x} {y} {p}')	×
62
63	for (x, y, p), _ in cores_finished:	×
64	print(f'finished core: {x} {y} {p}')	×
65
66	for (x, y, p), _ in cores_watchdog:	×
67	print(f'watchdog core: {x} {y} {p}')	×
68
69
70	def _make_transceiver(host, version, bmp_names):	1✔
71	"""
72	:param host:
73	Host to use or `None` to use test configuration for all parameters
74	:type host: str or None
75	:param version: Board version to use (`None` defaults to 5 unless host is
76	192.168.240.253 (spin 3)
77	:type version: int or None
78	:param bmp: bmp connection or `None` to auto detect (if applicable)
79	:type bmp: str or None
80	:rtype: Transceiver
81	"""
82	if host is None:	×
83	config = BoardTestConfiguration()	×
84	config.set_up_remote_board()	×
85	host = config.remotehost	×
86	bmp_names = None	×
87	auto_detect_bmp = config.auto_detect_bmp	×
88	else:
89	if version is None:	×
90	if host == "192.168.240.253":	×
91	version = 3	×
92	else:
93	version = 5	×
94	auto_detect_bmp = False	×
95	set_config("Machine", "version", version)	×
96
97	print(f"talking to SpiNNaker system at {host}")	×
98	return create_transceiver_from_hostname(	×
99	host, bmp_connection_data=bmp_names,
100	auto_detect_bmp=auto_detect_bmp)
101
102
103	def main(args):	1✔
104	"""
105	Runs the script.
106	"""
107	unittest_setup()	×
108	ap = argparse.ArgumentParser(	×
109	description="Check the state of a SpiNNaker machine.")
110	ap.add_argument(	×
111	"-a", "--appid", help="the application ID to check", type=int,
112	default=17)
113	ap.add_argument(	×
114	"-v", "--version", help="the version of your boards", type=int,
115	default=None)
116	ap.add_argument(	×
117	"-b", "--bmp_names",
118	help="the hostname or IP address of the BMP of the SpiNNaker machine "
119	"to inspects",
120	type=str, default=None)
121	ap.add_argument(	×
122	"-n", "--noprintchips", action="store_true", default=False,
123	help=("don't print all the chips out; avoids a great deal of "
124	"output for large machines"))
125	ap.add_argument(	×
126	"host", default=None, nargs='?',
127	help="the hostname or IP address of the SpiNNaker machine to inspect")
128	args = ap.parse_args(args)	×
129	# These ought to be parsed from command line arguments
130	app_id = args.appid	×
131	print_chips = not args.noprintchips	×
132
133	transceiver = _make_transceiver(args.host, args.version, args.bmp_names)	×
134	try:	×
135	get_cores_in_run_state(transceiver, app_id, print_chips)	×
136	finally:
137	transceiver.close()	×
138
139
140	if __name__ == "__main__": # pragma: no cover	1!
141	main(sys.argv[1:])	×

SpiNNakerManchester / SpiNNMan / 6574804013

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous