test/python/test_streaming.py

Snapshot from this documentation build. SHA-256: 0857fb66e1364de6cef444f40443983b6626ab95942663abbe117a93401f586e

    1 from pathlib import Path
    2 import json
    3 
    4 import numpy as np
    5 from PIL import Image
    6 import pytest
    7 
    8 from tools.reference.streaming import growth
    9 from tools.analysis.render_density_frames import render, animate
   10 
   11 
   12 def test_matter_growing_mode_and_transverse_streaming():
   13     a, y = growth()
   14     np.testing.assert_allclose(y[0], a/a[0], rtol=2e-9)
   15     np.testing.assert_allclose(y[2], a/a[0], rtol=2e-9)
   16     _, transverse = growth(vbc_kms=(0, 30, 0))
   17     np.testing.assert_allclose(y, transverse, rtol=1e-12)
   18 
   19 
   20 def test_streaming_phase_reversal_and_dm_growth_suppression():
   21     _, zero = growth(k_mpc=(1000, 0, 0))
   22     _, plus = growth(k_mpc=(1000, 0, 0), vbc_kms=(30, 0, 0))
   23     _, minus = growth(k_mpc=(1000, 0, 0), vbc_kms=(-30, 0, 0))
   24     np.testing.assert_allclose(plus, minus.conj(), rtol=2e-9, atol=1e-10)
   25     assert abs(plus[0, -1]) < 0.9*abs(zero[0, -1])
   26     assert abs(plus[2, -1]) < abs(zero[2, -1])
   27     assert abs(plus[0, -1].imag) > 1e-3
   28 
   29 
   30 def test_gif_uses_manifest_order_and_preserves_epochs_and_hashes(tmp_path: Path):
   31     index = 'cycle\tz\tN\tslice_f32\tmaxproj_f32\tlayout\n'
   32     for cycle, z, value in ((2, 900.25, 2), (0, 1100.5, 1)):
   33         np.full(16, value, dtype='<f4').tofile(tmp_path/f'{cycle}.f32')
   34         index += f'{cycle}\t{z}\t4\t{cycle}.f32\t{cycle}.f32\tjulia-column-major\n'
   35     (tmp_path/'frames.tsv').write_text(index)
   36     out = tmp_path/'plots'
   37     records = render(tmp_path, out)
   38     assert [r['cycle'] for r in records] == [0, 2]
   39     animate(out, records, out/'density.gif', fps=5)
   40     with Image.open(out/'density.gif') as image:
   41         assert image.n_frames == 2
   42         assert image.info['duration'] == 200
   43     summary = json.loads((out/'frames.json').read_text())
   44     assert summary['log10_limits'] == [-2, 2]
   45     assert summary['frames'][0]['z'] == 1100.5
   46     assert len(summary['frames'][0]['payload_sha256']['slice_f32']) == 64
   47     with pytest.raises(ValueError):
   48         animate(out, records, out/'bad.gif', fps=0)
   49     (tmp_path/'frames.tsv').write_text(index+index.splitlines()[-1]+'\n')
   50     with pytest.raises(ValueError, match='duplicate'):
   51         render(tmp_path, out)