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

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 params = { 

94 'bbox': bounds.extent, 

95 'width': width, 

96 'height': height, 

97 'crs': bounds.crs.epsg, 

98 'service': 'WMS', 

99 'request': 'GetMap', 

100 'version': '1.3.0', 

101 'format': 'image/png', 

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

103 'styles': '', 

104 'layers': layer_uid 

105 } 

106 if forward: 

107 params.update(forward) 

108 return _call_with_retry('wms', params) 

109 

110 

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

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

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

114 

115 Args: 

116 source_uid: The source layer identifier. 

117 x: The tile column. 

118 y: The tile row. 

119 z: The zoom level. 

120 tile_matrix: The tile matrix set identifier. 

121 tile_size: The size of the tile in pixels. 

122 

123 Returns: 

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

125 """ 

126 params = { 

127 'tilecol': x, 

128 'tilerow': y, 

129 'tilematrix': z, 

130 'service': 'WMTS', 

131 'request': 'GetTile', 

132 'version': '1.0.0', 

133 'format': 'image/png', 

134 'tilematrixset': tile_matrix, 

135 'style': 'default', 

136 'layer': source_uid 

137 } 

138 return _call_with_retry('ows', params)