Coverage for src/CSET/cset_workflow/app/finish_website/bin/finish_website.py: 100%

1#!/usr/bin/env python3 

2# © Crown copyright, Met Office (2022-2025) and CSET contributors. 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15 

16""" 

17Write finished status to website front page. 

18 

19Constructs the plot index, and does the final update to the workflow status on 

20the front page of the web interface. 

21""" 

22 

23import datetime 

24import json 

25import logging 

26import os 

27import shutil 

28from importlib.metadata import version 

29from pathlib import Path 

30 

31from CSET._common import combine_dicts, sort_dict 

32 

33logging.basicConfig( 

34 level=os.getenv("LOGLEVEL", "INFO"), format="%(asctime)s %(levelname)s %(message)s" 

35) 

36 

37 

38def construct_index(): 

39 """Construct the plot index. 

40 

41 Index should has the form ``{"Category Name": {"recipe_id": "Plot Name"}}`` 

42 where ``recipe_id`` is the name of the plot's directory. 

43 """ 

44 index = {} 

45 plots_dir = Path(os.environ["CYLC_WORKFLOW_SHARE_DIR"]) / "web/plots" 

46 # Loop over all diagnostics and append to index. 

47 for metadata_file in plots_dir.glob("**/*/meta.json"): 

48 try: 

49 with open(metadata_file, "rt", encoding="UTF-8") as fp: 

50 plot_metadata = json.load(fp) 

51 

52 category = plot_metadata["category"] 

53 case_date = plot_metadata.get("case_date", "") 

54 relative_url = str(metadata_file.parent.relative_to(plots_dir)) 

55 

56 record = { 

57 category: { 

58 case_date if case_date else "Aggregation": { 

59 relative_url: plot_metadata["title"].strip() 

60 } 

61 } 

62 } 

63 except (json.JSONDecodeError, KeyError, TypeError) as err: 

64 logging.error("%s is invalid, skipping.\n%s", metadata_file, err) 

65 continue 

66 index = combine_dicts(index, record) 

67 

68 # Sort index of diagnostics. 

69 index = sort_dict(index) 

70 

71 # Write out website index. 

72 with open(plots_dir / "index.json", "wt", encoding="UTF-8") as fp: 

73 json.dump(index, fp, indent=2) 

74 

75 

76def update_workflow_status(): 

77 """Update the workflow status on the front page of the web interface.""" 

78 web_dir = Path(os.environ["CYLC_WORKFLOW_SHARE_DIR"] + "/web") 

79 with open(web_dir / "status.html", "wt") as fp: 

80 finish_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") 

81 fp.write(f"<p>Completed at {finish_time} using CSET v{version('CSET')}</p>\n") 

82 

83 

84def copy_rose_config(): 

85 """Copy the rose-suite.conf file to add to output web directory.""" 

86 rose_suite_conf = Path(os.environ["CYLC_WORKFLOW_RUN_DIR"]) / "rose-suite.conf" 

87 web_conf_file = Path(os.environ["CYLC_WORKFLOW_SHARE_DIR"]) / "web/rose-suite.conf" 

88 shutil.copy(rose_suite_conf, web_conf_file) 

89 

90 

91def run(): 

92 """Do the final steps to finish the website.""" 

93 construct_index() 

94 update_workflow_status() 

95 copy_rose_config() 

96 

97 

98if __name__ == "__main__": # pragma: no cover 

99 run()