Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,5 +17,5 @@ requires = ["setuptools"]
build-backend = "setuptools.build_meta"

[[tool.mypy.overrides]]
module = ["matplotlib", "matplotlib.*", "pyNN.*", "quantities", "neo", "neo.*", "scipy", "scipy.*", "lazyarray"]
module = ["csa", "docstring_parser", "matplotlib", "matplotlib.*", "pyNN.*", "quantities", "neo", "neo.*", "scipy", "scipy.*", "lazyarray"]
ignore_missing_imports = true
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,14 @@
import logging
import math
import re
from typing import Dict, Optional, Sequence, Tuple, Union, TYPE_CHECKING
from typing import Any, Dict, Optional, Sequence, Tuple, Union, TYPE_CHECKING
from typing_extensions import Never

import numpy
from numpy import float64, uint32, uint16, uint8
from numpy.typing import NDArray

from pyNN import descriptions
from pyNN.random import NumpyRNG, RandomDistribution
from pyNN.space import Space

Expand Down Expand Up @@ -679,3 +680,60 @@ def validate_connection(
"Using a projection where the source or target is a "
"PopulationView on a multi-dimensional Population is not "
"supported")

def get_parameters(self) -> Dict[str, Any]:
"""
A list of parameters that would recreate this connector.

May not be exactly the same as the ones used to construct this.

Will also not include any information or changes added by later calls.

:return: A map of the init parameters to the values passed in.
"""
# The default is error
# This to avoid missing parameters in user Connectors
raise NotImplementedError(
f"{type(self)} does not implement "
f"Standard pyNN get_parameters method")

def _get_parameters(self) -> Dict[str, Any]:
"""
:return: A map of the init parameters to the values passed in.
"""
return {
"safe": self.safe,
"verbose": self.verbose,
"callback": None
}

def clone(self) -> "AbstractConnector":
"""
Create a clone of the Connector at init point

This is a best effort attempt and its use is not recommended

:return: A clone of the Connector. Ignoring any SynapseInformation
"""
theType = type(self)
params = self.get_parameters()
logger.warning(f"Cloning type{self} is not recommended "
f"and may lead to incorrect results.")
return theType(**params)

def describe(self, template: Optional[str] = None,
engine: str = 'default') -> str:
"""
Returns a human-readable description of the connection method.

The output may be customized by specifying a different template
together with an associated template engine (see pyNN.descriptions).

If template is None,
then a dictionary containing the template context will be returned.

:return: A human-readable description of the connector.
"""
context = {"Type": self.__class__.__name__}
context.update(self.get_parameters())
return descriptions.render(engine, template, context)
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
from typing import Sequence, Optional, TYPE_CHECKING
from typing import Any, Dict, Sequence, Optional, TYPE_CHECKING

import numpy
from numpy import uint32
Expand Down Expand Up @@ -92,6 +92,13 @@ def __init__(self, n_neurons_per_group: Optional[int] = None,
self.__weights = weights
self.__check_weights(weights, n_neurons_per_group)

@overrides(AbstractGenerateConnectorOnMachine.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
parameters["n_neurons_per_group"] = self.__n_neurons_per_group
parameters["weights"] = self.__weights
return parameters

def __check_weights(self, weights: Optional[NDArray[numpy.float64]],
n_neurons_per_group: Optional[int]) -> None:
if weights is not None and n_neurons_per_group is not None:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
from typing import Sequence, Optional, TYPE_CHECKING
from typing import Any, Dict, Optional, Sequence, TYPE_CHECKING

import numpy
from numpy import uint32
Expand Down Expand Up @@ -65,6 +65,12 @@ def __init__(self, allow_self_connections: bool = True, safe: bool = True,
super().__init__(safe, callback, verbose)
self.__allow_self_connections = allow_self_connections

@overrides(AbstractGenerateConnectorOnMachine.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
parameters["allow_self_connections"] = self.allow_self_connections
return parameters

@overrides(AbstractConnector.get_delay_maximum)
def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:
return self._get_delay_maximum(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
# limitations under the License.

from __future__ import annotations
from typing import Sequence, Optional, TYPE_CHECKING
from typing import Any, Dict, Optional, Sequence, TYPE_CHECKING

import numpy
from numpy import uint8
Expand Down Expand Up @@ -76,6 +76,12 @@ def __init__(self, array: NDArray[uint8], safe: bool = True,
self.__n_total_connections = n_total_connections
self.__array_dims = dims

@overrides(AbstractConnector.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
parameters["array"] = self.__array
return parameters

@overrides(AbstractConnector.get_delay_maximum)
def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:
return self._get_delay_maximum(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
from __future__ import annotations
from collections.abc import (Iterable, Sequence)
from typing import (
List, Optional, Sequence as TSequence, Tuple, Union,
Any, Dict, List, Optional, Sequence as TSequence, Tuple, Union,
cast, overload, TYPE_CHECKING)

import numpy
Expand Down Expand Up @@ -169,10 +169,34 @@ def __init__(self, kernel_weights: _Weights,
self.__pool_stride = self.__pool_shape
if self.__pool_shape is not None:
self.__kernel_weights /= numpy.prod(self.__pool_shape)

self.__positive_receptor_type = positive_receptor_type
self.__negative_receptor_type = negative_receptor_type

@overrides(AbstractConnector.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
if self.__pool_shape is None:
parameters["kernel_weights"] = self.__kernel_weights
else:
parameters["kernel_weights"] = (
self.__kernel_weights * numpy.prod(self.__pool_shape))
# Now included in weights
parameters["kernel_shape"] = None
parameters["strides"] = tuple(self.__strides)
parameters["padding"] = tuple(self.__padding_shape)
if self.__pool_shape is None:
parameters["pool_shape"] = None
else:
parameters["pool_shape"] = tuple(self.__pool_shape)
if self.__pool_stride is None:
parameters["pool_stride"] = None
else:
parameters["pool_stride"] = tuple(self.__pool_stride)
parameters["positive_receptor_type"] = self.__positive_receptor_type
parameters["negative_receptor_type"] = self.__negative_receptor_type
parameters["filter_edges"] = self.__filter_edges
return parameters

@property
def positive_receptor_type(self) -> str:
"""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from __future__ import annotations
from typing import List, Optional, Tuple, TYPE_CHECKING, Sequence
from typing import Any, Dict, List, Optional, Tuple, TYPE_CHECKING, Sequence

import numpy
from numpy.typing import NDArray
Expand Down Expand Up @@ -81,6 +81,12 @@ def __init__(self, cset: CSet, safe: bool = True, callback: None = None,
self.__full_connection_set: Optional[List[CSet]] = None
self.__full_cset: Optional[List[CSet]] = None

@overrides(AbstractConnector.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
parameters["cset"] = self.__cset
return parameters

@overrides(AbstractConnector.get_delay_maximum)
def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:
n_conns_max = synapse_info.n_pre_neurons * synapse_info.n_post_neurons
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@

from __future__ import annotations
import math
from typing import Optional, Sequence, TYPE_CHECKING
from typing import Any, Dict, Optional, Sequence, TYPE_CHECKING

import numpy
from numpy import (
Expand Down Expand Up @@ -102,6 +102,14 @@ def __init__(
"n_connections is not implemented for"
" DistanceDependentProbabilityConnector on this platform")

@overrides(AbstractConnector.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
parameters["d_expression"] = self.d_expression
parameters["allow_self_connections"] = self.__allow_self_connections
parameters["rng"] = self.__rng
return parameters

@overrides(AbstractConnector.set_projection_information)
def set_projection_information(
self, synapse_info: SynapseInformation) -> None:
Expand Down Expand Up @@ -131,7 +139,6 @@ def _set_probabilities(self, synapse_info: SynapseInformation) -> None:

self.__probs = _d_expr_context.eval(self.__d_expression, d=d)

@property
def _probs(self) -> NDArray[floating]:
if self.__probs is None:
raise ValueError("no projection information set")
Expand All @@ -144,7 +151,7 @@ def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:
get_probable_maximum_selected(
synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
numpy.amax(self._probs)),
numpy.amax(self._probs())),
synapse_info)

@overrides(AbstractConnector.get_delay_minimum)
Expand All @@ -154,15 +161,15 @@ def get_delay_minimum(self, synapse_info: SynapseInformation) -> float:
get_probable_minimum_selected(
synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
numpy.amax(self._probs)),
numpy.amax(self._probs())),
synapse_info)

@overrides(AbstractConnector.get_n_connections_from_pre_vertex_maximum)
def get_n_connections_from_pre_vertex_maximum(
self, n_post_atoms: int, synapse_info: SynapseInformation,
min_delay: Optional[float] = None,
max_delay: Optional[float] = None) -> int:
max_prob = numpy.amax(self._probs)
max_prob = numpy.amax(self._probs())
n_connections = get_probable_maximum_selected(
synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
n_post_atoms, max_prob)
Expand All @@ -181,7 +188,7 @@ def get_n_connections_to_post_vertex_maximum(
return get_probable_maximum_selected(
synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
synapse_info.n_post_neurons,
numpy.amax(self._probs))
numpy.amax(self._probs()))

@overrides(AbstractConnector.get_weight_maximum)
def get_weight_maximum(self, synapse_info: SynapseInformation) -> float:
Expand All @@ -190,14 +197,15 @@ def get_weight_maximum(self, synapse_info: SynapseInformation) -> float:
get_probable_maximum_selected(
synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
synapse_info.n_pre_neurons * synapse_info.n_post_neurons,
numpy.amax(self._probs)),
numpy.amax(self._probs())),
synapse_info)

@overrides(AbstractGenerateConnectorOnHost.create_synaptic_block)
def create_synaptic_block(
self, post_slices: Sequence[Slice], post_vertex_slice: Slice,
synapse_type: int, synapse_info: SynapseInformation) -> NDArray:
probs = self._probs[:, post_vertex_slice.get_raster_ids()].reshape(-1)
probs = self._probs()[
:, post_vertex_slice.get_raster_ids()].reshape(-1)
n_items = synapse_info.n_pre_neurons * post_vertex_slice.n_atoms
items = self.__rng.next(n_items)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
# limitations under the License.
from __future__ import annotations
import math
from typing import List, Optional, Sequence, TYPE_CHECKING
from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING

import numpy
from numpy import integer, uint32
Expand Down Expand Up @@ -99,6 +99,15 @@ def __init__(
self.__post_neurons_set = False
self.__rng = rng

@overrides(AbstractGenerateConnectorOnMachine.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
parameters["n"] = self.__n_post
parameters["allow_self_connections"] = self.allow_self_connections
parameters["with_replacement"] = self.__with_replacement
parameters["rng"] = self.__rng
return parameters

def set_projection_information(
self, synapse_info: SynapseInformation) -> None:
super().set_projection_information(synapse_info)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
# limitations under the License.
from __future__ import annotations
import math
from typing import List, Optional, Sequence, TYPE_CHECKING
from typing import Any, Dict, List, Optional, Sequence, TYPE_CHECKING

import numpy
from numpy import integer, uint32
Expand Down Expand Up @@ -96,6 +96,15 @@ def __init__(
self.__pre_neurons: List[NDArray[integer]] = []
self.__rng = rng

@overrides(AbstractGenerateConnectorOnMachine.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
parameters["n"] = self.__n_pre
parameters["allow_self_connections"] = self.allow_self_connections
parameters["with_replacement"] = self.__with_replacement
parameters["rng"] = self.__rng
return parameters

def set_projection_information(
self, synapse_info: SynapseInformation) -> None:
super().set_projection_information(synapse_info)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
from __future__ import annotations
import logging
import math
from typing import Optional, Sequence, TYPE_CHECKING
from typing import Any, Dict, Optional, Sequence, TYPE_CHECKING

import numpy
from numpy.typing import NDArray
Expand Down Expand Up @@ -100,6 +100,14 @@ def __init__(
self.__allow_self_connections = allow_self_connections
self.__rng = rng

@overrides(AbstractGenerateConnectorOnMachine.get_parameters)
def get_parameters(self) -> Dict[str, Any]:
parameters = self._get_parameters()
parameters["p_connect"] = self.p_connect
parameters["allow_self_connections"] = self.__allow_self_connections
parameters["rng"] = self.__rng
return parameters

@overrides(AbstractConnector.get_delay_maximum)
def get_delay_maximum(self, synapse_info: SynapseInformation) -> float:
n_connections = get_probable_maximum_selected(
Expand Down
Loading
Loading