Coverage for gws-app/gws/base/model/util.py: 54%

1"""Model utilities.""" 

2 

3from typing import Iterable 

4import datetime 

5 

6import gws 

7import gws.base.feature 

8 

9_ATTR_TO_PY = { 

10 gws.AttributeType.bool: bool, 

11 gws.AttributeType.bytes: bytes, 

12 gws.AttributeType.date: datetime.date, 

13 gws.AttributeType.datetime: datetime.datetime, 

14 # gws.AttributeType.feature 

15 # gws.AttributeType.featurelist 

16 gws.AttributeType.float: float, 

17 # gws.AttributeType.floatlist 

18 # gws.AttributeType.geometry 

19 gws.AttributeType.int: int, 

20 # gws.AttributeType.intlist 

21 gws.AttributeType.str: str, 

22 # gws.AttributeType.strlist 

23 gws.AttributeType.time: datetime.time, 

24} 

25 

26 

27def iter_features(features: list[gws.Feature], mc: gws.ModelContext) -> Iterable[gws.Feature]: 

28 """Iterate over features and their related features. 

29 

30 Related features are features stored in attributes, directly or in lists. They are 

31 visited depth-first, as long as the relation depth of the context is below its maximum. 

32 

33 Args: 

34 features: The features. 

35 mc: The model context. 

36 

37 Yields: 

38 Each feature, followed by its related features. 

39 """ 

40 for f in features: 

41 yield f 

42 

43 if mc.relDepth >= mc.maxDepth: 

44 continue 

45 

46 sub = [] 

47 

48 for val in f.attributes.values(): 

49 if isinstance(val, gws.base.feature.Feature): 

50 sub.append(val) 

51 elif isinstance(val, list): 

52 for v in val: 

53 if isinstance(v, gws.base.feature.Feature): 

54 sub.append(v) 

55 

56 yield from iter_features(sub, secondary_context(mc)) 

57 

58 

59def describe_from_record(fd: gws.FeatureRecord) -> gws.DataSetDescription: 

60 """Create a dataset description from a feature record. 

61 

62 Column types are derived from the python types of the attribute values. If the record 

63 has a shape, a ``geometry`` column is added. 

64 

65 Args: 

66 fd: The feature record. 

67 

68 Returns: 

69 The dataset description. 

70 """ 

71 py_to_attr = {str(v): k for k, v in _ATTR_TO_PY.items()} 

72 

73 desc = gws.DataSetDescription(columns=[]) 

74 

75 for k, v in fd.attributes: 

76 typ = str(type(v)) 

77 desc.columns.append(gws.ColumnDescription( 

78 name=k, 

79 nativeType=typ, 

80 type=py_to_attr.get(typ, gws.AttributeType.str), 

81 )) 

82 if fd.shape: 

83 col = gws.ColumnDescription( 

84 name='geometry', 

85 geometryType=fd.shape.type, 

86 geometrySrid=fd.shape.crs.srid, 

87 type=gws.AttributeType.geometry, 

88 ) 

89 desc.columns.append(col) 

90 desc.geometryName = col.name 

91 desc.geometryType = col.geometryType 

92 desc.geometrySrid = col.geometrySrid 

93 

94 desc.columnMap = {col.name: col for col in desc.columns} 

95 return desc 

96 

97 

98def copy_context(mc: gws.ModelContext, **kwargs) -> gws.ModelContext: 

99 """Copy a model context. 

100 

101 Args: 

102 mc: The model context. 

103 **kwargs: Properties to set in the copy. 

104 

105 Returns: 

106 A new model context. 

107 """ 

108 return gws.ModelContext(gws.u.merge(mc, kwargs)) 

109 

110def secondary_context(mc: gws.ModelContext, **kwargs) -> gws.ModelContext: 

111 """Copy a model context for related features, with the relation depth increased by one. 

112 

113 Args: 

114 mc: The model context. 

115 **kwargs: Properties to set in the copy. 

116 

117 Returns: 

118 A new model context. 

119 """ 

120 return gws.ModelContext(gws.u.merge(mc, kwargs, relDepth=mc.relDepth + 1))