7285532539

Committed 21 Dec 2023 08:04AM UTC coverage: 62.866% (+0.09%) from 62.778%

Build # 7285532539

Build Type

Pull #1425

github

Committed by

Christian-B

Commit Message

flake8

Pull Request Pull Request #1425: Overrides check

Run Details

1909 of 4475 branches covered (0.0%)

Branch coverage included in aggregate %.

317 of 328 new or added lines in 66 files covered. (96.65%)

26 existing lines in 17 files now uncovered.

12789 of 18905 relevant lines covered (67.65%)

0.68 hits per line

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

62.96

/spynnaker/pyNN/utilities/data_population.py

# Copyright (c) 2022 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 __future__ import annotations
import logging
import numpy
from numpy import floating
from numpy.typing import NDArray
from pyNN.descriptions import TemplateEngine
from typing import (
    Any, Dict, Iterable, Optional, overload, Sequence, Union, TYPE_CHECKING)
import neo  # type: ignore[import]
from spinn_utilities.ranged.abstract_sized import AbstractSized, Selector
from spinn_utilities.log import FormatAdapter
from spinn_utilities.overrides import overrides
from spynnaker.pyNN.models.populations import Population
from spynnaker.pyNN.utilities.neo_buffer_database import NeoBufferDatabase
from spynnaker.pyNN.utilities.utility_calls import get_neo_io
from spynnaker.pyNN.models.common.types import Names
if TYPE_CHECKING:
    from .neo_buffer_database import Annotations

logger = FormatAdapter(logging.getLogger(__file__))
_SELECTIVE_RECORDED_MSG = (
    "Getting data on a whole population when selective recording was "
    "active will result in only the recorded neurons being returned "
    "in numerical order and without repeats.")


class DataPopulation(object):
    __slots__ = (
        "__database_file",
        "__label",
        "_indexes",
        "_size")

    def __init__(self, database_file: str, label: str, indexes=None):
        self.__label = label
        self.__database_file = database_file
        # getting size right away also check the inputs or fails fast
        with NeoBufferDatabase(self.__database_file) as db:
            size = db.get_population_metadata(label)[0]
        self._size = size
        self._indexes = indexes

    @overrides(Population.write_data)
    def write_data(self, io: Union[str, neo.baseio.BaseIO],
                   variables: Names = 'all', gather: bool = True,
                   clear: bool = False, annotations: Annotations = None):
        # pylint: disable=protected-access
        Population._check_params(gather, annotations)
        if clear:
            logger.warning("Ignoring clear as supported in this mode")
        if isinstance(io, str):
            io = get_neo_io(io)
        data = self.get_data(variables)
        # write the neo block to the file
        io.write(bl=data)

    @overrides(Population.describe, adds_typing=True)
    def describe(self, template: Optional[str] = None,
                 engine: Optional[Union[str, TemplateEngine]] = None
                 ) -> Union[str, Dict[str, Any]]:
        if template is not None:
            logger.warning("Ignoring template as not supported in this mode")
        if engine is not None:
            logger.warning("Ignoring engine as not supported in this mode")
        with NeoBufferDatabase(self.__database_file) as db:
            _, _, description = db.get_population_metadata(self.label)
            return description

    @overrides(Population.get_data)
    def get_data(
            self, variables: Names = 'all',
            gather: bool = True, clear: bool = False, *,
            annotations: Optional[Dict[str, Any]] = None) -> neo.Block:
        # pylint: disable=protected-access
        Population._check_params(gather, annotations)
        if clear:
            logger.warning("Ignoring clear as supported in this mode")
        with NeoBufferDatabase(self.__database_file) as db:
            return db.get_full_block(
                self.__label, variables, self._indexes, annotations)

    @overrides(Population.spinnaker_get_data)
    def spinnaker_get_data(
            self, variable: str, as_matrix: bool = False,
            view_indexes: Optional[Sequence[int]] = None) -> NDArray[floating]:
        if view_indexes:
            return self[view_indexes].spinnaker_get_data(variable, as_matrix)
        with NeoBufferDatabase(self.__database_file) as db:
            return db.spinnaker_get_data(
                self.__label, variable, as_matrix, self._indexes)

    @overrides(Population.get_spike_counts)
    def get_spike_counts(self, gather: bool = True) -> Dict[int, int]:
        # pylint: disable=protected-access
        Population._check_params(gather)
        with NeoBufferDatabase(self.__database_file) as db:
            return db.get_spike_counts(self.__label, self._indexes)

    @overrides(Population.find_units)
    def find_units(self, variable: str) -> Optional[str]:
        with NeoBufferDatabase(self.__database_file) as db:
            (_, _, units) = db.get_recording_metadata(self.__label, variable)
        return units

    def __len__(self) -> int:
        return self._size

    @property
    def label(self) -> str:
        return self.__label

    @property
    def local_size(self) -> int:
        return self._size

    @property
    def size(self) -> int:
        return self._size

    @overload
    def id_to_index(self, id: int) -> int:  # @ReservedAssignment
        # pylint: disable=redefined-builtin
        ...

    @overload
    def id_to_index(
            self, id: Iterable[int]) -> Sequence[int]:  # @ReservedAssignment
        # pylint: disable=redefined-builtin
        ...

    @overrides(Population.id_to_index)
    def id_to_index(self, id: Union[int, Iterable[int]]
                    ) -> Union[int, Sequence[int]]:  # @ReservedAssignment
        # pylint: disable=redefined-builtin
        # assuming not called often so not caching first id
        with NeoBufferDatabase(self.__database_file) as db:
            _, first_id, _ = db.get_population_metadata(self.__label)
        last_id = self._size + first_id
        if not numpy.iterable(id):
            if not first_id <= id <= last_id:
                raise ValueError(
                    f"id should be in the range [{first_id},{last_id}], "
                    f"actually {id}")
            return int(id - first_id)  # assume IDs are consecutive
        return id - first_id

    @overload
    def index_to_id(self, index: int) -> int:
        ...

    @overload
    def index_to_id(self, index: Iterable[int]) -> Sequence[int]:
        ...

    @overrides(Population.index_to_id)
    def index_to_id(self, index: Union[int, Iterable[int]]
                    ) -> Union[int, Sequence[int]]:
        # assuming not called often so not caching first id
        with NeoBufferDatabase(self.__database_file) as db:
            _, first_id, _ = db.get_population_metadata(self.__label)
        if not numpy.iterable(index):
            if index >= self._size:
                raise ValueError(
                    f"indexes should be in the range [0,{self._size}],"
                    f" actually {index}")
            return int(index + first_id)
        # this assumes IDs are consecutive
        return index + first_id

    def __getitem__(self, index_or_slice: Selector) -> DataPopulation:
        """
        :param selector: a slice or numpy mask array.
            The mask array should either be a boolean array (ideally) of the
            same size as the parent,
            or an integer array containing cell indices,
            i.e. if `p.size == 5` then:

            ::

                PopulationView(p, array([False, False, True, False, True]))
                PopulationView(p, array([2, 4]))
                PopulationView(p, slice(2, 5, 2))

            will all create the same view.
        :type selector: None or slice or int or list(bool) or list(int) or
            ~numpy.ndarray(bool) or ~numpy.ndarray(int)
        :param index_or_slice:
        :return:
        """
        sized = AbstractSized(self._size)
        ids = sized.selector_to_ids(index_or_slice, warn=True)
        if self._indexes:
            indexes = [self._indexes[index] for index in ids]
        else:
            indexes = [range(self._size)[index] for index in ids]
        return DataPopulation(self.__database_file, self.__label, indexes)

    @overrides(Population.mean_spike_count)
    def mean_spike_count(self, gather: bool = True) -> float:
        Population._check_params(gather)  # pylint: disable=protected-access
        counts = self.get_spike_counts()
        return sum(counts.values()) / len(counts)

1	# Copyright (c) 2022 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	from __future__ import annotations	1✔
15	import logging	1✔
16	import numpy	1✔
17	from numpy import floating	1✔
18	from numpy.typing import NDArray	1✔
19	from pyNN.descriptions import TemplateEngine	1✔
20	from typing import (	1✔
21	Any, Dict, Iterable, Optional, overload, Sequence, Union, TYPE_CHECKING)
22	import neo # type: ignore[import]	1✔
23	from spinn_utilities.ranged.abstract_sized import AbstractSized, Selector	1✔
24	from spinn_utilities.log import FormatAdapter	1✔
25	from spinn_utilities.overrides import overrides	1✔
26	from spynnaker.pyNN.models.populations import Population	1✔
27	from spynnaker.pyNN.utilities.neo_buffer_database import NeoBufferDatabase	1✔
28	from spynnaker.pyNN.utilities.utility_calls import get_neo_io	1✔
29	from spynnaker.pyNN.models.common.types import Names	1✔
30	if TYPE_CHECKING:
31	from .neo_buffer_database import Annotations
32
33	logger = FormatAdapter(logging.getLogger(__file__))	1✔
34	_SELECTIVE_RECORDED_MSG = (	1✔
35	"Getting data on a whole population when selective recording was "
36	"active will result in only the recorded neurons being returned "
37	"in numerical order and without repeats.")
38
39
40	class DataPopulation(object):	1✔
41	__slots__ = (	1✔
42	"__database_file",
43	"__label",
44	"_indexes",
45	"_size")
46
47	def __init__(self, database_file: str, label: str, indexes=None):	1✔
48	self.__label = label	1✔
49	self.__database_file = database_file	1✔
50	# getting size right away also check the inputs or fails fast
51	with NeoBufferDatabase(self.__database_file) as db:	1✔
52	size = db.get_population_metadata(label)[0]	1✔
53	self._size = size	1✔
54	self._indexes = indexes	1✔
55
56	@overrides(Population.write_data)	1✔
57	def write_data(self, io: Union[str, neo.baseio.BaseIO],	1✔
58	variables: Names = 'all', gather: bool = True,
59	clear: bool = False, annotations: Annotations = None):
60	# pylint: disable=protected-access
61	Population._check_params(gather, annotations)	1✔
62	if clear:	1!
63	logger.warning("Ignoring clear as supported in this mode")	×
64	if isinstance(io, str):	1!
65	io = get_neo_io(io)	1✔
66	data = self.get_data(variables)	1✔
67	# write the neo block to the file
68	io.write(bl=data)	1✔
69
70	@overrides(Population.describe, adds_typing=True)	1✔
71	def describe(self, template: Optional[str] = None,	1✔
72	engine: Optional[Union[str, TemplateEngine]] = None
73	) -> Union[str, Dict[str, Any]]:
74	if template is not None:	×
NEW 75	logger.warning("Ignoring template as not supported in this mode")	×
76	if engine is not None:	×
NEW 77	logger.warning("Ignoring engine as not supported in this mode")	×
78	with NeoBufferDatabase(self.__database_file) as db:	×
79	_, _, description = db.get_population_metadata(self.label)	×
80	return description	×
81
82	@overrides(Population.get_data)	1✔
83	def get_data(	1✔
84	self, variables: Names = 'all',
85	gather: bool = True, clear: bool = False, *,
86	annotations: Optional[Dict[str, Any]] = None) -> neo.Block:
87	# pylint: disable=protected-access
88	Population._check_params(gather, annotations)	1✔
89	if clear:	1!
90	logger.warning("Ignoring clear as supported in this mode")	×
91	with NeoBufferDatabase(self.__database_file) as db:	1✔
92	return db.get_full_block(	1✔
93	self.__label, variables, self._indexes, annotations)
94
95	@overrides(Population.spinnaker_get_data)	1✔
96	def spinnaker_get_data(	1✔
97	self, variable: str, as_matrix: bool = False,
98	view_indexes: Optional[Sequence[int]] = None) -> NDArray[floating]:
99	if view_indexes:	1✔
100	return self[view_indexes].spinnaker_get_data(variable, as_matrix)	1✔
101	with NeoBufferDatabase(self.__database_file) as db:	1✔
102	return db.spinnaker_get_data(	1✔
103	self.__label, variable, as_matrix, self._indexes)
104
105	@overrides(Population.get_spike_counts)	1✔
106	def get_spike_counts(self, gather: bool = True) -> Dict[int, int]:	1✔
107	# pylint: disable=protected-access
108	Population._check_params(gather)	1✔
109	with NeoBufferDatabase(self.__database_file) as db:	1✔
110	return db.get_spike_counts(self.__label, self._indexes)	1✔
111
112	@overrides(Population.find_units)	1✔
113	def find_units(self, variable: str) -> Optional[str]:	1✔
114	with NeoBufferDatabase(self.__database_file) as db:	×
115	(_, _, units) = db.get_recording_metadata(self.__label, variable)	×
116	return units	×
117
118	def __len__(self) -> int:	1✔
119	return self._size	×
120
121	@property	1✔
122	def label(self) -> str:	1✔
123	return self.__label	×
124
125	@property	1✔
126	def local_size(self) -> int:	1✔
127	return self._size	×
128
129	@property	1✔
130	def size(self) -> int:	1✔
131	return self._size	×
132
133	@overload
134	def id_to_index(self, id: int) -> int: # @ReservedAssignment
135	# pylint: disable=redefined-builtin
136	...
137
138	@overload
139	def id_to_index(
140	self, id: Iterable[int]) -> Sequence[int]: # @ReservedAssignment
141	# pylint: disable=redefined-builtin
142	...
143
144	@overrides(Population.id_to_index)	1✔
145	def id_to_index(self, id: Union[int, Iterable[int]]	1✔
146	) -> Union[int, Sequence[int]]: # @ReservedAssignment
147	# pylint: disable=redefined-builtin
148	# assuming not called often so not caching first id
149	with NeoBufferDatabase(self.__database_file) as db:	×
150	_, first_id, _ = db.get_population_metadata(self.__label)	×
151	last_id = self._size + first_id	×
152	if not numpy.iterable(id):	×
153	if not first_id <= id <= last_id:	×
154	raise ValueError(	×
155	f"id should be in the range [{first_id},{last_id}], "
156	f"actually {id}")
157	return int(id - first_id) # assume IDs are consecutive	×
158	return id - first_id	×
159
160	@overload
161	def index_to_id(self, index: int) -> int:
162	...
163
164	@overload
165	def index_to_id(self, index: Iterable[int]) -> Sequence[int]:
166	...
167
168	@overrides(Population.index_to_id)	1✔
169	def index_to_id(self, index: Union[int, Iterable[int]]	1✔
170	) -> Union[int, Sequence[int]]:
171	# assuming not called often so not caching first id
172	with NeoBufferDatabase(self.__database_file) as db:	×
173	_, first_id, _ = db.get_population_metadata(self.__label)	×
174	if not numpy.iterable(index):	×
175	if index >= self._size:	×
176	raise ValueError(	×
177	f"indexes should be in the range [0,{self._size}],"
178	f" actually {index}")
179	return int(index + first_id)	×
180	# this assumes IDs are consecutive
181	return index + first_id	×
182
183	def __getitem__(self, index_or_slice: Selector) -> DataPopulation:	1✔
184	"""
185	:param selector: a slice or numpy mask array.
186	The mask array should either be a boolean array (ideally) of the
187	same size as the parent,
188	or an integer array containing cell indices,
189	i.e. if `p.size == 5` then:
190
191	::
192
193	PopulationView(p, array([False, False, True, False, True]))
194	PopulationView(p, array([2, 4]))
195	PopulationView(p, slice(2, 5, 2))
196
197	will all create the same view.
198	:type selector: None or slice or int or list(bool) or list(int) or
199	~numpy.ndarray(bool) or ~numpy.ndarray(int)
200	:param index_or_slice:
201	:return:
202	"""
203	sized = AbstractSized(self._size)	1✔
204	ids = sized.selector_to_ids(index_or_slice, warn=True)	1✔
205	if self._indexes:	1!
206	indexes = [self._indexes[index] for index in ids]	×
207	else:
208	indexes = [range(self._size)[index] for index in ids]	1✔
209	return DataPopulation(self.__database_file, self.__label, indexes)	1✔
210
211	@overrides(Population.mean_spike_count)	1✔
212	def mean_spike_count(self, gather: bool = True) -> float:	1✔
213	Population._check_params(gather) # pylint: disable=protected-access	1✔
214	counts = self.get_spike_counts()	1✔
215	return sum(counts.values()) / len(counts)	1✔

SpiNNakerManchester / sPyNNaker / 7285532539

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