summaryrefslogtreecommitdiffstats
path: root/tests/tests_keras.py
blob: 5b7db2873738df677c82a371b0029a597262dc1f (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
from .tests_tqdm import importorskip, mark

pytestmark = mark.slow


@mark.filterwarnings("ignore:.*:DeprecationWarning")
def test_keras(capsys):
    """Test tqdm.keras.TqdmCallback"""
    TqdmCallback = importorskip('tqdm.keras').TqdmCallback
    np = importorskip('numpy')
    try:
        import keras as K
    except ImportError:
        K = importorskip('tensorflow.keras')

    # 1D autoencoder
    dtype = np.float32
    model = K.models.Sequential([
        K.layers.InputLayer((1, 1), dtype=dtype), K.layers.Conv1D(1, 1)])
    model.compile("adam", "mse")
    x = np.random.rand(100, 1, 1).astype(dtype)
    batch_size = 10
    batches = len(x) / batch_size
    epochs = 5

    # just epoch (no batch) progress
    model.fit(
        x,
        x,
        epochs=epochs,
        batch_size=batch_size,
        verbose=False,
        callbacks=[
            TqdmCallback(
                epochs,
                desc="training",
                data_size=len(x),
                batch_size=batch_size,
                verbose=0)])
    _, res = capsys.readouterr()
    assert "training: " in res
    assert f"{epochs}/{epochs}" in res
    assert f"{batches}/{batches}" not in res

    # full (epoch and batch) progress
    model.fit(
        x,
        x,
        epochs=epochs,
        batch_size=batch_size,
        verbose=False,
        callbacks=[
            TqdmCallback(
                epochs,
                desc="training",
                data_size=len(x),
                batch_size=batch_size,
                verbose=2)])
    _, res = capsys.readouterr()
    assert "training: " in res
    assert f"{epochs}/{epochs}" in res
    assert f"{batches}/{batches}" in res

    # auto-detect epochs and batches
    model.fit(
        x,
        x,
        epochs=epochs,
        batch_size=batch_size,
        verbose=False,
        callbacks=[TqdmCallback(desc="training", verbose=2)])
    _, res = capsys.readouterr()
    assert "training: " in res
    assert f"{epochs}/{epochs}" in res
    assert f"{batches}/{batches}" in res

    # continue training (start from epoch != 0)
    initial_epoch = 3
    model.fit(
        x,
        x,
        initial_epoch=initial_epoch,
        epochs=epochs,
        batch_size=batch_size,
        verbose=False,
        callbacks=[TqdmCallback(desc="training", verbose=0,
                                miniters=1, mininterval=0, maxinterval=0)])
    _, res = capsys.readouterr()
    assert "training: " in res
    assert f"{initial_epoch - 1}/{initial_epoch - 1}" not in res
    assert f"{epochs}/{epochs}" in res