Coverage for src/backend/InvenTree/plugin/base/barcodes/mixins.py: 24%

1"""Plugin mixin classes for barcode plugin.""" 

2 

3from __future__ import annotations 

4 

5from django.core.exceptions import ValidationError 

6from django.db.models import Q 

7from django.utils.translation import gettext_lazy as _ 

8 

9import structlog 

10 

11from company.models import Company, ManufacturerPart, SupplierPart 

12from InvenTree.exceptions import log_error 

13from InvenTree.models import InvenTreeBarcodeMixin 

14from order.models import PurchaseOrder 

15from part.models import Part 

16from plugin import PluginMixinEnum 

17from plugin.base.integration.SettingsMixin import SettingsMixin 

18 

19logger = structlog.get_logger('inventree') 

20 

21 

22class BarcodeMixin: 

23 """Mixin that enables barcode handling. 

24 

25 Custom barcode plugins should use and extend this mixin as necessary. 

26 """ 

27 

28 ACTION_NAME = '' 

29 

30 class MixinMeta: 

31 """Meta options for this mixin.""" 

32 

33 MIXIN_NAME = 'Barcode' 

34 

35 def __init__(self): 

36 """Register mixin.""" 

37 super().__init__() 

38 self.add_mixin(PluginMixinEnum.BARCODE, 'has_barcode', __class__) 

39 

40 @property 

41 def has_barcode(self): 

42 """Does this plugin have everything needed to process a barcode.""" 

43 return True 

44 

45 def scan(self, barcode_data: str, user, **kwargs) -> dict | None: 

46 """Scan a barcode against this plugin. 

47 

48 This method is explicitly called from the /scan/ API endpoint, 

49 and thus it is expected that any barcode which matches this barcode will return a result. 

50 

51 If this plugin finds a match against the provided barcode, it should return a dict object 

52 with the intended result. 

53 

54 Default return value is None 

55 """ 

56 return None 

57 

58 @property 

59 def has_barcode_generation(self): 

60 """Does this plugin support barcode generation.""" 

61 try: 

62 # Attempt to call the generate method 

63 self.generate(None) 

64 except NotImplementedError: 

65 # If a NotImplementedError is raised, then barcode generation is not supported 

66 return False 

67 except: 

68 pass 

69 

70 return True 

71 

72 def generate(self, model_instance: InvenTreeBarcodeMixin) -> str: 

73 """Generate barcode data for the given model instance. 

74 

75 Arguments: 

76 model_instance: The model instance to generate barcode data for. It is extending the InvenTreeBarcodeMixin. 

77 

78 Returns: The generated barcode data. 

79 """ 

80 raise NotImplementedError('Generate must be implemented by a plugin') 

81 

82 

83class SupplierBarcodeMixin(BarcodeMixin): 

84 """Mixin that provides default implementations for scan functions for supplier barcodes. 

85 

86 Custom supplier barcode plugins should use this mixin and implement the 

87 extract_barcode_fields function. 

88 """ 

89 

90 # Set of standard field names which can be extracted from the barcode 

91 CUSTOMER_ORDER_NUMBER = 'customer_order_number' 

92 SUPPLIER_ORDER_NUMBER = 'supplier_order_number' 

93 PACKING_LIST_NUMBER = 'packing_list_number' 

94 SHIP_DATE = 'ship_date' 

95 CUSTOMER_PART_NUMBER = 'customer_part_number' 

96 SUPPLIER_PART_NUMBER = 'supplier_part_number' 

97 PURCHASE_ORDER_LINE = 'purchase_order_line' 

98 QUANTITY = 'quantity' 

99 DATE_CODE = 'date_code' 

100 LOT_CODE = 'lot_code' 

101 COUNTRY_OF_ORIGIN = 'country_of_origin' 

102 MANUFACTURER = 'manufacturer' 

103 MANUFACTURER_PART_NUMBER = 'manufacturer_part_number' 

104 

105 def __init__(self): 

106 """Register mixin.""" 

107 super().__init__() 

108 self.add_mixin(PluginMixinEnum.SUPPLIER_BARCODE, True, __class__) 

109 

110 def get_field_value(self, key, backup_value=None): 

111 """Return the value of a barcode field.""" 

112 fields = getattr(self, 'barcode_fields', None) or {} 

113 

114 return fields.get(key, backup_value) 

115 

116 def get_part(self) -> Part | None: 

117 """Extract the Part object from the barcode fields.""" 

118 # TODO: Implement this 

119 return None 

120 

121 @property 

122 def quantity(self): 

123 """Return the quantity from the barcode fields.""" 

124 return self.get_field_value(self.QUANTITY) 

125 

126 @property 

127 def supplier_part_number(self): 

128 """Return the supplier part number from the barcode fields.""" 

129 return self.get_field_value(self.SUPPLIER_PART_NUMBER) 

130 

131 def get_supplier_part(self) -> SupplierPart | None: 

132 """Return the SupplierPart object for the scanned barcode. 

133 

134 Returns: 

135 SupplierPart object or None 

136 

137 - Filter by the Supplier ID associated with the plugin 

138 - Filter by SKU (if available) 

139 - If more than one match is found, filter by MPN (if available) 

140 

141 """ 

142 sku = self.supplier_part_number 

143 mpn = self.manufacturer_part_number 

144 

145 # Require at least SKU or MPN for lookup 

146 if not sku and not mpn: 

147 return None 

148 

149 supplier_parts = SupplierPart.objects.all() 

150 

151 # Filter by supplier 

152 if supplier := self.get_supplier(cache=True): 

153 supplier_parts = supplier_parts.filter(supplier=supplier) 

154 

155 if sku: 

156 supplier_parts = supplier_parts.filter(SKU=sku) 

157 

158 # Attempt additional filtering by MPN if multiple matches are found 

159 if mpn and supplier_parts.count() > 1: 

160 manufacturer_parts = ManufacturerPart.objects.filter(MPN=mpn) 

161 if manufacturer_parts.count() > 0: 

162 supplier_parts = supplier_parts.filter( 

163 manufacturer_part__in=manufacturer_parts 

164 ) 

165 

166 # Requires a unique match 

167 if len(supplier_parts) == 1: 

168 return supplier_parts.first() 

169 

170 @property 

171 def manufacturer_part_number(self): 

172 """Return the manufacturer part number from the barcode fields.""" 

173 return self.get_field_value(self.MANUFACTURER_PART_NUMBER) 

174 

175 def get_manufacturer_part(self) -> ManufacturerPart | None: 

176 """Return the ManufacturerPart object for the scanned barcode. 

177 

178 Returns: 

179 ManufacturerPart object or None 

180 """ 

181 mpn = self.manufacturer_part_number 

182 

183 if not mpn: 

184 return None 

185 

186 parts = ManufacturerPart.objects.filter(MPN=mpn) 

187 

188 if supplier := self.get_supplier(cache=True): 

189 # Manufacturer part must be associated with the supplier 

190 # Case 1: Manufactured by this supplier 

191 q1 = Q(manufacturer=supplier) 

192 # Case 2: Supplied by this supplier 

193 m = ( 

194 SupplierPart.objects 

195 .filter(supplier=supplier) 

196 .values_list('manufacturer_part', flat=True) 

197 .distinct() 

198 ) 

199 q2 = Q(pk__in=m) 

200 

201 parts = parts.filter(q1 | q2).distinct() 

202 

203 # Requires a unique match 

204 if len(parts) == 1: 

205 return parts.first() 

206 

207 @property 

208 def customer_order_number(self): 

209 """Return the customer order number from the barcode fields.""" 

210 return self.get_field_value(self.CUSTOMER_ORDER_NUMBER) 

211 

212 @property 

213 def supplier_order_number(self): 

214 """Return the supplier order number from the barcode fields.""" 

215 return self.get_field_value(self.SUPPLIER_ORDER_NUMBER) 

216 

217 def get_purchase_order(self) -> PurchaseOrder | None: 

218 """Extract the PurchaseOrder object from the barcode fields. 

219 

220 Inspect the customer_order_number and supplier_order_number fields, 

221 and try to find a matching PurchaseOrder object. 

222 

223 Returns: 

224 PurchaseOrder object or None 

225 """ 

226 customer_order_number = self.customer_order_number 

227 supplier_order_number = self.supplier_order_number 

228 

229 if not (customer_order_number or supplier_order_number): 

230 return None 

231 

232 # First, attempt lookup based on the customer_order_number 

233 

234 if customer_order_number: 

235 orders = PurchaseOrder.objects.filter(reference=customer_order_number) 

236 elif supplier_order_number: 

237 orders = PurchaseOrder.objects.filter( 

238 supplier_reference=supplier_order_number 

239 ) 

240 

241 if supplier := self.get_supplier(cache=True): 

242 orders = orders.filter(supplier=supplier) 

243 

244 # Requires a unique match 

245 if len(orders) == 1: 

246 return orders.first() 

247 

248 def extract_barcode_fields(self, barcode_data: str) -> dict[str, str]: 

249 """Method to extract barcode fields from barcode data. 

250 

251 This method should return a dict object where the keys are the field names, 

252 as per the "standard field names" (defined in the SuppliedBarcodeMixin class). 

253 

254 This method *must* be implemented by each plugin 

255 

256 Returns: 

257 A dict object containing the barcode fields. 

258 

259 """ 

260 raise NotImplementedError( 

261 'extract_barcode_fields must be implemented by each plugin' 

262 ) 

263 

264 def scan(self, barcode_data: str, user, **kwargs) -> dict | None: 

265 """Perform a generic 'scan' operation on a supplier barcode. 

266 

267 The supplier barcode may provide sufficient information to match against 

268 one of the following model types: 

269 

270 - SupplierPart 

271 - ManufacturerPart 

272 - PurchaseOrder 

273 - PurchaseOrderLineItem (todo) 

274 - StockItem (todo) 

275 - Part (todo) 

276 

277 If any matches are made, return a dict object containing the relevant information. 

278 """ 

279 barcode_data = str(barcode_data).strip() 

280 

281 self.barcode_fields = self.extract_barcode_fields(barcode_data) 

282 

283 # Generate possible matches for this barcode 

284 # Note: Each of these functions can be overridden by the plugin (if necessary) 

285 matches = { 

286 Part.barcode_model_type(): self.get_part(), 

287 PurchaseOrder.barcode_model_type(): self.get_purchase_order(), 

288 SupplierPart.barcode_model_type(): self.get_supplier_part(), 

289 ManufacturerPart.barcode_model_type(): self.get_manufacturer_part(), 

290 } 

291 

292 data = {} 

293 

294 # At least one matching item was found 

295 has_match = False 

296 

297 for k, v in matches.items(): 

298 if v and hasattr(v, 'pk'): 

299 has_match = True 

300 data[k] = v.format_matched_response(user=user) 

301 

302 if not has_match: 

303 return None 

304 

305 # Add in supplier information (if available) 

306 if supplier := self.get_supplier(): 

307 data['company'] = {'pk': supplier.pk} 

308 

309 data['success'] = _('Found matching item') 

310 

311 return data 

312 

313 def scan_receive_item( 

314 self, 

315 barcode_data: str, 

316 user, 

317 supplier=None, 

318 line_item=None, 

319 purchase_order=None, 

320 location=None, 

321 auto_allocate: bool = True, 

322 **kwargs, 

323 ) -> dict: 

324 """Attempt to receive an item against a PurchaseOrder via barcode scanning. 

325 

326 Arguments: 

327 barcode_data: The raw barcode data 

328 user: The User performing the action 

329 supplier: The Company object to receive against (or None) 

330 purchase_order: The PurchaseOrder object to receive against (or None) 

331 line_item: The PurchaseOrderLineItem object to receive against (or None) 

332 location: The StockLocation object to receive into (or None) 

333 auto_allocate: If True, automatically receive the item (if possible) 

334 

335 Returns: 

336 A dict object containing the result of the action. 

337 

338 The more "context" data that can be provided, the better the chances of a successful match. 

339 """ 

340 barcode_data = str(barcode_data).strip() 

341 

342 self.barcode_fields = self.extract_barcode_fields(barcode_data) 

343 

344 # Extract supplier information 

345 supplier = supplier or self.get_supplier(cache=True) 

346 

347 """Construct Debug Response 

348 This is returned if a perfect match is not found with the info provided from the barcode 

349 

350 Response Info: 

351 'supplier': get supplier ID 

352 'PO': Represented for "Purchase Order", find PO number to supplier 

353 'supplier_part': find supplier part info to supplier 

354 'no_match': Boolean, did we find a perfect match with info given? False is Yes, True is No 

355 """ 

356 debug_response = {} 

357 

358 if supplier is None: 

359 # No supplier information available 

360 debug_response['supplier'] = None 

361 else: 

362 debug_response['supplier'] = supplier.name 

363 

364 # Extract purchase order information 

365 purchase_order = purchase_order or self.get_purchase_order() 

366 

367 if purchase_order is None or purchase_order.supplier != supplier: 

368 # Purchase order does not match supplier 

369 debug_response['PO'] = None 

370 else: 

371 debug_response['PO'] = purchase_order.reference 

372 

373 supplier_part = self.get_supplier_part() 

374 

375 if supplier_part is None: 

376 # No supplier part information available 

377 debug_response['supplier_part'] = None 

378 else: 

379 debug_response['supplier_part'] = str(supplier_part.part) 

380 

381 # Attempt to find matching line item 

382 if not line_item and purchase_order != None: 

383 line_items = purchase_order.lines.filter(part=supplier_part) 

384 if line_items.count() == 1: 

385 line_item = line_items.first() 

386 

387 # If Purchase Order or Supplier Part does not exist, throw debug response 

388 if debug_response['PO'] is None or debug_response['supplier_part'] is None: 

389 debug_response['no_match'] = True 

390 return debug_response 

391 

392 if not line_item or not line_item.part: 

393 return {'error': _('No matching line item found'), 'no_match': False} 

394 

395 if line_item.part != supplier_part: 

396 return { 

397 'error': _('Supplier part does not match line item'), 

398 'no_match': False, 

399 } 

400 

401 if line_item.is_completed(): 

402 return {'error': _('Line item is already completed'), 'no_match': False} 

403 

404 # Extract location information for the line item 

405 location = ( 

406 location 

407 or line_item.destination 

408 or purchase_order.destination 

409 or line_item.part.part.get_default_location() 

410 ) 

411 

412 # Extract quantity information 

413 quantity = self.quantity 

414 

415 # At this stage, we *should* have enough information to attempt to receive the item 

416 # If auto_allocate is True, attempt to receive the item automatically 

417 # Otherwise, return the required information to the client 

418 action_required = not auto_allocate or location is None or quantity is None 

419 

420 if quantity is None: 

421 quantity = line_item.remaining() 

422 

423 quantity = float(quantity) 

424 

425 # Construct a response object 

426 response = { 

427 'lineitem': { 

428 'pk': line_item.pk, 

429 'quantity': quantity, 

430 'supplier_part': supplier_part.pk, 

431 'purchase_order': purchase_order.pk, 

432 'location': location.pk if location else None, 

433 }, 

434 'no_match': False, 

435 } 

436 

437 if action_required: 

438 # Further information is required to receive the item 

439 response['action_required'] = _( 

440 'Further information required to receive line item' 

441 ) 

442 else: 

443 # Use the information we have to attempt to receive the item into stock 

444 try: 

445 purchase_order.receive_line_item( 

446 line_item, location, quantity, user, barcode=barcode_data 

447 ) 

448 response['success'] = _('Received purchase order line item') 

449 except ValidationError as e: 

450 # Pass a ValidationError back to the client 

451 response['error'] = e.message 

452 except Exception: 

453 # Handle any other exceptions 

454 log_error('scan_receive_item', plugin=self.slug) 

455 response['error'] = _('Failed to receive line item') 

456 

457 return response 

458 

459 def get_supplier(self, cache: bool = False) -> Company | None: 

460 """Get the supplier for the SUPPLIER_ID set in the plugin settings. 

461 

462 If it's not defined, try to guess it and set it if possible. 

463 """ 

464 if not isinstance(self, SettingsMixin): 

465 return None 

466 

467 def _cache_supplier(supplier): 

468 """Cache and return the supplier object.""" 

469 if cache: 

470 self._supplier = supplier 

471 return supplier 

472 

473 # Cache the supplier object, so we don't have to look it up every time 

474 if cache and hasattr(self, '_supplier'): 

475 return self._supplier 

476 

477 if supplier_pk := self.get_setting('SUPPLIER_ID'): 

478 return _cache_supplier(Company.objects.filter(pk=supplier_pk).first()) 

479 

480 if not (supplier_name := getattr(self, 'DEFAULT_SUPPLIER_NAME', None)): 

481 return _cache_supplier(None) 

482 

483 suppliers = Company.objects.filter( 

484 name__icontains=supplier_name, is_supplier=True 

485 ) 

486 

487 if len(suppliers) != 1: 

488 return _cache_supplier(None) 

489 

490 supplier = suppliers.first() 

491 assert supplier 

492 

493 self.set_setting('SUPPLIER_ID', supplier.pk) 

494 

495 return _cache_supplier(supplier) 

496 

497 @classmethod 

498 def ecia_field_map(cls): 

499 """Return a dict mapping ECIA field names to internal field names. 

500 

501 Ref: https://www.ecianow.org/assets/docs/ECIA_Specifications.pdf 

502 

503 Note that a particular plugin may need to reimplement this method, 

504 if it does not use the standard field names. 

505 """ 

506 return { 

507 'K': cls.CUSTOMER_ORDER_NUMBER, 

508 '1K': cls.SUPPLIER_ORDER_NUMBER, 

509 '11K': cls.PACKING_LIST_NUMBER, 

510 '6D': cls.SHIP_DATE, 

511 '9D': cls.DATE_CODE, 

512 '10D': cls.DATE_CODE, 

513 '4K': cls.PURCHASE_ORDER_LINE, 

514 '14K': cls.PURCHASE_ORDER_LINE, 

515 'P': cls.SUPPLIER_PART_NUMBER, 

516 '1P': cls.MANUFACTURER_PART_NUMBER, 

517 '30P': cls.SUPPLIER_PART_NUMBER, 

518 '1T': cls.LOT_CODE, 

519 '4L': cls.COUNTRY_OF_ORIGIN, 

520 '1V': cls.MANUFACTURER, 

521 'Q': cls.QUANTITY, 

522 } 

523 

524 @classmethod 

525 def parse_ecia_barcode2d(cls, barcode_data: str) -> dict[str, str]: 

526 """Parse a standard ECIA 2D barcode. 

527 

528 Ref: https://www.ecianow.org/assets/docs/ECIA_Specifications.pdf 

529 

530 Arguments: 

531 barcode_data: The raw barcode data 

532 

533 Returns: 

534 A dict containing the parsed barcode fields 

535 """ 

536 # Split data into separate fields 

537 fields = cls.parse_isoiec_15434_barcode2d(barcode_data) 

538 

539 barcode_fields = {} 

540 

541 if not fields: 

542 return barcode_fields 

543 

544 for field in fields: 

545 for identifier, field_name in cls.ecia_field_map().items(): 

546 if field.startswith(identifier): 

547 barcode_fields[field_name] = field[len(identifier) :] 

548 break 

549 

550 return barcode_fields 

551 

552 @staticmethod 

553 def split_fields( 

554 barcode_data: str, delimiter: str = ',', header: str = '', trailer: str = '' 

555 ) -> list[str]: 

556 """Generic method for splitting barcode data into separate fields.""" 

557 if header and barcode_data.startswith(header): 

558 barcode_data = barcode_data[len(header) :] 

559 

560 if trailer and barcode_data.endswith(trailer): 

561 barcode_data = barcode_data[: -len(trailer)] 

562 

563 return barcode_data.split(delimiter) 

564 

565 @staticmethod 

566 def parse_isoiec_15434_barcode2d(barcode_data: str) -> list[str]: 

567 """Parse a ISO/IEC 15434 barcode, returning the split data section.""" 

568 OLD_MOUSER_HEADER = '>[)>06\x1d' 

569 HEADER = '[)>\x1e06\x1d' 

570 TRAILER = '\x1e\x04' 

571 DELIMITER = '\x1d' 

572 

573 # Some old mouser barcodes start with this messed up header 

574 if barcode_data.startswith(OLD_MOUSER_HEADER): 

575 barcode_data = barcode_data.replace(OLD_MOUSER_HEADER, HEADER, 1) 

576 

577 # Check that the barcode starts with the necessary header 

578 if not barcode_data.startswith(HEADER): 

579 return [] 

580 

581 return SupplierBarcodeMixin.split_fields( 

582 barcode_data, delimiter=DELIMITER, header=HEADER, trailer=TRAILER 

583 )