8322459436

Committed 18 Mar 2024 06:30AM UTC coverage: 48.033% (+4.0%) from 43.995%

Build # 8322459436

Build Type

push

github

Committed by

Christian-B

Commit Message

merged in master

Run Details

2048 of 4791 branches covered (42.75%)

Branch coverage included in aggregate %.

2 of 4 new or added lines in 4 files covered. (50.0%)

2307 existing lines in 55 files now uncovered.

5497 of 10917 relevant lines covered (50.35%)

0.5 hits per line

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

89.16

/spinn_front_end_common/interface/profiling/profile_data.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.

import logging
import math
from typing import Dict, Iterable, Mapping, Tuple

import numpy
import scipy.stats  # type: ignore[import]

from spinn_utilities.log import FormatAdapter
from spinn_front_end_common.data import FecDataView

logger = FormatAdapter(logging.getLogger(__name__))

# Define profiler time scale in ms
_MS_SCALE = (1.0 / 200000.0)

# Define the positions in tags
_START_TIME = 0
_DURATION = 1


class ProfileData(object):
    """
    A container for profile data.
    """

    START_TIME = _START_TIME
    DURATION = _DURATION

    __slots__ = (
        # A dictionary of tag label to numpy array of start times and durations
        "_tags",

        # A list of tag labels indexed by the tag ID
        "_tag_labels",

        # The maximum time recorded
        "_max_time")

    def __init__(self, tag_labels: Mapping[int, str]):
        """
        :param list(str) tag_labels: A list of labels indexed by tag ID
        """
        self._tag_labels = tag_labels
        self._tags: Dict[str, Tuple[numpy.ndarray, numpy.ndarray]] = dict()
        self._max_time: float = 0.0

    def add_data(self, data: bytes):
        """
        Add profiling data read from the profile section.

        :param bytearray data:
            Data read from the profile section on the machine
        """
        samples = numpy.asarray(data, dtype="uint8").view(dtype="<u4")

        # Slice data to separate times, tags and flags
        sample_times = samples[::2]
        sample_tags_and_flags = samples[1::2]

        # Further split the tags and flags word into separate arrays of tags
        # and flags
        sample_tags = numpy.bitwise_and(sample_tags_and_flags, 0x7FFFFFFF)
        sample_flags = numpy.right_shift(sample_tags_and_flags, 31)

        # Find indices of samples relating to entries and exits
        sample_entry_indices = numpy.where(sample_flags == 1)
        sample_exit_indices = numpy.where(sample_flags == 0)

        # Convert count-down times to count up times from 1st sample
        sample_times_ms = numpy.multiply(
            numpy.subtract(sample_times[0], sample_times),
            _MS_SCALE, dtype=float)

        # Slice tags and times into entry and exits
        entry_tags = sample_tags[sample_entry_indices]
        entry_times_ms = sample_times_ms[sample_entry_indices]
        exit_tags = sample_tags[sample_exit_indices]
        exit_times_ms = sample_times_ms[sample_exit_indices]

        # Loop through unique tags
        for tag in numpy.unique(sample_tags):
            self._add_tag_data(
                entry_tags, entry_times_ms, exit_tags, exit_times_ms, tag)

    def _add_tag_data(
            self, entry_tags: numpy.ndarray, entry_times: numpy.ndarray,
            exit_tags: numpy.ndarray, exit_times: numpy.ndarray, tag: int):
        """
        :param ~numpy.ndarray entry_tags:
        :param ~numpy.ndarray entry_times:
        :param ~numpy.ndarray exit_tags:
        :param ~numpy.ndarray exit_times:
        :param int tag:
        """
        # pylint: disable=too-many-arguments
        tag_label = self._tag_labels.get(tag, None)
        if tag_label is None:
            logger.warning("Unknown tag {} in profile data", tag)
            tag_label = "UNKNOWN"

        # Get indices where these tags occur
        tag_entry_indices = numpy.where(entry_tags == tag)
        tag_exit_indices = numpy.where(exit_tags == tag)

        # Use these to get subset for this tag
        tag_entry_times = entry_times[tag_entry_indices]
        tag_exit_times = exit_times[tag_exit_indices]

        # If the first exit is before the first
        # Entry, add a dummy entry at beginning
        if tag_exit_times[0] < tag_entry_times[0]:
            logger.warning("Profile starts mid-tag")
            tag_entry_times = numpy.append(0.0, tag_entry_times)

        if len(tag_entry_times) > len(tag_exit_times):
            logger.warning("profile finishes mid-tag")
            tag_entry_times = tag_entry_times[
                :len(tag_exit_times) - len(tag_entry_times)]

        # Subtract entry times from exit times to get durations of each
        # call in ms
        tag_durations = numpy.subtract(tag_exit_times, tag_entry_times)

        # Add entry times and durations to dictionary
        self._tags[tag_label] = (tag_entry_times, tag_durations)

        # Keep track of the maximum time
        self._max_time = numpy.max(tag_entry_times + tag_durations)

    @property
    def tags(self) -> Iterable[str]:
        """
        The tags recorded as labels.

        :rtype: list(str)
        """
        return self._tags.keys()

    def get_mean_ms(self, tag: str) -> float:
        """
        Get the mean time in milliseconds spent on operations with the
        given tag.

        :param str tag: The tag to get the mean time for
        :rtype: float
        """
        return numpy.average(self._tags[tag][_DURATION])

    def get_n_calls(self, tag: str) -> int:
        """
        Get the number of times the given tag was recorded.

        :param str tag: The tag to get the number of calls of
        :rtype: int
        """
        return self._tags[tag][_DURATION].size

    def get_mean_n_calls_per_ts(self, tag: str) -> float:
        """
        Get the mean number of times the given tag was recorded per timestep.

        :param str tag: The tag to get the data for
        :rtype: float
        """
        time_step_ms = FecDataView.get_simulation_time_step_ms()
        n_points = math.ceil(self._max_time / time_step_ms)
        endpoint = n_points * time_step_ms
        bins = numpy.linspace(0, endpoint, n_points + 1)
        return numpy.average(numpy.histogram(
            self._tags[tag][_START_TIME], bins)[0])

    def get_mean_ms_per_ts(self, tag: str) -> float:
        """
        Get the mean time in milliseconds spent on operations with the
        given tag per timestep.

        :param str tag: The tag to get the data for
        :rtype: float
        """
        time_step_ms = FecDataView.get_simulation_time_step_ms()
        n_points = math.ceil(self._max_time / time_step_ms)
        endpoint = n_points * time_step_ms
        bins = numpy.linspace(0, endpoint, n_points + 1)
        mean_per_ts = scipy.stats.binned_statistic(
            self._tags[tag][_START_TIME], self._tags[tag][_DURATION],
            "mean", bins).statistic
        mean_per_ts[numpy.isnan(mean_per_ts)] = 0
        return numpy.average(
            mean_per_ts[numpy.logical_not(numpy.isnan(mean_per_ts))])

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	import logging	1✔
16	import math	1✔
17	from typing import Dict, Iterable, Mapping, Tuple	1✔
18
19	import numpy	1✔
20	import scipy.stats # type: ignore[import]	1✔
21
22	from spinn_utilities.log import FormatAdapter	1✔
23	from spinn_front_end_common.data import FecDataView	1✔
24
25	logger = FormatAdapter(logging.getLogger(__name__))	1✔
26
27	# Define profiler time scale in ms
28	_MS_SCALE = (1.0 / 200000.0)	1✔
29
30	# Define the positions in tags
31	_START_TIME = 0	1✔
32	_DURATION = 1	1✔
33
34
35	class ProfileData(object):	1✔
36	"""
37	A container for profile data.
38	"""
39
40	START_TIME = _START_TIME	1✔
41	DURATION = _DURATION	1✔
42
43	__slots__ = (	1✔
44	# A dictionary of tag label to numpy array of start times and durations
45	"_tags",
46
47	# A list of tag labels indexed by the tag ID
48	"_tag_labels",
49
50	# The maximum time recorded
51	"_max_time")
52
53	def __init__(self, tag_labels: Mapping[int, str]):	1✔
54	"""
55	:param list(str) tag_labels: A list of labels indexed by tag ID
56	"""
57	self._tag_labels = tag_labels	1✔
58	self._tags: Dict[str, Tuple[numpy.ndarray, numpy.ndarray]] = dict()	1✔
59	self._max_time: float = 0.0	1✔
60
61	def add_data(self, data: bytes):	1✔
62	"""
63	Add profiling data read from the profile section.
64
65	:param bytearray data:
66	Data read from the profile section on the machine
67	"""
68	samples = numpy.asarray(data, dtype="uint8").view(dtype="<u4")	1✔
69
70	# Slice data to separate times, tags and flags
71	sample_times = samples[::2]	1✔
72	sample_tags_and_flags = samples[1::2]	1✔
73
74	# Further split the tags and flags word into separate arrays of tags
75	# and flags
76	sample_tags = numpy.bitwise_and(sample_tags_and_flags, 0x7FFFFFFF)	1✔
77	sample_flags = numpy.right_shift(sample_tags_and_flags, 31)	1✔
78
79	# Find indices of samples relating to entries and exits
80	sample_entry_indices = numpy.where(sample_flags == 1)	1✔
81	sample_exit_indices = numpy.where(sample_flags == 0)	1✔
82
83	# Convert count-down times to count up times from 1st sample
84	sample_times_ms = numpy.multiply(	1✔
85	numpy.subtract(sample_times[0], sample_times),
86	_MS_SCALE, dtype=float)
87
88	# Slice tags and times into entry and exits
89	entry_tags = sample_tags[sample_entry_indices]	1✔
90	entry_times_ms = sample_times_ms[sample_entry_indices]	1✔
91	exit_tags = sample_tags[sample_exit_indices]	1✔
92	exit_times_ms = sample_times_ms[sample_exit_indices]	1✔
93
94	# Loop through unique tags
95	for tag in numpy.unique(sample_tags):	1✔
96	self._add_tag_data(	1✔
97	entry_tags, entry_times_ms, exit_tags, exit_times_ms, tag)
98
99	def _add_tag_data(	1✔
100	self, entry_tags: numpy.ndarray, entry_times: numpy.ndarray,
101	exit_tags: numpy.ndarray, exit_times: numpy.ndarray, tag: int):
102	"""
103	:param ~numpy.ndarray entry_tags:
104	:param ~numpy.ndarray entry_times:
105	:param ~numpy.ndarray exit_tags:
106	:param ~numpy.ndarray exit_times:
107	:param int tag:
108	"""
109	# pylint: disable=too-many-arguments
110	tag_label = self._tag_labels.get(tag, None)	1✔
111	if tag_label is None:	1!
UNCOV 112	logger.warning("Unknown tag {} in profile data", tag)	×
UNCOV 113	tag_label = "UNKNOWN"	×
114
115	# Get indices where these tags occur
116	tag_entry_indices = numpy.where(entry_tags == tag)	1✔
117	tag_exit_indices = numpy.where(exit_tags == tag)	1✔
118
119	# Use these to get subset for this tag
120	tag_entry_times = entry_times[tag_entry_indices]	1✔
121	tag_exit_times = exit_times[tag_exit_indices]	1✔
122
123	# If the first exit is before the first
124	# Entry, add a dummy entry at beginning
125	if tag_exit_times[0] < tag_entry_times[0]:	1!
UNCOV 126	logger.warning("Profile starts mid-tag")	×
UNCOV 127	tag_entry_times = numpy.append(0.0, tag_entry_times)	×
128
129	if len(tag_entry_times) > len(tag_exit_times):	1!
UNCOV 130	logger.warning("profile finishes mid-tag")	×
UNCOV 131	tag_entry_times = tag_entry_times[	×
132	:len(tag_exit_times) - len(tag_entry_times)]
133
134	# Subtract entry times from exit times to get durations of each
135	# call in ms
136	tag_durations = numpy.subtract(tag_exit_times, tag_entry_times)	1✔
137
138	# Add entry times and durations to dictionary
139	self._tags[tag_label] = (tag_entry_times, tag_durations)	1✔
140
141	# Keep track of the maximum time
142	self._max_time = numpy.max(tag_entry_times + tag_durations)	1✔
143
144	@property	1✔
145	def tags(self) -> Iterable[str]:	1✔
146	"""
147	The tags recorded as labels.
148
149	:rtype: list(str)
150	"""
151	return self._tags.keys()	1✔
152
153	def get_mean_ms(self, tag: str) -> float:	1✔
154	"""
155	Get the mean time in milliseconds spent on operations with the
156	given tag.
157
158	:param str tag: The tag to get the mean time for
159	:rtype: float
160	"""
161	return numpy.average(self._tags[tag][_DURATION])	1✔
162
163	def get_n_calls(self, tag: str) -> int:	1✔
164	"""
165	Get the number of times the given tag was recorded.
166
167	:param str tag: The tag to get the number of calls of
168	:rtype: int
169	"""
170	return self._tags[tag][_DURATION].size	1✔
171
172	def get_mean_n_calls_per_ts(self, tag: str) -> float:	1✔
173	"""
174	Get the mean number of times the given tag was recorded per timestep.
175
176	:param str tag: The tag to get the data for
177	:rtype: float
178	"""
179	time_step_ms = FecDataView.get_simulation_time_step_ms()	1✔
180	n_points = math.ceil(self._max_time / time_step_ms)	1✔
181	endpoint = n_points * time_step_ms	1✔
182	bins = numpy.linspace(0, endpoint, n_points + 1)	1✔
183	return numpy.average(numpy.histogram(	1✔
184	self._tags[tag][_START_TIME], bins)[0])
185
186	def get_mean_ms_per_ts(self, tag: str) -> float:	1✔
187	"""
188	Get the mean time in milliseconds spent on operations with the
189	given tag per timestep.
190
191	:param str tag: The tag to get the data for
192	:rtype: float
193	"""
194	time_step_ms = FecDataView.get_simulation_time_step_ms()	1✔
195	n_points = math.ceil(self._max_time / time_step_ms)	1✔
196	endpoint = n_points * time_step_ms	1✔
197	bins = numpy.linspace(0, endpoint, n_points + 1)	1✔
198	mean_per_ts = scipy.stats.binned_statistic(	1✔
199	self._tags[tag][_START_TIME], self._tags[tag][_DURATION],
200	"mean", bins).statistic
201	mean_per_ts[numpy.isnan(mean_per_ts)] = 0	1✔
202	return numpy.average(	1✔
203	mean_per_ts[numpy.logical_not(numpy.isnan(mean_per_ts))])

SpiNNakerManchester / SpiNNFrontEndCommon / 8322459436

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