Coverage for brodata / epc.py: 71%

1import logging 

2from functools import partial 

3 

4import pandas as pd 

5 

6from . import bro 

7 

8logger = logging.getLogger(__name__) 

9 

10 

11class ExplorationProductionConstruction(bro.FileOrUrl): 

12 """Class to represent an Exploration Production Construction (EPC) from the BRO.""" 

13 

14 _rest_url = "https://publiek.broservices.nl/ep/epc/v1" 

15 _xmlns = "http://www.broservices.nl/xsd/dsepc/1.0" 

16 _char = "EPC_C" 

17 

18 def _read_contents(self, tree): 

19 ns = { 

20 "brocom": "http://www.broservices.nl/xsd/brocommon/3.0", 

21 "gml": "http://www.opengis.net/gml/3.2", 

22 "epccommon": "http://www.broservices.nl/xsd/epccommon/1.0", 

23 "xmlns": self._xmlns, 

24 } 

25 

26 # Try to find EPC_PO_Borehole, EPC_PO_DP_Borehole or EPC_PPO_Borehole 

27 object_names = ["EPC_PO_Borehole", "EPC_PO_DP_Borehole", "EPC_PPO_Borehole"] 

28 epc = self._get_main_object(tree, object_names, ns) 

29 

30 # Parse attributes 

31 for key in epc.attrib: 

32 setattr(self, key.split("}", 1)[1], epc.attrib[key]) 

33 

34 # Parse child elements 

35 for child in epc: 

36 key = self._get_tag(child) 

37 

38 if len(child) == 0: 

39 # Leaf elements (text content) 

40 setattr(self, key, child.text) 

41 elif key == "standardizedLocation": 

42 self._read_standardized_location(child) 

43 elif key == "deliveredVerticalPosition": 

44 self._read_delivered_vertical_position(child) 

45 elif key == "location": 

46 setattr(self, key, self._read_geometry(child)) 

47 elif key == "owner": 

48 self._read_owner(child) 

49 elif key == "sourceReference": 

50 self._read_source_reference(child) 

51 elif key in ["registrationHistory", "reportHistory"]: 

52 self._read_children_of_children(child) 

53 elif key == "lifespan": 

54 self._read_lifespan(child) 

55 elif key == "constructionHistory": 

56 self._read_construction_history(child) 

57 elif key == "horizontalPositioning": 

58 self._read_horizontal_positioning(child) 

59 elif key == "boreholeSegment": 

60 if self._check_single_child_with_tag(child, "BoreholeSegment"): 

61 child = child[0] 

62 self._read_borehole_segment(child) 

63 elif key == "explorationProductionLicence": 

64 self._read_exploration_production_licence(child) 

65 elif key == "deliveryContext": 

66 setattr(self, key, child.text) 

67 elif key == "legalStatus": 

68 setattr(self, key, child.text) 

69 else: 

70 self._warn_unknown_tag(key) 

71 

72 if hasattr(self, "boreholeSegment") and self.boreholeSegment: 

73 setattr(self, "boreholeSegment", pd.DataFrame(self.boreholeSegment)) 

74 

75 def _read_owner(self, node): 

76 """Parse owner element with optional chamber of commerce or registration number.""" 

77 for child in node: 

78 key = self._get_tag(child) 

79 if key in ["chamberOfCommerceNumber", "europeanCompanyRegistrationNumber"]: 

80 setattr(self, f"owner_{key}", child.text) 

81 

82 def _read_source_reference(self, node): 

83 """Parse sourceReference element.""" 

84 for child in node: 

85 key = self._get_tag(child) 

86 if key in ["chamberOfCommerceNumber", "europeanCompanyRegistrationNumber"]: 

87 setattr(self, f"sourceReference_{key}", child.text) 

88 

89 def _read_construction_history(self, node): 

90 """Parse constructionHistory element with multiple events.""" 

91 events = [] 

92 for child in node: 

93 key = self._get_tag(child) 

94 if key == "event": 

95 event = {} 

96 for grandchild in child: 

97 key2 = self._get_tag(grandchild) 

98 if key2 == "date": 

99 event["date"] = self._read_date(grandchild) 

100 elif key2 == "name": 

101 event["name"] = grandchild.text 

102 elif key2 == "identifier": 

103 event["identifier"] = grandchild.text 

104 else: 

105 self._warn_unknown_tag(key2) 

106 events.append(event) 

107 

108 if events: 

109 setattr(self, "constructionHistory", pd.DataFrame(events)) 

110 

111 def _read_horizontal_positioning(self, node): 

112 """Parse horizontalPositioning element.""" 

113 for child in node: 

114 key = self._get_tag(child) 

115 if key in ["horizontalPositioningDate"]: 

116 setattr(self, key, self._read_date(child)) 

117 elif key == "horizontalPositioningMethod": 

118 setattr(self, key, child.text) 

119 elif key == "horizontalPositioningOperator": 

120 self._read_operator(child) 

121 else: 

122 self._warn_unknown_tag(key) 

123 

124 def _read_borehole_segment(self, node): 

125 """Parse boreholeSegment element with BoreholeSegment wrapper.""" 

126 if not hasattr(self, "boreholeSegment"): 

127 self.boreholeSegment = [] 

128 segment = {} 

129 for child in node: 

130 key = self._get_tag(child) 

131 if key in [ 

132 "boreholeSegmentCode", 

133 "boreholeSegmentName", 

134 "boreholeSegmentCategory", 

135 "purpose", 

136 "drillingStartDate", 

137 "drillingEndDate", 

138 "dateOfDisclosure", 

139 "boreholeSegmentLocation", 

140 ]: 

141 segment[key] = child.text 

142 else: 

143 self._warn_unknown_tag(key) 

144 self.boreholeSegment.append(segment) 

145 

146 def _read_exploration_production_licence(self, node): 

147 """Parse explorationProductionLicence element.""" 

148 for child in node: 

149 key = self._get_tag(child) 

150 if key == "ExplorationProductionLicence": 

151 for grandchild in child: 

152 key2 = self._get_tag(grandchild) 

153 if key2 == "broId": 

154 setattr( 

155 self, "explorationProductionLicence_broId", grandchild.text 

156 ) 

157 else: 

158 self._warn_unknown_tag(key2) 

159 

160 

161cl = ExplorationProductionConstruction 

162 

163get_bro_ids_of_bronhouder = partial(bro._get_bro_ids_of_bronhouder, cl) 

164get_bro_ids_of_bronhouder.__doc__ = bro._get_bro_ids_of_bronhouder.__doc__ 

165 

166get_data_for_bro_ids = partial(bro._get_data_for_bro_ids, cl) 

167get_data_for_bro_ids.__doc__ = bro._get_data_for_bro_ids.__doc__ 

168 

169get_characteristics = partial(bro._get_characteristics, cl) 

170get_characteristics.__doc__ = bro._get_characteristics.__doc__ 

171 

172get_data_in_extent = partial(bro._get_data_in_extent, cl) 

173get_data_in_extent.__doc__ = bro._get_data_in_extent.__doc__