Skip to content

Commit

Permalink
move data to fixture to hopefully fix network issues
Browse files Browse the repository at this point in the history
  • Loading branch information
ljwolf committed Aug 16, 2024
1 parent f948ac4 commit 29b2312
Showing 1 changed file with 51 additions and 42 deletions.
93 changes: 51 additions & 42 deletions esda/tests/test_moran_local_mv.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,39 +8,48 @@
from esda.moran_local_mv import Partial_Moran_Local, Auxiliary_Moran_Local
from esda.moran import Moran_Local_BV

df = geopandas.read_file(geodatasets.get_path("geoda.lansing1"))
df = df[df.FIPS.str.match("2606500[01234]...") | (df.FIPS == "26065006500")]

y = df.HH_INC.values.reshape(-1,1)
X = df.HSG_VAL.values.reshape(-1,1)
def rsqueen(df):
w_classic = Queen.from_dataframe(df)
w_classic.transform = 'r'
return w_classic

test_over_w = pytest.mark.parametrize("w",
[rsqueen(df),
Graph.build_contiguity(df).transform("r")],
ids=['W', 'Graph'])

@test_over_w
def test_partial_runs(w):
@pytest.fixture(scope='module')
def data():
df = geopandas.read_file(geodatasets.get_path("geoda.lansing1"))
df = df[df.FIPS.str.match("2606500[01234]...") | (df.FIPS == "26065006500")]
y = df.HH_INC.values.reshape(-1,1)
X = df.HSG_VAL.values.reshape(-1,1)
yield y,X,df

@pytest.fixture(scope='module',
params = [
rsqueen,
lambda df: Graph.build_contiguity(df).transform('r')
],
ids=['W', 'Graph']
)
def graph(data, request):
_,_,df = data
return request.param(df)

def test_partial_runs(data, graph):
"""Check if the class computes successfully in a default configuration"""
m = Partial_Moran_Local(y,X,w, permutations=1)
y,X,df = data
m = Partial_Moran_Local(y,X,graph, permutations=1)
# done, just check if it runs

@test_over_w
def test_partial_accuracy(w):
def test_partial_accuracy(data, graph):
"""Check if the class outputs expected results at a given seed"""
y,X,df = data
numpy.random.seed(111221)
m = Partial_Moran_Local(y,X,w, permutations=10)
m = Partial_Moran_Local(y,X,graph, permutations=10)
# compute result by hand
zy = (y - y.mean())/y.std()
wz = (w.sparse @ zy)
wz = (graph.sparse @ zy)
zx = (X - X.mean(axis=0))/X.std(axis=0)
rho = numpy.corrcoef(zy.squeeze(), zx.squeeze())[0,1]
left = zy - rho * zx
scale = (w.n-1) / (w.n * (1 - rho**2))
scale = (graph.n-1) / (graph.n * (1 - rho**2))
# (y - rho x)*wy
manual = (left*wz).squeeze() * scale
# check values
Expand All @@ -55,45 +64,45 @@ def test_partial_accuracy(w):
is_odd_label = m.labels_ % 2
numpy.testing.assert_equal(is_cluster, is_odd_label)

@test_over_w
def test_partial_unscaled(w):
def test_partial_unscaled(data, graph):
"""Check if the variance scaling behaves as expected"""
m = Partial_Moran_Local(y,X,w, permutations=0, unit_scale=True)
m2 = Partial_Moran_Local(y,X,w, permutations=0, unit_scale=False)
y,X,df = data
m = Partial_Moran_Local(y,X,graph, permutations=0, unit_scale=True)
m2 = Partial_Moran_Local(y,X,graph, permutations=0, unit_scale=False)
# variance in the partials_ should be different
s1y,s1x = m.partials_.std(axis=0)
s2y,s2x = m2.partials_.std(axis=0)
assert s1y > s2y, "variance is incorrectly scaled for y"
assert s1x < s2x, "variance is incorrectly scaled for x"

@test_over_w
def test_partial_uvquads(w):
def test_partial_uvquads(data, graph):
"""Check that the quadrant decisions vary correctly with the inputs"""
m = Partial_Moran_Local(y,X,w, permutations=0, mvquads=False)
bvx = Moran_Local_BV(X,y,w,permutations=0)
y,X,df = data
m = Partial_Moran_Local(y,X,graph, permutations=0, mvquads=False)
bvx = Moran_Local_BV(X,y,graph,permutations=0)
# TODO: this currently fails, and it should pass. I am probably mis-calculating the bivariate quadrants for this option, and need to correct the code.
numpy.testing.assert_array_equal(m.quads_, bvx.q)

@test_over_w
def test_aux_runs(w):
def test_aux_runs(data, graph):
"""Check that the class completes successfully in a default configuration"""
a = Auxiliary_Moran_Local(y,X, w, permutations=1)
y,X,df = data
a = Auxiliary_Moran_Local(y,X, graph, permutations=1)
#done, just check if it runs

@test_over_w
def test_aux_accuracy(w):
def test_aux_accuracy(data, graph):
"""Check that the class outputs expected values for a given seed"""
y,X,df = data
numpy.random.seed(111221)
a = Auxiliary_Moran_Local(y,X,w, permutations=10)
a = Auxiliary_Moran_Local(y,X,graph, permutations=10)

# compute result by hand
zy = (y - y.mean())/y.std()
wz = (w.sparse @ zy)
wz = (graph.sparse @ zy)
zx = (X - X.mean(axis=0))/X.std(axis=0)
wzx = w.sparse @ zx
wzx = graph.sparse @ zx
rho = numpy.corrcoef(zy.squeeze(), zx.squeeze())[0,1]
mean = zy * wz - rho * zx * wz - rho * zy * wzx + rho**2 * zx * wzx
scale = (w.n-1) / (w.n * (1 - rho**2))
scale = (graph.n-1) / (graph.n * (1 - rho**2))

manual = numpy.asarray(mean * scale).squeeze()
# check values, may not be identical because of the
Expand All @@ -108,17 +117,17 @@ def test_aux_accuracy(w):
is_odd_label = (a.labels_ % 2).astype(bool)
numpy.testing.assert_equal(is_cluster, is_odd_label)

@test_over_w
def test_aux_unscaled(w):
def test_aux_unscaled(data, graph):
"""Check that the variance scaling behaves as expected"""
a = Auxiliary_Moran_Local(y,X/10000,w, permutations=0, unit_scale=True)
a2 = Auxiliary_Moran_Local(y,X,w, permutations=0, unit_scale=False)
y,X,df = data
a = Auxiliary_Moran_Local(y,X/10000,graph, permutations=0, unit_scale=True)
a2 = Auxiliary_Moran_Local(y,X,graph, permutations=0, unit_scale=False)
assert (a.partials_.std(axis=0) < a2.partials_.std(axis=0)).all(), (
"variance is not scaled correctly in partial regression."
)

@test_over_w
def test_aux_transformer(w):
def test_aux_transformer(data, graph):
"""Check that an alternative regressor can be used to calculate y|X"""
a = Auxiliary_Moran_Local(y,X,w, permutations=0, transformer=TheilSenRegressor)
y,X,df = data
a = Auxiliary_Moran_Local(y,X,graph, permutations=0, transformer=TheilSenRegressor)
# done, should just complete

0 comments on commit 29b2312

Please sign in to comment.