Coverage for src/backend/InvenTree/plugin/builtin/exporter/bom_exporter.py: 17%

1"""Multi-level BOM exporter plugin.""" 

2 

3from decimal import Decimal 

4from typing import Optional 

5 

6from django.utils.translation import gettext_lazy as _ 

7 

8import rest_framework.serializers as serializers 

9 

10from InvenTree.helpers import normalize 

11from part.models import BomItem 

12from part.serializers import BomItemSerializer 

13from plugin import InvenTreePlugin 

14from plugin.mixins import DataExportMixin 

15 

16 

17class BomExporterOptionsSerializer(serializers.Serializer): 

18 """Custom export options for the BOM exporter plugin.""" 

19 

20 export_levels = serializers.IntegerField( 

21 default=0, 

22 label=_('Levels'), 

23 help_text=_( 

24 'Number of levels to export - set to zero to export all BOM levels' 

25 ), 

26 min_value=0, 

27 ) 

28 

29 export_total_quantity = serializers.BooleanField( 

30 default=True, 

31 label=_('Total Quantity'), 

32 help_text=_('Include total quantity of each part in the BOM'), 

33 ) 

34 

35 export_stock_data = serializers.BooleanField( 

36 default=True, label=_('Stock Data'), help_text=_('Include part stock data') 

37 ) 

38 

39 export_pricing_data = serializers.BooleanField( 

40 default=True, label=_('Pricing Data'), help_text=_('Include part pricing data') 

41 ) 

42 

43 export_supplier_data = serializers.BooleanField( 

44 default=True, label=_('Supplier Data'), help_text=_('Include supplier data') 

45 ) 

46 

47 export_manufacturer_data = serializers.BooleanField( 

48 default=True, 

49 label=_('Manufacturer Data'), 

50 help_text=_('Include manufacturer data'), 

51 ) 

52 

53 export_substitute_data = serializers.BooleanField( 

54 default=True, 

55 label=_('Substitute Data'), 

56 help_text=_('Include substitute part data'), 

57 ) 

58 

59 export_parameter_data = serializers.BooleanField( 

60 default=True, 

61 label=_('Parameter Data'), 

62 help_text=_('Include part parameter data'), 

63 ) 

64 

65 

66class BomExporterPlugin(DataExportMixin, InvenTreePlugin): 

67 """Builtin plugin for performing multi-level BOM exports.""" 

68 

69 NAME = 'BOM Exporter' 

70 SLUG = 'bom-exporter' 

71 TITLE = _('Multi-Level BOM Exporter') 

72 DESCRIPTION = _('Provides support for exporting multi-level BOMs') 

73 VERSION = '1.1.0' 

74 AUTHOR = _('InvenTree contributors') 

75 

76 ExportOptionsSerializer = BomExporterOptionsSerializer 

77 

78 def supports_export(self, model_class: type, user, *args, **kwargs) -> bool: 

79 """This exported only supports the BomItem model.""" 

80 return ( 

81 model_class == BomItem 

82 and kwargs.get('serializer_class') == BomItemSerializer 

83 ) 

84 

85 def update_headers(self, headers, context, **kwargs): 

86 """Update headers for the BOM export.""" 

87 export_total_quantity = context.get('export_total_quantity', True) 

88 

89 if not self.export_stock_data: 

90 # Remove stock data from the headers 

91 for field in [ 

92 'available_stock', 

93 'available_substitute_stock', 

94 'available_variant_stock', 

95 'external_stock', 

96 'on_order', 

97 'building', 

98 'can_build', 

99 ]: 

100 headers.pop(field, None) 

101 

102 if not self.export_pricing_data: 

103 # Remove pricing data from the headers 

104 for field in [ 

105 'pricing_min', 

106 'pricing_max', 

107 'pricing_min_total', 

108 'pricing_max_total', 

109 'pricing_updated', 

110 ]: 

111 headers.pop(field, None) 

112 

113 # Append a "BOM Level" field 

114 headers['level'] = _('BOM Level') 

115 

116 if export_total_quantity: 

117 # Append a 'total quantity' field 

118 headers['total_quantity'] = _('Total Quantity') 

119 

120 # Append variant part columns 

121 if self.export_substitute_data and self.n_substitute_cols > 0: 

122 for idx in range(self.n_substitute_cols): 

123 n = idx + 1 

124 headers[f'substitute_{idx}'] = _(f'Substitute {n}') 

125 

126 # Append supplier part columns 

127 if self.export_supplier_data and self.n_supplier_cols > 0: 

128 for idx in range(self.n_supplier_cols): 

129 n = idx + 1 

130 headers[f'supplier_name_{idx}'] = _(f'Supplier {n}') 

131 headers[f'supplier_sku_{idx}'] = _(f'Supplier {n} SKU') 

132 headers[f'supplier_mpn_{idx}'] = _(f'Supplier {n} MPN') 

133 

134 # Append manufacturer part columns 

135 if self.export_manufacturer_data and self.n_manufacturer_cols > 0: 

136 for idx in range(self.n_manufacturer_cols): 

137 n = idx + 1 

138 headers[f'manufacturer_name_{idx}'] = _(f'Manufacturer {n}') 

139 headers[f'manufacturer_mpn_{idx}'] = _(f'Manufacturer {n} MPN') 

140 

141 # Append part parameter columns 

142 if self.export_parameter_data and len(self.parameters) > 0: 

143 for key, value in self.parameters.items(): 

144 headers[f'parameter_{key}'] = value 

145 

146 return headers 

147 

148 def prefetch_queryset(self, queryset): 

149 """Perform pre-fetch on the provided queryset.""" 

150 queryset = queryset.prefetch_related('sub_part') 

151 

152 if self.export_substitute_data: 

153 queryset = queryset.prefetch_related('substitutes') 

154 

155 if self.export_supplier_data: 

156 queryset = queryset.prefetch_related( 

157 'sub_part__supplier_parts', 

158 'sub_part__supplier_parts__supplier', 

159 'sub_part__supplier_parts__manufacturer_part', 

160 'sub_part__supplier_parts__manufacturer_part__manufacturer', 

161 ) 

162 

163 if self.export_manufacturer_data: 

164 queryset = queryset.prefetch_related( 

165 'sub_part__manufacturer_parts', 

166 'sub_part__manufacturer_parts__manufacturer', 

167 ) 

168 

169 if self.export_parameter_data: 

170 queryset = queryset.prefetch_related( 

171 'sub_part__parameters_list', 'sub_part__parameters_list__template' 

172 ) 

173 

174 return queryset 

175 

176 def export_data( 

177 self, queryset, serializer_class, headers, context, output, **kwargs 

178 ): 

179 """Export BOM data from the queryset.""" 

180 self.serializer_class = serializer_class 

181 

182 # Track how many extra columns we need 

183 self.n_substitute_cols = 0 

184 self.n_supplier_cols = 0 

185 self.n_manufacturer_cols = 0 

186 

187 # A dict of "Parameter ID" -> "Parameter Name" 

188 self.parameters = {} 

189 

190 # Extract the export options from the context (and cache for later) 

191 self.export_levels = context.get('export_levels', 1) 

192 self.export_stock_data = context.get('export_stock_data', True) 

193 self.export_pricing_data = context.get('export_pricing_data', True) 

194 self.export_supplier_data = context.get('export_supplier_data', True) 

195 self.export_manufacturer_data = context.get('export_manufacturer_data', True) 

196 self.export_substitute_data = context.get('export_substitute_data', True) 

197 self.export_parameter_data = context.get('export_parameter_data', True) 

198 self.export_total_quantity = context.get('export_total_quantity', True) 

199 

200 # Pre-fetch related data to reduce database queries 

201 queryset = self.prefetch_queryset(queryset) 

202 

203 self.bom_data = [] 

204 

205 # Run through each item in the queryset 

206 for bom_item in queryset: 

207 self.process_bom_row(bom_item, 1, **kwargs) 

208 

209 return self.bom_data 

210 

211 def process_bom_row( 

212 self, bom_item, level: int = 1, multiplier: Optional[Decimal] = None, **kwargs 

213 ) -> list: 

214 """Process a single BOM row. 

215 

216 Arguments: 

217 bom_item: The BomItem object to process 

218 level: The current level of export 

219 multiplier: The multiplier for the quantity (used for recursive calls) 

220 """ 

221 # Add this row to the output dataset 

222 row = self.serializer_class(bom_item, exporting=True).data 

223 row['level'] = level 

224 

225 if multiplier is None: 

226 multiplier = Decimal(1) 

227 

228 # Extend with additional data 

229 

230 if self.export_substitute_data: 

231 row.update(self.get_substitute_data(bom_item)) 

232 

233 if self.export_supplier_data: 

234 row.update(self.get_supplier_data(bom_item)) 

235 

236 if self.export_manufacturer_data: 

237 row.update(self.get_manufacturer_data(bom_item)) 

238 

239 if self.export_parameter_data: 

240 row.update(self.get_parameter_data(bom_item)) 

241 

242 if self.export_total_quantity: 

243 # Calculate the total quantity for this BOM item 

244 total_quantity = Decimal(bom_item.quantity) * multiplier 

245 row['total_quantity'] = normalize(total_quantity) 

246 

247 self.bom_data.append(row) 

248 

249 # If we have reached the maximum export level, return just this bom item 

250 if bom_item.sub_part.assembly and ( 

251 self.export_levels <= 0 or level < self.export_levels 

252 ): 

253 sub_items = bom_item.sub_part.get_bom_items() 

254 sub_items = self.prefetch_queryset(sub_items) 

255 sub_items = BomItemSerializer.annotate_queryset(sub_items) 

256 

257 for item in sub_items.all(): 

258 self.process_bom_row( 

259 item, 

260 level=level + 1, 

261 multiplier=multiplier * bom_item.quantity, 

262 **kwargs, 

263 ) 

264 

265 def get_substitute_data(self, bom_item: BomItem) -> dict: 

266 """Return substitute part data for a BomItem.""" 

267 substitute_part_data = {} 

268 

269 idx = 0 

270 

271 for substitute in bom_item.substitutes.all(): 

272 substitute_part_data.update({f'substitute_{idx}': substitute.part.name}) 

273 

274 idx += 1 

275 

276 self.n_substitute_cols = max(self.n_substitute_cols, idx) 

277 

278 return substitute_part_data 

279 

280 def get_supplier_data(self, bom_item: BomItem) -> dict: 

281 """Return supplier and manufacturer data for a BomItem.""" 

282 supplier_part_data = {} 

283 

284 idx = 0 

285 

286 for supplier_part in bom_item.sub_part.supplier_parts.all(): 

287 manufacturer_part = supplier_part.manufacturer_part 

288 supplier_part_data.update({ 

289 f'supplier_name_{idx}': supplier_part.supplier.name 

290 if supplier_part.supplier 

291 else '', 

292 f'supplier_sku_{idx}': supplier_part.SKU, 

293 f'supplier_mpn_{idx}': manufacturer_part.MPN 

294 if manufacturer_part 

295 else '', 

296 }) 

297 

298 idx += 1 

299 

300 self.n_supplier_cols = max(self.n_supplier_cols, idx) 

301 

302 return supplier_part_data 

303 

304 def get_manufacturer_data(self, bom_item: BomItem) -> dict: 

305 """Return manufacturer data for a BomItem.""" 

306 manufacturer_part_data = {} 

307 

308 idx = 0 

309 

310 for manufacturer_part in bom_item.sub_part.manufacturer_parts.all(): 

311 manufacturer_part_data.update({ 

312 f'manufacturer_name_{idx}': manufacturer_part.manufacturer.name 

313 if manufacturer_part.manufacturer 

314 else '', 

315 f'manufacturer_mpn_{idx}': manufacturer_part.MPN, 

316 }) 

317 

318 idx += 1 

319 

320 self.n_manufacturer_cols = max(self.n_manufacturer_cols, idx) 

321 

322 return manufacturer_part_data 

323 

324 def get_parameter_data(self, bom_item: BomItem) -> dict: 

325 """Return parameter data for a BomItem.""" 

326 parameter_data = {} 

327 

328 for parameter in bom_item.sub_part.parameters.all(): 

329 template = parameter.template 

330 if template.pk not in self.parameters: 

331 self.parameters[template.pk] = template.name 

332 

333 parameter_data.update({f'parameter_{template.pk}': parameter.data}) 

334 

335 return parameter_data