Coverage for skema/gromet/execution_engine/types/indexable.py: 42%

57 statements  

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

1import numpy 

2from typing import Union, Dict, Any, Tuple, List 

3 

4from skema.gromet.execution_engine.types.defined_types import ( 

5 Field, 

6 Sequence, 

7 Indexable, 

8 Record, 

9 RecordField, 

10) 

11 

12 

13class Indexable_get(object): 

14 source_language_name = {"CAST": "_get"} 

15 inputs = [Field("indexable_input", "Indexable"), Field("index", "Any")] 

16 outputs = [Field("element", "Any")] 

17 shorthand = "_get" 

18 documentation = "" 

19 

20 def exec(indexable_input: Indexable, index: Any) -> Any: 

21 if isinstance(indexable_input, Record): 

22 return Indexable_get.Record_get(indexable_input, index) 

23 else: 

24 return indexable_input[index] 

25 

26 def Record_get(record_input: Record, field_name: str) -> RecordField: 

27 for field in record_input.fields: 

28 if field.name == field_name: 

29 return field 

30 

31 

32class Indexable_set(object): 

33 source_language_name = {"CAST": "_set"} 

34 inputs = [ 

35 Field("indexable_input", "Indexable"), 

36 Field("index", "Any"), 

37 Field("value", "Any"), 

38 ] 

39 outputs = [Field("indexable_output", "Indexable")] 

40 shorthand = "_set" 

41 documentation = "" 

42 

43 def exec(indexable_input: Indexable, index: Any, value: Any) -> Indexable: 

44 if isinstance(indexable_input, Sequence): 

45 return Indexable_set.Sequence_set(indexable_input, index, value) 

46 elif isinstance(indexable_input, Dict): 

47 return Indexable_set.Map_set(indexable_input, index, value) 

48 elif isinstance(indexable_input, Record): 

49 return Indexable_set.Record_set(indexable_input, index, value) 

50 

51 def Sequence_set(sequence_input: Sequence, index: int, value: Any) -> Any: 

52 if isinstance(sequence_input, range): 

53 # TODO: You cannot assign to a range object 

54 pass 

55 elif isinstance(sequence_input, List): 

56 return Indexable_set.List_set(sequence_input, index, value) 

57 elif isinstance(sequence_input, numpy.ndarray): 

58 return Indexable_set.Array_set(sequence_input, index, value) 

59 elif isinstance(sequence_input, Tuple): 

60 return Indexable_set.Tuple_set(sequence_input, index, value) 

61 

62 def List_set(list_input: List, index: int, value: Any) -> List: 

63 list_input[index] = value 

64 return list_input 

65 

66 def Array_set( 

67 array_input: numpy.ndarray, index: int, value: Any 

68 ) -> numpy.ndarray: 

69 array_input[index] = value 

70 return array_input 

71 

72 def Tuple_set(tuple_input: Tuple, index: int, value: Any) -> Tuple: 

73 converted_tuple = list(tuple_input) 

74 converted_tuple[index] = value 

75 return tuple(converted_tuple) 

76 

77 def Map_set(map_input: Dict, index: Any, value: Any) -> Dict: 

78 map_input[index] = value 

79 return map_input 

80 

81 def Record_set( 

82 record_input: Record, field_name: str, value: Any 

83 ) -> Record: # TODO: Double check this implementation, how to copy Record type 

84 for field in record_input.fields: 

85 if field.name == field_name: 

86 field.value = value # TODO: Do we need type checking here? 

87 return record_input