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

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 for f in features: 

29 yield f 

30 

31 if mc.relDepth >= mc.maxDepth: 

32 continue 

33 

34 sub = [] 

35 

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

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

38 sub.append(val) 

39 elif isinstance(val, list): 

40 for v in val: 

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

42 sub.append(v) 

43 

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

45 

46 

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

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

49 

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

51 

52 for k, v in fd.attributes: 

53 typ = str(type(v)) 

54 desc.columns.append(gws.ColumnDescription( 

55 name=k, 

56 nativeType=typ, 

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

58 )) 

59 if fd.shape: 

60 col = gws.ColumnDescription( 

61 name='geometry', 

62 geometryType=fd.shape.type, 

63 geometrySrid=fd.shape.crs.srid, 

64 type=gws.AttributeType.geometry, 

65 ) 

66 desc.columns.append(col) 

67 desc.geometryName = col.name 

68 desc.geometryType = col.geometryType 

69 desc.geometrySrid = col.geometrySrid 

70 

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

72 return desc 

73 

74 

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

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