argopy-status/check_save_status.py at master · euroargodev/argopy-status · GitHub

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
92
93
94
95
96
97
98
import os
import json
from pathlib import Path

from argopy.utils import isAPIconnected, list_available_data_src, GreenCoding


COLORS = {"up": "green", "down": "red", "unknown": "black"}


def save_to_json(label, message, color, outfile):
    # Create json file with full results for a shieldio badge
    data = {}
    data["schemaVersion"] = 1
    data["label"] = label
    data["message"] = message
    data["color"] = color
    with open(outfile, "w") as f:
        json.dump(data, f)


def save_to_txt(status, outfile):
    # Create text file with status
    with open(outfile, "w") as f:
        f.write(status.upper())


def check_this_api(out_dir, api_name, mod):
    print("\nChecking status for '%s'" % api_name)

    if hasattr(mod, "api_server_check"):
        print(mod.api_server_check)

        label = "Data source '%s'" % api_name
        status = "down"
        if isAPIconnected(src=api_name, data=1):
            status = "up"

        print("status='%s'" % status)

        outfile = os.path.join(out_dir, "argopy_api_status_%s.json" % api_name)
        save_to_json(label, status, COLORS[status], outfile)

        outfile = os.path.join(out_dir, "%s.txt" % api_name.upper())
        save_to_txt(status, outfile)

    else:
        raise ValueError("Can't get status for this fetcher !")


def skip_this_api(out_dir, api_name):
    label = "Data source '%s'" % api_name
    status = "unknown"

    outfile = os.path.join(out_dir, "argopy_api_status_%s.json" % api_name)
    save_to_json(label, status, COLORS[status], outfile)

    outfile = os.path.join(out_dir, "%s.txt" % api_name.upper())
    save_to_txt(status, outfile)


def save_api_status(out_dir: str = "."):
    api_expected = ["erddap", "argovis", "gdac"]
    api_available = list_available_data_src()
    for api_name in api_expected:
        if api_name in api_available:
            check_this_api(out_dir, api_name, api_available[api_name])
        else:
            skip_this_api(out_dir, api_name)


def save_carbonfootprint(out_dir: str = "."):
    print("\nChecking carbon footprint")
    gc = GreenCoding()

    try:
        GreenCoding.shieldsio_endpoint(
            gc.footprint_since_last_release(errors="ignore"),
            label="Upcoming release footprint [gCO2eq]",
            outfile=Path(out_dir).joinpath("argopy_carbonfootprint_since_last_release.json"),
        )

        GreenCoding.shieldsio_endpoint(
            gc.footprint_baseline(errors="ignore"),
            label="Historical baseline footprint [gCO2eq]",
            outfile=Path(out_dir).joinpath("argopy_carbonfootprint_baseline.json"),
        )

        print("OK")
    except Exception:
        # This is probably because we hit a:
        # Error: 403, message='rate limit exceeded'
        print("ERROR: Can't update carbon footprint, probably because of a 'rate limit exceeded' with the Github API")


if __name__ == "__main__":
    save_api_status(".")
    save_carbonfootprint(".")