Coverage for mapology/db.py: 43.21%

1"""Database for the complete tree.""" 

2 

3import copy 

4import json 

5import pathlib 

6from typing import Collection, Dict, Mapping, no_type_check 

7 

8from mapology import ENCODING, FS_DB_ROOT_PATH 

9 

10FS_DB_STORE_PART = 'prefix-store' 

11FS_DB_TABLE_PART = 'prefix-table' 

12FS_DB_HULLS_PART = 'prefix-hulls' 

13FS_DB_APT_SEARCH_PART = 'prefix-apt-search' 

14FS_DB_REGION_COUNTRY_PART = 'prefix-country-region' 

15 

16DB_ROOT = pathlib.Path(FS_DB_ROOT_PATH) 

17DB_FOLDER_PATHS = { 

18 'hulls': DB_ROOT / FS_DB_HULLS_PART, 

19 'store': DB_ROOT / FS_DB_STORE_PART, 

20 'table': DB_ROOT / FS_DB_TABLE_PART, 

21 'apt_search': DB_ROOT / FS_DB_APT_SEARCH_PART, 

22 'region_country': DB_ROOT / FS_DB_REGION_COUNTRY_PART, 

23} 

24 

25DB_INDEX_PATHS = { 

26 'hulls': DB_ROOT / f'{FS_DB_HULLS_PART}.json', 

27 'store': DB_ROOT / f'{FS_DB_STORE_PART}.json', 

28 'table': DB_ROOT / f'{FS_DB_TABLE_PART}.json', 

29 'apt_search': DB_ROOT / f'{FS_DB_APT_SEARCH_PART}.json', 

30 'region_country': DB_ROOT / f'{FS_DB_REGION_COUNTRY_PART}.json', 

31} 

32PHeaderDict = Dict[str, Collection[str]] 

33CC_HINT = 'CC_HINT' 

34IC_PREFIX = 'IC_PREFIX' 

35GEO_JSON_PREFIX_HEADER: PHeaderDict = { 

36 'type': 'FeatureCollection', 

37 'id': f'{IC_PREFIX}', 

38 'name': f'Region - {IC_PREFIX} ({CC_HINT})', 

39 'crs': { 

40 'type': 'name', 

41 'properties': { 

42 'name': 'urn:ogc:def:crs:OGC:1.3:CRS84', 

43 }, 

44 }, 

45 'features': [], 

46} 

47JSON_PREFIX_TABLE_HEADER = { 

48 'type': 'x-prefix-table', 

49 'id': f'{IC_PREFIX}', 

50 'name': f'Region - {IC_PREFIX} ({CC_HINT})', 

51 'crs': { 

52 'type': 'name', 

53 'properties': { 

54 'name': 'urn:ogc:def:crs:OGC:1.3:CRS84', 

55 }, 

56 }, 

57 'airports': [], 

58} 

59 

60JSON_PREFIX_APT_SEARCH_HEADER = [] # type: ignore 

61JSON_PREFIX_COUNTRY_HEADER = {} # type: ignore 

62 

63 

64def ensure_fs_tree() -> None: 

65 """Ensure the DB folder tree exists.""" 

66 for db in DB_FOLDER_PATHS.values(): 

67 db.mkdir(parents=True, exist_ok=True) 

68 

69 for index in DB_INDEX_PATHS.values(): 

70 if not index.exists(): 

71 with open(index, 'wt', encoding=ENCODING) as handle: 

72 json.dump({}, handle, indent=2) 

73 

74 

75@no_type_check 

76def load_index(kind: str) -> Mapping[str, str]: 

77 """DRY.""" 

78 with open(DB_INDEX_PATHS[kind], 'rt', encoding=ENCODING) as handle: 

79 return json.load(handle) 

80 

81 

82def dump_index(kind: str, data: Mapping[str, str]) -> None: 

83 """DRY.""" 

84 with open(DB_INDEX_PATHS[kind], 'wt', encoding=ENCODING) as handle: 

85 json.dump(data, handle, indent=2) 

86 

87 

88def hull_path(some_prefix: str) -> str: 

89 """DRY.""" 

90 return str(DB_FOLDER_PATHS['hulls'] / f'{some_prefix}.json') 

91 

92 

93def apt_search_path(some_prefix: str) -> str: 

94 """DRY.""" 

95 return str(DB_FOLDER_PATHS['apt_search'] / f'{some_prefix}.json') 

96 

97 

98def region_country_path(some_prefix: str) -> str: 

99 """DRY.""" 

100 return str(DB_FOLDER_PATHS['region_country'] / f'{some_prefix}.json') 

101 

102 

103def add_prefix(icp: str, cc: str) -> PHeaderDict: 

104 """DRY.""" 

105 geojson = copy.deepcopy(GEO_JSON_PREFIX_HEADER) 

106 geojson['name'] = geojson['name'].replace(IC_PREFIX, icp).replace(CC_HINT, cc) # type: ignore 

107 geojson['id'] = geojson['id'].replace(IC_PREFIX, icp) # type: ignore 

108 return geojson 

109 

110 

111def add_table_prefix(icp: str, cc: str) -> PHeaderDict: 

112 """DRY.""" 

113 table = copy.deepcopy(JSON_PREFIX_TABLE_HEADER) 

114 table['name'] = table['name'].replace(IC_PREFIX, icp).replace(CC_HINT, cc) # type: ignore 

115 table['id'] = table['id'].replace(IC_PREFIX, icp) # type: ignore 

116 return table 

117 

118 

119@no_type_check 

120def update_aspect(index_db: Mapping[str, object], a_prefix: str, a_cc: str, kind: str) -> Mapping[str, object]: 

121 """Process the kinds' index and ensure prefix aspect db is present""" 

122 if a_prefix not in index_db: 

123 index_db[a_prefix] = str(DB_FOLDER_PATHS[kind] / f'{a_prefix}.json') # noqa 

124 # Create initial kinds' store data entry for ICAO prefix 

125 factory = add_prefix if kind == 'store' else add_table_prefix 

126 with open(index_db[a_prefix], 'wt', encoding=ENCODING) as handle: # noqa 

127 json.dump(factory(a_prefix, a_cc), handle) 

128 with open(index_db[a_prefix], 'rt', encoding=ENCODING) as handle: # noqa 

129 return json.load(handle)