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

28.05

/spinnman/model/executable_targets.py

# Copyright (c) 2016 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 collections import defaultdict
from typing import Collection, Dict, Iterable, Optional, Set, cast
from spinn_utilities.ordered_set import OrderedSet
from spinn_machine import CoreSubsets, FrozenCoreSubsets
from spinnman.exceptions import SpinnmanInvalidParameterException
from .enums import ExecutableType


class ExecutableTargets(object):
    """
    Encapsulate the binaries and cores on which to execute them.
    """
    __slots__ = [
        "_all_core_subsets",
        "_targets",
        "_total_processors",
        "_binary_type_map"]

    __EMPTY_SUBSET = FrozenCoreSubsets()

    def __init__(self) -> None:
        self._targets: Dict[str, CoreSubsets] = dict()
        self._total_processors = 0
        self._all_core_subsets = CoreSubsets()
        self._binary_type_map: Dict[
            ExecutableType, Set[str]] = defaultdict(
                # Need to pretend!
                lambda: cast(Set, OrderedSet()))

    def add_subsets(
            self, binary: str, subsets: CoreSubsets,
            executable_type: Optional[ExecutableType] = None):
        """
        Add core subsets to a binary.

        :param str binary: the path to the binary needed to be executed
        :param ~spinn_machine.CoreSubsets subsets:
            the subset of cores that the binary needs to be loaded on
        :param ~spinnman.model.enum.ExecutableType executable_type:
            The type of this executable.
            ``None`` means don't record it.
        """
        try:
            for subset in subsets.core_subsets:
                for p in subset.processor_ids:
                    self.add_processor(binary, subset.x, subset.y, p)
        except AttributeError:
            if subsets is not None:
                raise
        if executable_type is not None:
            self._binary_type_map[executable_type].add(binary)

    def add_processor(
            self, binary: str, chip_x: int, chip_y: int, chip_p: int,
            executable_type: Optional[ExecutableType] = None):
        """
        Add a processor to the executable targets

        :param str binary: the binary path for executable
        :param int chip_x:
            the coordinate on the machine in terms of x for the chip
        :param int chip_y:
            the coordinate on the machine in terms of y for the chip
        :param int chip_p: the processor ID to place this executable on
        :param ~spinnman.model.enum.ExecutableType executable_type:
            the executable type for locating n cores of
        """
        if self.known(binary, chip_x, chip_y, chip_p):
            return
        if binary not in self._targets:
            self._targets[binary] = CoreSubsets()
        if executable_type is not None:
            self._binary_type_map[executable_type].add(binary)
        self._targets[binary].add_processor(chip_x, chip_y, chip_p)
        self._all_core_subsets.add_processor(chip_x, chip_y, chip_p)
        self._total_processors += 1

    def get_n_cores_for_executable_type(
            self, executable_type: ExecutableType) -> int:
        """
        Get the number of cores that the executable type is using.

        :param ~spinnman.model.enum.ExecutableType executable_type:
        :return: the number of cores using this executable type
        :rtype: int
        """
        return sum(
            len(self.get_cores_for_binary(aplx))
            for aplx in self._binary_type_map[executable_type])

    def get_binaries_of_executable_type(
            self, executable_type: ExecutableType) -> Iterable[str]:
        """
        Get the binaries of a given a executable type.

        :param ~spinnman.model.enum.ExecutableType executable_type:
            the executable type enum value
        :return: iterable of binaries with that executable type
        :rtype: iterable(str)
        """
        return self._binary_type_map[executable_type]

    def executable_types_in_binary_set(self) -> Iterable[ExecutableType]:
        """
        Get the executable types in the set of binaries.

        :return: iterable of the executable types in this binary set.
        :rtype:
            iterable(~spinnman.model.enum.ExecutableType)
        """
        return self._binary_type_map.keys()

    def get_cores_for_binary(self, binary: str) -> CoreSubsets:
        """
        Get the cores that a binary is to run on.

        :param str binary: The binary to find the cores for
        :rtype: ~spinn_machine.CoreSubsets
        """
        return self._targets.get(binary, self.__EMPTY_SUBSET)

    @property
    def binaries(self) -> Collection[str]:
        """
        The binaries of the executables.

        :rtype: iterable(str)
        """
        return self._targets.keys()

    @property
    def total_processors(self) -> int:
        """
        The total number of cores to be loaded.

        :rtype: int
        """
        return self._total_processors

    @property
    def all_core_subsets(self) -> CoreSubsets:
        """
        All the core subsets for all the binaries.

        :rtype: ~spinn_machine.CoreSubsets
        """
        return self._all_core_subsets

    def known(self, binary, chip_x, chip_y, chip_p) -> bool:
        """
        :param str binary:
        :param int chip_x:
        :param int chip_y:
        :param int chip_p:
        :rtype: bool
        """
        if not self._all_core_subsets.is_core(chip_x, chip_y, chip_p):
            return False
        # OK if and only if the chip is in this binary already
        if binary in self._targets:
            if self._targets[binary].is_core(chip_x, chip_y, chip_p):
                return True

        raise SpinnmanInvalidParameterException(
            f"x:{chip_x} y:{chip_y} p:{chip_p}", binary,
            "Already associated with a different binary")

1	# Copyright (c) 2016 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	from collections import defaultdict	1✔
16	from typing import Collection, Dict, Iterable, Optional, Set, cast	1✔
17	from spinn_utilities.ordered_set import OrderedSet	1✔
18	from spinn_machine import CoreSubsets, FrozenCoreSubsets	1✔
19	from spinnman.exceptions import SpinnmanInvalidParameterException	1✔
20	from .enums import ExecutableType	1✔
21
22
23	class ExecutableTargets(object):	1✔
24	"""
25	Encapsulate the binaries and cores on which to execute them.
26	"""
27	__slots__ = [	1✔
28	"_all_core_subsets",
29	"_targets",
30	"_total_processors",
31	"_binary_type_map"]
32
33	__EMPTY_SUBSET = FrozenCoreSubsets()	1✔
34
35	def __init__(self) -> None:	1✔
36	self._targets: Dict[str, CoreSubsets] = dict()	×
37	self._total_processors = 0	×
38	self._all_core_subsets = CoreSubsets()	×
39	self._binary_type_map: Dict[	×
40	ExecutableType, Set[str]] = defaultdict(
41	# Need to pretend!
42	lambda: cast(Set, OrderedSet()))
43
44	def add_subsets(	1✔
45	self, binary: str, subsets: CoreSubsets,
46	executable_type: Optional[ExecutableType] = None):
47	"""
48	Add core subsets to a binary.
49
50	:param str binary: the path to the binary needed to be executed
51	:param ~spinn_machine.CoreSubsets subsets:
52	the subset of cores that the binary needs to be loaded on
53	:param ~spinnman.model.enum.ExecutableType executable_type:
54	The type of this executable.
55	``None`` means don't record it.
56	"""
57	try:	×
58	for subset in subsets.core_subsets:	×
59	for p in subset.processor_ids:	×
60	self.add_processor(binary, subset.x, subset.y, p)	×
61	except AttributeError:	×
62	if subsets is not None:	×
63	raise	×
64	if executable_type is not None:	×
65	self._binary_type_map[executable_type].add(binary)	×
66
67	def add_processor(	1✔
68	self, binary: str, chip_x: int, chip_y: int, chip_p: int,
69	executable_type: Optional[ExecutableType] = None):
70	"""
71	Add a processor to the executable targets
72
73	:param str binary: the binary path for executable
74	:param int chip_x:
75	the coordinate on the machine in terms of x for the chip
76	:param int chip_y:
77	the coordinate on the machine in terms of y for the chip
78	:param int chip_p: the processor ID to place this executable on
79	:param ~spinnman.model.enum.ExecutableType executable_type:
80	the executable type for locating n cores of
81	"""
82	if self.known(binary, chip_x, chip_y, chip_p):	×
83	return	×
84	if binary not in self._targets:	×
85	self._targets[binary] = CoreSubsets()	×
86	if executable_type is not None:	×
87	self._binary_type_map[executable_type].add(binary)	×
88	self._targets[binary].add_processor(chip_x, chip_y, chip_p)	×
89	self._all_core_subsets.add_processor(chip_x, chip_y, chip_p)	×
90	self._total_processors += 1	×
91
92	def get_n_cores_for_executable_type(	1✔
93	self, executable_type: ExecutableType) -> int:
94	"""
95	Get the number of cores that the executable type is using.
96
97	:param ~spinnman.model.enum.ExecutableType executable_type:
98	:return: the number of cores using this executable type
99	:rtype: int
100	"""
101	return sum(	×
102	len(self.get_cores_for_binary(aplx))
103	for aplx in self._binary_type_map[executable_type])
104
105	def get_binaries_of_executable_type(	1✔
106	self, executable_type: ExecutableType) -> Iterable[str]:
107	"""
108	Get the binaries of a given a executable type.
109
110	:param ~spinnman.model.enum.ExecutableType executable_type:
111	the executable type enum value
112	:return: iterable of binaries with that executable type
113	:rtype: iterable(str)
114	"""
115	return self._binary_type_map[executable_type]	×
116
117	def executable_types_in_binary_set(self) -> Iterable[ExecutableType]:	1✔
118	"""
119	Get the executable types in the set of binaries.
120
121	:return: iterable of the executable types in this binary set.
122	:rtype:
123	iterable(~spinnman.model.enum.ExecutableType)
124	"""
125	return self._binary_type_map.keys()	×
126
127	def get_cores_for_binary(self, binary: str) -> CoreSubsets:	1✔
128	"""
129	Get the cores that a binary is to run on.
130
131	:param str binary: The binary to find the cores for
132	:rtype: ~spinn_machine.CoreSubsets
133	"""
134	return self._targets.get(binary, self.__EMPTY_SUBSET)	×
135
136	@property	1✔
137	def binaries(self) -> Collection[str]:	1✔
138	"""
139	The binaries of the executables.
140
141	:rtype: iterable(str)
142	"""
143	return self._targets.keys()	×
144
145	@property	1✔
146	def total_processors(self) -> int:	1✔
147	"""
148	The total number of cores to be loaded.
149
150	:rtype: int
151	"""
152	return self._total_processors	×
153
154	@property	1✔
155	def all_core_subsets(self) -> CoreSubsets:	1✔
156	"""
157	All the core subsets for all the binaries.
158
159	:rtype: ~spinn_machine.CoreSubsets
160	"""
161	return self._all_core_subsets	×
162
163	def known(self, binary, chip_x, chip_y, chip_p) -> bool:	1✔
164	"""
165	:param str binary:
166	:param int chip_x:
167	:param int chip_y:
168	:param int chip_p:
169	:rtype: bool
170	"""
171	if not self._all_core_subsets.is_core(chip_x, chip_y, chip_p):	×
172	return False	×
173	# OK if and only if the chip is in this binary already
174	if binary in self._targets:	×
175	if self._targets[binary].is_core(chip_x, chip_y, chip_p):	×
176	return True	×
177
178	raise SpinnmanInvalidParameterException(	×
179	f"x:{chip_x} y:{chip_y} p:{chip_p}", binary,
180	"Already associated with a different binary")

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