Commit 7593328d authored by Chanelle Lee's avatar Chanelle Lee
Browse files

Working on adding the simulation testing. Almost complete - just need to...

Working on adding the simulation testing. Almost complete - just need to finish the tests for test_run_valid but the template is there. All tests working.
parent 37bc9be3
......@@ -11,24 +11,29 @@ class Simulation(object):
saves the data as a numpy file.
"""
def __init__(self, parameters, optionWeights):
def __init__(self, totalIts, popSize, numOptions, poolSize, distrust, w,
evidenceRate, optionWeights):
"""
Initialising method for Simulation class
Parameters
----------
parameters : dict
Contains simulation parameters generated by runParams.py
{totalIts: int, popSize: int, k: int, eR: int}
optionWeights : list
totalIts : int
total number of iterations to run the simulation for
popSize : int
number of agents in the population
numOptions : int
number of options system is choosing between
poolSize : int
number of agents in each aggregation pool
evidenceRate : float in [0,1]
chance that the agents receive evidence each iteration
optionWeights : tuple of floats sums to 1
Contains the weightings for the evidence
Attributes
----------
totalIts : int
total number of iterations for which to update the simulation
neighbourhoodSize : int
pool size for aggregation i.e. number of agents in aggregation
population : instance
Instance of the Population class
evidence : instance
......@@ -37,92 +42,78 @@ class Simulation(object):
To contain the beliefs of every agents at every iteration.
[totalIts x popSize]
"""
self._totalIts = parameters['totalIts']
self._neighbourhoodSize = parameters['k']
self._evidence = Evidence(parameters['numOptions'],
parameters['eR'],
optionWeights)
self._population = Population(distrust=parameters['distrust'],
w=parameters['w'],
numOptions=parameters['numOptions'],
size=parameters['popSize']
)
self.populationBeliefs = np.empty((self._totalIts,
parameters['popSize'],
parameters['numOptions']),
self._totalIts = totalIts
self._poolSize = poolSize
self._evidence = Evidence(numOptions, evidenceRate, optionWeights)
self._population = Population(popSize, numOptions, distrust, w)
self.populationBeliefs = np.empty((totalIts, popSize, numOptions),
dtype='float32')
def save(self, i, filepath):
"""
Saves the populationBeliefs as a numpy file. Pickling is not allowed.
def checkConvergenceReached(self, i):
try:
return all([any(belief == 1.0 for belief in agent)
for agent in self.populationBeliefs[i]])
except IndexError:
raise ValueError("Passed index is out of range")
Parameters
----------
i : int
Identifying number of simulation for naming the file.
filePath : str
Path and file name for saving simulation numpy files.
e.g. path + 'AbBoNK{}E{}W{}'.format(str(k),
str(e).replace('.', '-'),
str(w).replace('.', '-'))
def checkForCorrectConsensus(self, i):
"""
outfile = filepath + '/sim{}'.format(i)
np.save(outfile, self.populationBeliefs, allow_pickle=False)
Checks whether all agents in the population have total belief in option
0 - the best option. Does not count the cases where the population have
reached consensus on another option.
"""
try:
return all(self.populationBeliefs[i][:, 0] == 1.0)
except IndexError:
raise ValueError("Passed index is outside of range")
def savez(self, i, filepath):
def update(self, i):
"""
Saves the populationBeliefs as a compressed numpy file.
Updates population at each iteration.
Agents received evidence, move and then aggregate.
"""
outfile = filepath + '/sim{}'.format(i)
np.savez_compressed(outfile, self.populationBeliefs)
if i > self._totalIts:
raise ValueError('Current iterations greater than total!!')
def cleanUp(self):
"""Ensures large objects stored in the instance are deleted"""
del self._population
del self.populationBeliefs
# Population receives evidence
self._population.receiveEvidence(self._evidence)
# Mimic movement in the population
movedAgents = self._population.moveAgents()
# Now perform aggregation based on chosen agg type.
self._population.aggregation(movedAgents, self._poolSize)
self.populationBeliefs[i] = self._population.retBeliefs()
del movedAgents
def run(self):
"""
Updates simulation and stores population belief at each iteration
Differs from Simulation.run() as it doesn't wait for ten
iterations before checking consensus. Also checks that consensus is to
the `best' option which is always option 0 in this simulation.
Checks that consensus is to the `best' option which is always option 0
in this simulation.
"""
self.populationBeliefs[0] = self._population.retBeliefs()
for i in range(1, self._totalIts):
self.update(i)
self.populationBeliefs[i] = self._population.retBeliefs()
if self.checkConvergenceReached(i):
self.populationBeliefs[i:] = self.populationBeliefs[i]
if self.checkForConsensus(i):
if self.checkForCorrectConsensus(i):
return i
else:
return self._totalIts
return self._totalIts
def update(self, i):
def cleanUp(self):
"""Ensures large objects stored in the instance are deleted"""
del self._population
del self.populationBeliefs
def savez(self, i, filepath):
"""
Updates population at each iteration.
Agents received evidece, move and then aggregate.
Saves the populationBeliefs as a compressed numpy file.
"""
if i > self._totalIts:
raise ValueError('Current iterations greater than total!!')
# Population receives evidence
self._population.receiveEvidence(self._evidence)
# Mimic movement in the population
movedAgents = self._population.moveAgents()
# Now perform aggregation based on chosen agg type.
self._population.aggregation(movedAgents, self._neighbourhoodSize)
del movedAgents
def checkForConsensus(self, i):
return all(self.populationBeliefs[i][:, 0] == 1.0)
def checkConvergenceReached(self, i):
return all([any(belief == 1.0 for belief in agent)
for agent in self.populationBeliefs[i]])
outfile = filepath + '/sim{}'.format(i)
np.savez_compressed(outfile, self.populationBeliefs)
import pytest
import numpy.testing
from noComparisonSimulation.simulation import *
from noComparisonSimulation.population import Population
from noComparisonSimulation.evidence import Evidence
def test_initialise_valid():
total_its = 20
pop_size = 10
num_options = 3
pool_size = 10
distrust = 0.001
w = 1.0
evidence_rate = 0.7
option_weights = (0.8, 0.1, 0.1)
sim = Simulation(total_its, pop_size, num_options, pool_size, distrust, w,
evidence_rate, option_weights)
assert sim._totalIts == total_its
assert sim._poolSize == pool_size
assert isinstance(sim._evidence, Evidence)
assert sim._evidence.numOptions == num_options
assert sim._evidence.evidenceRate == evidence_rate
assert sim._evidence.optionWeights == option_weights
assert isinstance(sim._population, Population)
assert sim._population._size == pop_size
assert sim._population._numOptions == num_options
assert sim.populationBeliefs.shape == (total_its, pop_size, num_options)
@pytest.mark.parametrize("num_options, pop_size, i, pop_belief, expected", [
(2, 2, 0, np.array([[1.0, 0.0], [1.0, 0.0]]), True),
(2, 2, 0, np.array([[1.0, 0.0], [0.0, 1.0]]), True),
(3, 3, 0, np.array([[1.0, 0.0, 0.0], [1.0, 0.0, 0.0], [1.0, 0.0, 0.0]]),
True), # correct consensus reached - all full belief in option 0
(3, 3, 0, np.array([[0.6, 0.2, 0.2], [0.8, 0.1, 0.1], [0.5, 0.3, 0.2]]),
False), # not even convergence
(3, 3, 5, np.array([[0.0, 0.0, 1.0], [0.0, 0.0, 1.0], [0.0, 0.0, 1.0]]),
True), # convergence and consensus but to wrong option
(3, 3, 5, np.array([[1.0, 0.0, 0.0], [0.0, 0.0, 1.0], [1.0, 0.0, 0.0]]),
True) # convergence but no consensus
])
def test_checkConvergenceReached_valid(num_options, pop_size, i, pop_belief,
expected):
sim = Simulation(20, pop_size, num_options, 2, 0.001, 1.0, 0.7,
num_options * (1 / num_options, ))
sim.populationBeliefs[i] = pop_belief
result = sim.checkConvergenceReached(i)
assert result is expected
def test_checkConvergenceReached_invalid_index():
sim = Simulation(20, 10, 4, 2, 0.001, 1.0, 0.7, 4 * (0.25, ))
with pytest.raises(ValueError):
sim.checkConvergenceReached(21)
@pytest.mark.parametrize("num_options, pop_size, i, pop_belief, expected", [
(2, 2, 0, np.array([[1.0, 0.0], [1.0, 0.0]]), True),
(2, 2, 0, np.array([[1.0, 0.0], [0.0, 1.0]]), False),
(3, 3, 0, np.array([[1.0, 0.0, 0.0], [1.0, 0.0, 0.0], [1.0, 0.0, 0.0]]),
True), # correct consensus reached - all full belief in option 0
(3, 3, 0, np.array([[0.6, 0.2, 0.2], [0.8, 0.1, 0.1], [0.5, 0.3, 0.2]]),
False), # not even convergence
(3, 3, 5, np.array([[0.0, 0.0, 1.0], [0.0, 0.0, 1.0], [0.0, 0.0, 1.0]]),
False), # convergence and consensus but to wrong option
(3, 3, 5, np.array([[1.0, 0.0, 0.0], [0.0, 0.0, 1.0], [1.0, 0.0, 0.0]]),
False) # convergence but no consensus
])
def test_checkForCorrectConsensus_valid(num_options, pop_size, i, pop_belief,
expected):
sim = Simulation(20, pop_size, num_options, 2, 0.001, 1.0, 0.7,
num_options * (1 / num_options, ))
sim.populationBeliefs[i] = pop_belief
result = sim.checkForCorrectConsensus(i)
assert result is expected
def test_checkForCorrectConsensus_invalid_index():
sim = Simulation(20, 10, 4, 2, 0.001, 1.0, 0.7, 4 * (0.25, ))
with pytest.raises(ValueError):
sim.checkForCorrectConsensus(21)
@pytest.mark.parametrize(
("num_options, pop_size, pool_size, distrust, evidence_rate,"
"option_weights, index, expected"), [
# no pooling and no evidence
(2, 2, 0, 0.2, 0.0, (1.0, 0.0), 1, np.array([[0.5, 0.5],
[0.5, 0.5]])),
# complete pooling and no evidence
(2, 2, 2, 0.2, 0.0, (1.0, 0.0), 1, np.array([[0.5, 0.5],
[0.5, 0.5]])),
# no pooling and complete evidence
(2, 2, 0, 0.2, 1.0, (1.0, 0.0), 1, np.array([[0.8, 0.2],
[0.8, 0.2]])),
# complete pooling and complete evidence
(2, 2, 2, 0.2, 1.0, (1.0, 0.0), 1, np.array([[0.9412, 0.0588],
[0.9412, 0.0588]])),
# maintained with index
(2, 2, 2, 0.2, 1.0, (1.0, 0.0), 3, np.array([[0.9412, 0.0588],
[0.9412, 0.0588]])),
# different distrust
(2, 2, 2, 0.4, 1.0, (0.0, 1.0), 1, np.array([[0.3077, 0.6923],
[0.3077, 0.6923]])),
# Different n and pop size / no evidence / partial pooling
(4, 3, 2, 0.2, 0.0, (0.0, 1.0, 0.0, 0.0), 1,
np.array([[0.25, 0.25, 0.25, 0.25],
[0.25, 0.25, 0.25, 0.25],
[0.25, 0.25, 0.25, 0.25]])),
(4, 3, 0, 0.2, 1.0, (0.0, 1.0, 0.0, 0.0), 1,
np.array([[0.1429, 0.5714, 0.1429, 0.1429],
[0.1429, 0.5714, 0.1429, 0.1429],
[0.1429, 0.5714, 0.1429, 0.1429]]))
])
def test_update_valid(num_options, pop_size, pool_size, distrust,
evidence_rate, option_weights, index, expected):
sim = Simulation(20, pop_size, num_options, pool_size, distrust, 1.0,
evidence_rate, option_weights)
sim.update(index)
numpy.testing.assert_almost_equal(sim.populationBeliefs[index],
expected,
decimal=4)
# N.B. Not fully testing partial pooling as that requires mocking the
# moveAgents function and has already been tested in test_population
def test_update_invalid_value():
sim = Simulation(20, 10, 4, 2, 0.001, 1.0, 0.7, 4 * (0.25, ))
with pytest.raises(ValueError):
sim.update(21)
@pytest.mark.parametrize(
("num_options, pop_size, pool_size, distrust, evidence_rate,"
"option_weights, expected_result, expected_beliefs"), [
# no pooling and no evidence
(2, 2, 0, 0.2, 0.0, (1.0, 0.0), 3, 0.5 * np.ones((3, 2, 2)))
])
def test_run_valid(num_options, pop_size, pool_size, distrust, evidence_rate,
option_weights, expected_result, expected_beliefs):
sim = Simulation(3, pop_size, num_options, pool_size, distrust, 1.0,
evidence_rate, option_weights)
result = sim.run()
assert result == expected_result
numpy.testing.assert_almost_equal(sim.populationBeliefs,
expected_beliefs,
decimal=4)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment