Coverage for gws-app / gws / gis / mpx / __init__.py: 34%

1"""MapProxy integration module for GWS. 

2 

3This module provides functions to interact with MapProxy services for WMS and WMTS requests. 

4""" 

5 

6import time 

7from typing import Any, Dict, Optional, Union 

8import os 

9 

10import gws 

11import gws.config 

12import gws.lib.net 

13 

14 

15class ServiceException(Exception): 

16 """Exception raised when a MapProxy service request fails.""" 

17 pass 

18 

19 

20def _call(service: str, params: Dict[str, Any]) -> bytes: 

21 """Make a direct call to a MapProxy service. 

22 

23 Args: 

24 service: The service endpoint name. 

25 params: The parameters to send with the request. 

26 

27 Returns: 

28 The image content as bytes if successful. 

29 

30 Raises: 

31 ServiceException: If the response is not an image. 

32 """ 

33 url = getattr(gws.config.get_root().app, 'mpxUrl') + '/' + service 

34 resp = gws.lib.net.http_request(url, params=params) 

35 if resp.content_type.startswith('image'): 

36 return resp.content 

37 raise ServiceException(resp.text) 

38 

39 

40_retry_count = 3 

41_retry_pause = 5 

42_request_number = 0 

43 

44 

45def _call_with_retry(service: str, params: Dict[str, Any]) -> Optional[bytes]: 

46 """Call a MapProxy service with retry logic. 

47 

48 Makes multiple attempts to call the service if initial attempts fail. 

49 

50 Args: 

51 service: The service endpoint name. 

52 params: The parameters to send with the request. 

53 

54 Returns: 

55 The image content as bytes if successful, None if all retries fail. 

56 """ 

57 global _request_number 

58 

59 _request_number += 1 

60 rc = 0 

61 

62 while True: 

63 try: 

64 return _call(service, params) 

65 except Exception as exc: 

66 gws.log.exception() 

67 err = repr(exc) 

68 

69 if rc >= _retry_count: 

70 gws.log.error(f'MAPPROXY_ERROR: {_request_number}/{rc} FAILED error={err} {params!r}') 

71 return None 

72 

73 gws.log.error(f'MAPPROXY_ERROR: {_request_number}/{rc} retry error={err}') 

74 time.sleep(_retry_pause) 

75 rc += 1 

76 

77 

78def wms_request(layer_uid: str, bounds: gws.Bounds, width: int, height: int, 

79 forward: Optional[Dict[str, Any]] = None) -> Optional[bytes]: 

80 """Make a WMS GetMap request to MapProxy. 

81 

82 Args: 

83 layer_uid: The layer identifier. 

84 bounds: The bounding box for the map. 

85 width: The width of the requested image in pixels. 

86 height: The height of the requested image in pixels. 

87 forward: Additional parameters to forward to the WMS request. 

88 

89 Returns: 

90 The image content as bytes if successful, None if the request fails. 

91 """ 

92 mpx_no_transparency = os.getenv('GWS_MPX_NO_TRANSPARENCY', '0') == '1' 

93 ext = bounds.extent 

94 if bounds.crs.isYX: 

95 ext = (ext[1], ext[0], ext[3], ext[2]) 

96 params = { 

97 'bbox': ext, 

98 'width': width, 

99 'height': height, 

100 'crs': bounds.crs.epsg, 

101 'service': 'WMS', 

102 'request': 'GetMap', 

103 'version': '1.3.0', 

104 'format': 'image/png', 

105 'transparent': 'false' if mpx_no_transparency else 'true', 

106 'styles': '', 

107 'layers': layer_uid 

108 } 

109 if forward: 

110 params.update(forward) 

111 return _call_with_retry('wms', params) 

112 

113 

114def wmts_request(source_uid: str, x: int, y: int, z: int, 

115 tile_matrix: str, tile_size: int) -> Optional[bytes]: 

116 """Make a WMTS GetTile request to MapProxy. 

117 

118 Args: 

119 source_uid: The source layer identifier. 

120 x: The tile column. 

121 y: The tile row. 

122 z: The zoom level. 

123 tile_matrix: The tile matrix set identifier. 

124 tile_size: The size of the tile in pixels. 

125 

126 Returns: 

127 The tile image content as bytes if successful, None if the request fails. 

128 """ 

129 params = { 

130 'tilecol': x, 

131 'tilerow': y, 

132 'tilematrix': z, 

133 'service': 'WMTS', 

134 'request': 'GetTile', 

135 'version': '1.0.0', 

136 'format': 'image/png', 

137 'tilematrixset': tile_matrix, 

138 'style': 'default', 

139 'layer': source_uid 

140 } 

141 return _call_with_retry('ows', params)