Coverage for skema/program_analysis/JSON2GroMEt/json2gromet.py: 15%

62 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-04-30 17:15 +0000

1import json 

2import sys 

3import inspect 

4from typing import List, Dict, Any 

5from pathlib import Path 

6 

7import skema.gromet.fn 

8import skema.gromet.metadata 

9from skema.gromet.fn import GrometFNModuleCollection 

10 

11 

12def json_to_gromet(path: str) -> GrometFNModuleCollection: 

13 """Given the path to a Gromet JSON file as input, return the Python data-model representation. 

14 The return type is the top-level GrometFNModuleCollection object. 

15 """ 

16 

17 # gromet_fn_map: Mapping from gromet_type field to Python type object. Supports all Gromet FN classes. 

18 # gromet_metadata_map: Mapping from metadata_type field to Python type object. Supports most Gromet Metadata classes (all with metadata_type field). 

19 # gromet_field_map: Mapping from a tuple of class fields to a Python type object. Supports remaining metadata types. Requires unique field signature between types. 

20 # NOTE: These dicts map to a Python type object, not an instance of a type. 

21 # An instance can be created by calling the constructor on the type object: gromet_fn_map[gromet_type]() 

22 gromet_fn_map = {} 

23 gromet_metadata_map = {} 

24 gromet_field_map = {} 

25 

26 for fn_name, fn_object in inspect.getmembers( 

27 sys.modules["skema.gromet.fn"], inspect.isclass 

28 ): 

29 gromet_fn_map[fn_name] = fn_object 

30 

31 for metadata_name, metadata_object in inspect.getmembers( 

32 sys.modules["skema.gromet.metadata"], inspect.isclass 

33 ): 

34 instance = metadata_object() 

35 if "is_metadatum" in instance.attribute_map and instance.is_metadatum: 

36 gromet_metadata_map[metadata_name] = metadata_object 

37 else: 

38 gromet_fn_map[metadata_name] = metadata_object 

39 

40 def get_obj_type(obj: Dict) -> Any: 

41 """Given a dictionary representing a Gromet object (i.e. BoxFunction), return an instance of that object. 

42 Returns None if the dictionary does not represent a Gromet object. 

43 """ 

44 

45 # First check if we already have a mapping to a data-class memeber. All Gromet FN and most Gromet Metadata classes will fall into this category. 

46 # There are a few Gromet Metadata fields such as Provenance that do not have a "metadata_type" field 

47 if "gromet_type" in obj and ("is_metadatum" not in obj or obj["is_metadatum"] != True): 

48 return gromet_fn_map[obj["gromet_type"]]() 

49 elif obj["is_metadatum"]: 

50 return gromet_metadata_map[obj["gromet_type"]]() 

51 

52 # If there is not a mapping to an object, we will check the fields to see if they match an existing class in the data-model. 

53 # For example: (id, box, metadata) would map to GrometPort 

54 obj_fields = tuple(obj.keys()) 

55 if obj_fields in gromet_field_map: 

56 return gromet_field_map[obj_fields]() 

57 

58 for gromet_name, gromet_object in inspect.getmembers( 

59 sys.modules[__name__], inspect.isclass 

60 ): 

61 found = True 

62 for field, value in obj.items(): 

63 if not hasattr(gromet_object, field): 

64 found = False 

65 break 

66 if found: 

67 gromet_field_map[obj_fields] = gromet_object 

68 return gromet_object() 

69 

70 return None 

71 

72 def process_list(obj: List) -> List: 

73 """Handles importing a JSON list into the Gromet data-model""" 

74 

75 # NOTE: The reason this is a seperate function from process_object() is to handle the case of nested Lists. 

76 # This is something that can occur in a few places, for example the metadata_collection field. 

77 out = [] 

78 for element in obj: 

79 if isinstance(element, List): 

80 out.append(process_list(element)) 

81 elif isinstance(element, Dict): 

82 # A dictionary may or may not represent a Gromet object. 

83 # Some LiteralValue representations are dictionaries too. 

84 obj_type = get_obj_type(element) 

85 if obj_type: 

86 out.append(process_object(element, obj_type)) 

87 else: 

88 out.append(element) 

89 else: 

90 out.append(element) 

91 

92 return out 

93 

94 def process_object(obj: Dict, gromet_obj: Any): 

95 """Recursivly itterate over a Gromet JSON obj, importing the values into the fields of gromet_obj""" 

96 for field, value in obj.items(): 

97 if isinstance(value, List): 

98 setattr(gromet_obj, field, process_list(value)) 

99 elif isinstance(value, Dict): 

100 # Like the case in process_list, not all dicts will represent a Gromet object, so we check for that first. 

101 obj_type = get_obj_type(value) 

102 if obj_type: 

103 setattr(gromet_obj, field, process_object(value, obj_type)) 

104 else: 

105 setattr(gromet_obj, field, value) 

106 else: 

107 setattr(gromet_obj, field, value) 

108 return gromet_obj 

109 

110 json_object = json.loads(Path(path).read_text()) 

111 return process_object(json_object, get_obj_type(json_object))