Coverage for src/CSET/cset_workflow/lib/python/jinja_utils.py: 100%

17 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-09-05 21:08 +0000

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

2# 

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

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

5# You may obtain a copy of the License at 

6# 

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

8# 

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

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

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

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

13# limitations under the License. 

14 

15"""Useful functions for the workflow.""" 

16 

17import base64 

18import json 

19 

20 

21def get_models(rose_variables: dict) -> list[dict]: 

22 """Load per-model configuration into a single object. 

23 

24 Returns a list of dictionaries, each one containing a per-model 

25 configuration. 

26 """ 

27 models = [] 

28 for model in range(1, 20): 

29 model_prefix = f"m{model}_" 

30 model_vars = { 

31 key.removeprefix(model_prefix): value 

32 for key, value in rose_variables.items() 

33 if key.startswith(model_prefix) 

34 } 

35 if model_vars: 

36 model_vars["id"] = model 

37 models.append(model_vars) 

38 return models 

39 

40 

41def b64json(d: dict | list) -> str: 

42 """Encode an object as base64 encoded JSON for transport though cylc.""" 

43 new_d = d.copy() 

44 if isinstance(new_d, dict): 

45 # Remove circular reference to ROSE_SUITE_VARIABLES. 

46 new_d.pop("ROSE_SUITE_VARIABLES", None) 

47 output = base64.b64encode(json.dumps(new_d).encode()).decode() 

48 return output