6822522522

Committed 10 Nov 2023 08:32AM UTC coverage: 44.012% (+1.6%) from 42.408%

Build # 6822522522

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1141 from SpiNNakerManchester/roc_part

FrontEndCommon typing

Run Details

817 of 3284 branches covered (0.0%)

Branch coverage included in aggregate %.

2141 of 3218 new or added lines in 161 files covered. (66.53%)

157 existing lines in 53 files now uncovered.

5537 of 11153 relevant lines covered (49.65%)

0.5 hits per line

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

76.92

/spinn_front_end_common/interface/interface_functions/database_interface.py

# Copyright (c) 2015 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
from typing import Optional, Set, Tuple
from spinn_utilities.config_holder import (
    get_config_bool, get_config_bool_or_none)
from spinn_utilities.progress_bar import ProgressBar
from spinn_utilities.log import FormatAdapter
from pacman.model.graphs.machine import MachineVertex
from pacman.model.graphs.application import ApplicationVertex
from spinn_front_end_common.utilities.database import DatabaseWriter
from spinn_front_end_common.abstract_models import (
    AbstractSupportsDatabaseInjection)
from spinn_front_end_common.data import FecDataView

logger = FormatAdapter(logging.getLogger(__name__))


def database_interface(runtime: Optional[float]) -> Optional[str]:
    """
    :param int runtime:
    :return: where the database is located, if one is made
    :rtype: str or None
    """
    needs_db = DatabaseWriter.auto_detect_database()
    user_create_database = get_config_bool_or_none(
        "Database", "create_database")
    if user_create_database is not None:
        if user_create_database != needs_db:
            logger.warning(
                "Database creating changed to {} due to cfg settings",
                user_create_database)
            needs_db = user_create_database
    if not needs_db:
        return None

    with DatabaseWriter() as writer:
        logger.info("Creating live event connection database in {}",
                    writer.database_path)
        _write_to_db(writer, runtime)
        return writer.database_path


def _write_to_db(w: DatabaseWriter, runtime: Optional[float]):
    """
    :param DatabaseWriter w:
    :param int runtime:
    """
    with ProgressBar(6, "Creating graph description database") as p:
        w.add_system_params(runtime)
        w.add_proxy_configuration()
        p.update()
        w.add_machine_objects()
        p.update()
        w.add_application_vertices()
        p.update()
        w.add_placements()
        p.update()
        w.add_tags()
        p.update()
        lpg_source_machine_vertices = w.add_lpg_mapping()

        if get_config_bool(
                "Database", "create_routing_info_to_neuron_id_mapping"):
            machine_vertices: Set[Tuple[MachineVertex, str]] = {
                (vertex, vertex.injection_partition_id)
                for vertex in FecDataView.iterate_machine_vertices()
                if isinstance(vertex, AbstractSupportsDatabaseInjection)
                and vertex.is_in_injection_mode}
            machine_vertices.update(lpg_source_machine_vertices)
            live_vertices = FecDataView.iterate_live_output_vertices()
            for vertex, part_id in live_vertices:
                if isinstance(vertex, ApplicationVertex):
                    machine_vertices.update(
                        (m_vertex, part_id)
                        for m_vertex in vertex.splitter.get_out_going_vertices(
                            part_id))
                else:
                    machine_vertices.add((vertex, part_id))
            w.create_atom_to_event_id_mapping(machine_vertices)
        p.update()

1	# Copyright (c) 2015 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	from typing import Optional, Set, Tuple	1✔
17	from spinn_utilities.config_holder import (	1✔
18	get_config_bool, get_config_bool_or_none)
19	from spinn_utilities.progress_bar import ProgressBar	1✔
20	from spinn_utilities.log import FormatAdapter	1✔
21	from pacman.model.graphs.machine import MachineVertex	1✔
22	from pacman.model.graphs.application import ApplicationVertex	1✔
23	from spinn_front_end_common.utilities.database import DatabaseWriter	1✔
24	from spinn_front_end_common.abstract_models import (	1✔
25	AbstractSupportsDatabaseInjection)
26	from spinn_front_end_common.data import FecDataView	1✔
27
28	logger = FormatAdapter(logging.getLogger(__name__))	1✔
29
30
31	def database_interface(runtime: Optional[float]) -> Optional[str]:	1✔
32	"""
33	:param int runtime:
34	:return: where the database is located, if one is made
35	:rtype: str or None
36	"""
37	needs_db = DatabaseWriter.auto_detect_database()	1✔
38	user_create_database = get_config_bool_or_none(	1✔
39	"Database", "create_database")
40	if user_create_database is not None:	1!
41	if user_create_database != needs_db:	1!
NEW 42	logger.warning(	×
43	"Database creating changed to {} due to cfg settings",
44	user_create_database)
UNCOV 45	needs_db = user_create_database	×
46	if not needs_db:	1!
NEW 47	return None	×
48
49	with DatabaseWriter() as writer:	1✔
50	logger.info("Creating live event connection database in {}",	1✔
51	writer.database_path)
52	_write_to_db(writer, runtime)	1✔
53	return writer.database_path	1✔
54
55
56	def _write_to_db(w: DatabaseWriter, runtime: Optional[float]):	1✔
57	"""
58	:param DatabaseWriter w:
59	:param int runtime:
60	"""
61	with ProgressBar(6, "Creating graph description database") as p:	1✔
62	w.add_system_params(runtime)	1✔
63	w.add_proxy_configuration()	1✔
64	p.update()	1✔
65	w.add_machine_objects()	1✔
66	p.update()	1✔
67	w.add_application_vertices()	1✔
68	p.update()	1✔
69	w.add_placements()	1✔
70	p.update()	1✔
71	w.add_tags()	1✔
72	p.update()	1✔
73	lpg_source_machine_vertices = w.add_lpg_mapping()	1✔
74
75	if get_config_bool(	1!
76	"Database", "create_routing_info_to_neuron_id_mapping"):
77	machine_vertices: Set[Tuple[MachineVertex, str]] = {	1✔
78	(vertex, vertex.injection_partition_id)
79	for vertex in FecDataView.iterate_machine_vertices()
80	if isinstance(vertex, AbstractSupportsDatabaseInjection)
81	and vertex.is_in_injection_mode}
82	machine_vertices.update(lpg_source_machine_vertices)	1✔
83	live_vertices = FecDataView.iterate_live_output_vertices()	1✔
84	for vertex, part_id in live_vertices:	1!
85	if isinstance(vertex, ApplicationVertex):	×
86	machine_vertices.update(	×
87	(m_vertex, part_id)
88	for m_vertex in vertex.splitter.get_out_going_vertices(
89	part_id))
90	else:
91	machine_vertices.add((vertex, part_id))	×
92	w.create_atom_to_event_id_mapping(machine_vertices)	1✔
93	p.update()	1✔

SpiNNakerManchester / SpiNNFrontEndCommon / 6822522522

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