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

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""" 

17Create the CSET diagnostic viewing website. 

18 

19Copies the static files that make up the web interface, constructs the plot 

20index, and updates the workflow status on the front page of the 

21web interface. 

22""" 

23 

24import argparse 

25import json 

26import logging 

27import os 

28import re 

29import shutil 

30import sys 

31import time 

32from collections.abc import Iterable 

33from importlib.metadata import version 

34from pathlib import Path 

35 

36logging.basicConfig( 

37 level=os.getenv("LOGLEVEL", "INFO"), 

38 format="%(asctime)s %(levelname)s %(message)s", 

39 stream=sys.stdout, 

40) 

41logger = logging.getLogger(__name__) 

42 

43 

44def human_sorted(iterable: Iterable, reverse: bool = False) -> list: 

45 """Sort such numbers within strings are sorted correctly.""" 

46 # Adapted from https://nedbatchelder.com/blog/200712/human_sorting.html 

47 

48 def alphanum_key(s): 

49 """Turn a string into a list of string and number chunks. 

50 

51 >>> alphanum_key("z23a") 

52 ["z", 23, "a"] 

53 """ 

54 try: 

55 return [int(c) if c.isdecimal() else c for c in re.split(r"(\d+)", s)] 

56 except TypeError: 

57 return s 

58 

59 return sorted(iterable, key=alphanum_key, reverse=reverse) 

60 

61 

62def sort_dict(d: dict) -> dict: 

63 """Recursively sort a dictionary.""" 

64 # Thank you to https://stackoverflow.com/a/47882384 

65 return { 

66 k: sort_dict(v) if isinstance(v, dict) else v 

67 for k, v in human_sorted(d.items()) 

68 } 

69 

70 

71def install_website_skeleton( 

72 www_root_link: Path | None, www_content: Path, skeleton_dir: Path 

73): 

74 """Copy static website files and create symlink from web document root.""" 

75 logger.info("Installing website files to %s.", www_content) 

76 # Create directory for web content. 

77 www_content.mkdir(parents=True, exist_ok=True) 

78 # Copy static HTML/CSS/JS. 

79 shutil.copytree(skeleton_dir, www_content, dirs_exist_ok=True) 

80 # Setup symbolic link from web root. 

81 if www_root_link is not None: 81 ↛ exitline 81 didn't return from function 'install_website_skeleton' because the condition on line 81 was always true

82 logger.info("Linking %s to web content.", www_root_link) 

83 # Remove existing link to output ahead of creating new symlink. 

84 www_root_link.unlink(missing_ok=True) 

85 # Ensure parent directories of WEB_DIR exist. 

86 www_root_link.parent.mkdir(parents=True, exist_ok=True) 

87 # Create symbolic link to web directory. 

88 # NOTE: While good for space, it means `cylc clean` removes output. 

89 www_root_link.symlink_to(www_content) 

90 

91 

92def construct_index(www_content: Path): 

93 """Construct the plot index.""" 

94 with open(www_content / "index.jsonl", "wt", encoding="UTF-8") as index_fp: 

95 # Loop over all diagnostics and append to index. The glob is sorted to 

96 # ensure a consistent ordering. 

97 for metadata_file in sorted(www_content.glob("**/*.json")): 

98 try: 

99 with open(metadata_file, "rt", encoding="UTF-8") as plot_fp: 

100 plot_metadata = json.load(plot_fp) 

101 plot_metadata["path"] = str( 

102 metadata_file.parent.relative_to(www_content) 

103 ) 

104 # Remove keys that are not useful for the index. 

105 removed_index_keys = [ 

106 "description", 

107 "plot_resolution", 

108 "plots", 

109 "skip_write", 

110 "COORD_LIST", 

111 "ONE_TO_ONE", 

112 "PERCENTILES", 

113 "SUBAREA_EXTENT", 

114 "SUBAREA_TYPE", 

115 ] 

116 for key in removed_index_keys: 

117 plot_metadata.pop(key, None) 

118 # Sort plot metadata for consistency. 

119 plot_metadata = sort_dict(plot_metadata) 

120 # Write metadata into website index. 

121 json.dump(plot_metadata, index_fp, separators=(",", ":")) 

122 index_fp.write("\n") 

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

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

125 continue 

126 

127 

128def bust_cache(www_content: Path): 

129 """Add a unique query string to static requests to avoid stale caches. 

130 

131 We only need to do this for static resources referenced from the index page, 

132 as each plot already uses a unique filename based on the recipe. 

133 """ 

134 # Search and replace the string "CACHEBUSTER" with a time-based cache key. 

135 cache_key = str(int(time.time())) 

136 with open(www_content / "index.html", "r+t") as fp: 

137 content = fp.read() 

138 new_content = content.replace("CACHEBUSTER", cache_key) 

139 fp.seek(0) 

140 fp.truncate() 

141 fp.write(new_content) 

142 

143 

144def update_workflow_status(www_content: Path): 

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

146 with open(www_content / "placeholder.html", "r+t") as fp: 

147 content = fp.read() 

148 finish_time = time.strftime("%Y-%m-%d %H:%M", time.localtime()) 

149 status = f"Completed at {finish_time} using CSET v{version('CSET')}" 

150 new_content = content.replace( 

151 '<p id="workflow-status">Unknown</p>', 

152 f'<p id="workflow-status">{status}</p>', 

153 ) 

154 fp.seek(0) 

155 fp.truncate() 

156 fp.write(new_content) 

157 

158 

159def copy_rose_config(www_content: Path): 

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

161 cylc_run_dir = os.getenv("CYLC_WORKFLOW_RUN_DIR", None) 

162 if cylc_run_dir: 162 ↛ exitline 162 didn't return from function 'copy_rose_config' because the condition on line 162 was always true

163 rose_suite_conf = Path(cylc_run_dir) / "rose-suite.conf" 

164 web_conf_file = www_content / "rose-suite.conf" 

165 try: 

166 shutil.copyfile(rose_suite_conf, web_conf_file) 

167 except FileNotFoundError: 

168 logger.warning("No rose-suite.conf file found for cylc workflow.") 

169 

170 

171def parse_args(args: list[str] | None = None) -> argparse.Namespace: 

172 """Parse command line arguments.""" 

173 parser = argparse.ArgumentParser( 

174 description="Create a navigation interface for a set of diagnostic webpages.", 

175 ) 

176 parser.add_argument( 

177 "web_content", 

178 type=Path, 

179 help="Where to save the HTML content and find the diagnostic webpages.", 

180 ) 

181 parser.add_argument( 

182 "--web-root-link", 

183 type=Path, 

184 help="Where to create a symlink to the content, optional.", 

185 ) 

186 parser.add_argument( 

187 "--skeleton", 

188 type=Path, 

189 help="Directory containing static website files. Defaults to $PWD/html", 

190 default=Path.cwd() / "html", 

191 ) 

192 return parser.parse_args(args) 

193 

194 

195def run(): # pragma: no cover 

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

197 args = parse_args() 

198 logger.debug("Arguments: %s", args) 

199 

200 install_website_skeleton(args.web_root_link, args.web_content, args.skeleton) 

201 copy_rose_config(args.web_content) 

202 construct_index(args.web_content) 

203 bust_cache(args.web_content) 

204 update_workflow_status(args.web_content) 

205 

206 

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

208 run()