Source code for pacman.data.pacman_data_writer
# Copyright (c) 2021 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 spinn_utilities.log import FormatAdapter
from spinn_utilities.overrides import overrides
from spinn_machine.data.machine_data_writer import MachineDataWriter
from pacman.model.placements import Placements
from pacman.model.routing_info import RoutingInfo
from pacman.model.routing_table_by_partition import (
MulticastRoutingTableByPartition)
from pacman.model.routing_tables import MulticastRoutingTables
from pacman.model.tags import Tags
from pacman.exceptions import PacmanConfigurationException
from .pacman_data_view import PacmanDataView, _PacmanDataModel
logger = FormatAdapter(logging.getLogger(__name__))
__temp_dir = None
REPORTS_DIRNAME = "reports"
# pylint: disable=protected-access
[docs]class PacmanDataWriter(MachineDataWriter, PacmanDataView):
"""
See :py:class:`spinn_utilities.data.UtilsDataWriter`.
This class is designed to only be used directly within the PACMAN
repository unit tests as all methods are available to subclasses.
"""
__pacman_data = _PacmanDataModel()
__slots__ = []
@overrides(MachineDataWriter._mock)
def _mock(self):
MachineDataWriter._mock(self)
self.__pacman_data._clear()
@overrides(MachineDataWriter._setup)
def _setup(self):
MachineDataWriter._setup(self)
self.__pacman_data._clear()
@overrides(MachineDataWriter._hard_reset)
def _hard_reset(self):
MachineDataWriter._hard_reset(self)
self.__pacman_data._hard_reset()
@overrides(MachineDataWriter._soft_reset)
def _soft_reset(self):
MachineDataWriter._soft_reset(self)
self.__pacman_data._soft_reset()
[docs] def set_placements(self, placements):
"""
Set the placements.
:param Placements placements:
:raises TypeError: if the placements is not a Placements
"""
if not isinstance(placements, Placements):
raise TypeError("placements should be a Placements")
self.__pacman_data._placements = placements
[docs] def set_routing_infos(self, routing_infos):
"""
Set the routing_infos.
:param RoutingInfo routing_infos:
:raises TypeError: if the routing_infos is not a RoutingInfo
"""
if not isinstance(routing_infos, RoutingInfo):
raise TypeError("routing_infos should be a RoutingInfo")
self.__pacman_data._routing_infos = routing_infos
[docs] def set_uncompressed(self, router_tables):
"""
Sets the uncompressed `router_tables` value.
:param MulticastRoutingTables router_tables: new value
:raises TypeError:
if the router_tables is not a MulticastRoutingTables
"""
if not isinstance(router_tables, MulticastRoutingTables):
raise TypeError(
"router_tables should be a MulticastRoutingTables")
self.__pacman_data._uncompressed = router_tables
[docs] def set_precompressed(self, router_tables):
"""
Sets the precompressed `router_tables` value.
:param MulticastRoutingTables router_tables: new value
:raises TypeError:
if the router_tables is not a MulticastRoutingTables
"""
if not isinstance(router_tables, MulticastRoutingTables):
raise TypeError(
"router_tables should be a MulticastRoutingTables")
self.__pacman_data._precompressed = router_tables
[docs] def set_plan_n_timesteps(self, plan_n_timesteps):
"""
Sets the `plan_n_timestep`. Use `None` for run forever.
:param plan_n_timesteps:
:type plan_n_timesteps: int or None
:raises TypeError: if the plan_n_timesteps are not an int or `None`
:raises PacmanConfigurationException: On a negative plan_n_timesteps
"""
if plan_n_timesteps is not None:
if not isinstance(plan_n_timesteps, int):
raise TypeError("plan_n_timesteps should be an int")
if plan_n_timesteps < 0:
raise PacmanConfigurationException(
f"plan_n_timesteps {plan_n_timesteps} "
f"must not be negative")
self.__pacman_data._plan_n_timesteps = plan_n_timesteps
[docs] def set_routing_table_by_partition(self, routing_table_by_partition):
"""
Sets the `_routing_table_by_partition`.
:param MulticastRoutingTableByPartition routing_table_by_partition:
raises TypeError: if routing_table_by_partition is no a
MulticastRoutingTableByPartition
"""
if not isinstance(
routing_table_by_partition, MulticastRoutingTableByPartition):
raise TypeError(
"routing_table_by_partition should be a "
"MulticastRoutingTableByPartition")
self.__pacman_data._routing_table_by_partition = \
routing_table_by_partition
[docs] @classmethod
def add_vertex(cls, vertex):
if cls.__pacman_data._graph is None:
raise cls._exception("graph")
if not cls.get_requires_mapping():
raise PacmanConfigurationException(
"This call is only expected if requires mapping is True")
cls.__pacman_data._graph.add_vertex(vertex)
[docs] @classmethod
def add_edge(cls, edge, outgoing_edge_partition_name):
if cls.__pacman_data._graph is None:
raise cls._exception("graph")
if not cls.get_requires_mapping():
raise PacmanConfigurationException(
"This call is only expected if requires mapping is True")
cls.__pacman_data._graph.add_edge(edge, outgoing_edge_partition_name)
[docs] def add_monitor_all_chips(self, vertex):
"""
Reports that a monitor has been added to every Chip.
Should be called once for each monitor added to all Chips.
Should not be called for Ethernet only monitors.
Only affect is to change the numbers reported by the get_monitor
methods
:param ~pacman.model.graphs.machine.MachineVertex vertex:
One of the Vertcies added to each Core
"""
self.__pacman_data._monitor_cores += 1
self.__pacman_data._monitor_sdram += \
vertex.sdram_required.get_total_sdram(self.get_plan_n_timestep())