""" Part database model definitions """ # -*- coding: utf-8 -*- from __future__ import unicode_literals import os from django.utils.translation import gettext_lazy as _ from django.core.exceptions import ValidationError from django.urls import reverse from django.db import models, transaction from django.db.utils import IntegrityError from django.db.models import Sum, UniqueConstraint from django.db.models.functions import Coalesce from django.core.validators import MinValueValidator from django.contrib.auth.models import User from django.db.models.signals import pre_delete from django.dispatch import receiver from markdownx.models import MarkdownxField from django_cleanup import cleanup from mptt.models import TreeForeignKey, MPTTModel from stdimage.models import StdImageField from decimal import Decimal from datetime import datetime from rapidfuzz import fuzz import hashlib from InvenTree import helpers from InvenTree import validators from InvenTree.models import InvenTreeTree, InvenTreeAttachment from InvenTree.fields import InvenTreeURLField from InvenTree.helpers import decimal2string, normalize from InvenTree.status_codes import BuildStatus, PurchaseOrderStatus from build import models as BuildModels from order import models as OrderModels from company.models import SupplierPart from stock import models as StockModels import common.models import part.settings as part_settings class PartCategory(InvenTreeTree): """ PartCategory provides hierarchical organization of Part objects. Attributes: name: Name of this category parent: Parent category default_location: Default storage location for parts in this category or child categories default_keywords: Default keywords for parts created in this category """ default_location = TreeForeignKey( 'stock.StockLocation', related_name="default_categories", null=True, blank=True, on_delete=models.SET_NULL, help_text=_('Default location for parts in this category') ) default_keywords = models.CharField(null=True, blank=True, max_length=250, help_text=_('Default keywords for parts in this category')) def get_absolute_url(self): return reverse('category-detail', kwargs={'pk': self.id}) class Meta: verbose_name = _("Part Category") verbose_name_plural = _("Part Categories") def get_parts(self, cascade=True): """ Return a queryset for all parts under this category. args: cascade - If True, also look under subcategories (default = True) """ if cascade: """ Select any parts which exist in this category or any child categories """ query = Part.objects.filter(category__in=self.getUniqueChildren(include_self=True)) else: query = Part.objects.filter(category=self.pk) return query @property def item_count(self): return self.partcount() def partcount(self, cascade=True, active=False): """ Return the total part count under this category (including children of child categories) """ query = self.get_parts(cascade=cascade) if active: query = query.filter(active=True) return query.count() @property def has_parts(self): """ True if there are any parts in this category """ return self.partcount() > 0 def prefetch_parts_parameters(self, cascade=True): """ Prefectch parts parameters """ return self.get_parts(cascade=cascade).prefetch_related('parameters', 'parameters__template').all() def get_unique_parameters(self, cascade=True, prefetch=None): """ Get all unique parameter names for all parts from this category """ unique_parameters_names = [] if prefetch: parts = prefetch else: parts = self.prefetch_parts_parameters(cascade=cascade) for part in parts: for parameter in part.parameters.all(): parameter_name = parameter.template.name if parameter_name not in unique_parameters_names: unique_parameters_names.append(parameter_name) return sorted(unique_parameters_names) def get_parts_parameters(self, cascade=True, prefetch=None): """ Get all parameter names and values for all parts from this category """ category_parameters = [] if prefetch: parts = prefetch else: parts = self.prefetch_parts_parameters(cascade=cascade) for part in parts: part_parameters = { 'pk': part.pk, 'name': part.name, 'description': part.description, } # Add IPN only if it exists if part.IPN: part_parameters['IPN'] = part.IPN for parameter in part.parameters.all(): parameter_name = parameter.template.name parameter_value = parameter.data part_parameters[parameter_name] = parameter_value category_parameters.append(part_parameters) return category_parameters @classmethod def get_parent_categories(cls): """ Return tuple list of parent (root) categories """ # Get root nodes root_categories = cls.objects.filter(level=0) parent_categories = [] for category in root_categories: parent_categories.append((category.id, category.name)) return parent_categories def get_parameter_templates(self): """ Return parameter templates associated to category """ prefetch = PartCategoryParameterTemplate.objects.prefetch_related('category', 'parameter_template') return prefetch.filter(category=self.id) @receiver(pre_delete, sender=PartCategory, dispatch_uid='partcategory_delete_log') def before_delete_part_category(sender, instance, using, **kwargs): """ Receives before_delete signal for PartCategory object Before deleting, update child Part and PartCategory objects: - For each child category, set the parent to the parent of *this* category - For each part, set the 'category' to the parent of *this* category """ # Update each part in this category to point to the parent category for part in instance.parts.all(): part.category = instance.parent part.save() # Update each child category for child in instance.children.all(): child.parent = instance.parent child.save() def rename_part_image(instance, filename): """ Function for renaming a part image file Args: instance: Instance of a Part object filename: Name of original uploaded file Returns: Cleaned filename in format part__img """ base = 'part_images' fname = os.path.basename(filename) return os.path.join(base, fname) def match_part_names(match, threshold=80, reverse=True, compare_length=False): """ Return a list of parts whose name matches the search term using fuzzy search. Args: match: Term to match against threshold: Match percentage that must be exceeded (default = 65) reverse: Ordering for search results (default = True - highest match is first) compare_length: Include string length checks Returns: A sorted dict where each element contains the following key:value pairs: - 'part' : The matched part - 'ratio' : The matched ratio """ match = str(match).strip().lower() if len(match) == 0: return [] parts = Part.objects.all() matches = [] for part in parts: compare = str(part.name).strip().lower() if len(compare) == 0: continue ratio = fuzz.partial_token_sort_ratio(compare, match) if compare_length: # Also employ primitive length comparison # TODO - Improve this somewhat... l_min = min(len(match), len(compare)) l_max = max(len(match), len(compare)) ratio *= (l_min / l_max) if ratio >= threshold: matches.append({ 'part': part, 'ratio': round(ratio, 1) }) matches = sorted(matches, key=lambda item: item['ratio'], reverse=reverse) return matches @cleanup.ignore class Part(MPTTModel): """ The Part object represents an abstract part, the 'concept' of an actual entity. An actual physical instance of a Part is a StockItem which is treated separately. Parts can be used to create other parts (as part of a Bill of Materials or BOM). Attributes: name: Brief name for this part variant: Optional variant number for this part - Must be unique for the part name category: The PartCategory to which this part belongs description: Longer form description of the part keywords: Optional keywords for improving part search results IPN: Internal part number (optional) revision: Part revision is_template: If True, this part is a 'template' part and cannot be instantiated as a StockItem link: Link to an external page with more information about this part (e.g. internal Wiki) image: Image of this part default_location: Where the item is normally stored (may be null) default_supplier: The default SupplierPart which should be used to procure and stock this part minimum_stock: Minimum preferred quantity to keep in stock units: Units of measure for this part (default='pcs') salable: Can this part be sold to customers? assembly: Can this part be build from other parts? component: Can this part be used to make other parts? purchaseable: Can this part be purchased from suppliers? trackable: Trackable parts can have unique serial numbers assigned, etc, etc active: Is this part active? Parts are deactivated instead of being deleted virtual: Is this part "virtual"? e.g. a software product or similar notes: Additional notes field for this part creation_date: Date that this part was added to the database creation_user: User who added this part to the database responsible: User who is responsible for this part (optional) """ class Meta: verbose_name = _("Part") verbose_name_plural = _("Parts") ordering = ['name', ] class MPTTMeta: # For legacy reasons the 'variant_of' field is used to indicate the MPTT parent parent_attr = 'variant_of' def save(self, *args, **kwargs): """ Overrides the save() function for the Part model. If the part image has been updated, then check if the "old" (previous) image is still used by another part. If not, it is considered "orphaned" and will be deleted. """ # Get category templates settings add_category_templates = kwargs.pop('add_category_templates', None) if self.pk: previous = Part.objects.get(pk=self.pk) if previous.image and not self.image == previous.image: # Are there any (other) parts which reference the image? n_refs = Part.objects.filter(image=previous.image).exclude(pk=self.pk).count() if n_refs == 0: previous.image.delete(save=False) self.clean() self.validate_unique() super().save(*args, **kwargs) if add_category_templates: # Get part category category = self.category if add_category_templates: # Store templates added to part template_list = [] # Create part parameters for selected category category_templates = add_category_templates['main'] if category_templates: for template in category.get_parameter_templates(): parameter = PartParameter.create(part=self, template=template.parameter_template, data=template.default_value, save=True) if parameter: template_list.append(template.parameter_template) # Create part parameters for parent category category_templates = add_category_templates['parent'] if category_templates: # Get parent categories parent_categories = category.get_ancestors() for category in parent_categories: for template in category.get_parameter_templates(): # Check that template wasn't already added if template.parameter_template not in template_list: try: PartParameter.create(part=self, template=template.parameter_template, data=template.default_value, save=True) except IntegrityError: # PartParameter already exists pass def __str__(self): return f"{self.full_name} - {self.description}" def checkAddToBOM(self, parent): """ Check if this Part can be added to the BOM of another part. This will fail if: a) The parent part is the same as this one b) The parent part is used in the BOM for *this* part c) The parent part is used in the BOM for any child parts under this one Failing this check raises a ValidationError! """ if parent is None: return if self.pk == parent.pk: raise ValidationError({'sub_part': _("Part '{p1}' is used in BOM for '{p2}' (recursive)".format( p1=str(self), p2=str(parent) ))}) # Ensure that the parent part does not appear under any child BOM item! for item in self.bom_items.all(): # Check for simple match if item.sub_part == parent: raise ValidationError({'sub_part': _("Part '{p1}' is used in BOM for '{p2}' (recursive)".format( p1=str(parent), p2=str(self) ))}) # And recursively check too item.sub_part.checkAddToBOM(parent) def checkIfSerialNumberExists(self, sn, exclude_self=False): """ Check if a serial number exists for this Part. Note: Serial numbers must be unique across an entire Part "tree", so here we filter by the entire tree. """ parts = Part.objects.filter(tree_id=self.tree_id) stock = StockModels.StockItem.objects.filter(part__in=parts, serial=sn) if exclude_self: stock = stock.exclude(pk=self.pk) return stock.exists() def find_conflicting_serial_numbers(self, serials): """ For a provided list of serials, return a list of those which are conflicting. """ conflicts = [] for serial in serials: if self.checkIfSerialNumberExists(serial, exclude_self=True): conflicts.append(serial) return conflicts def getLatestSerialNumber(self): """ Return the "latest" serial number for this Part. If *all* the serial numbers are integers, then this will return the highest one. Otherwise, it will simply return the serial number most recently added. Note: Serial numbers must be unique across an entire Part "tree", so we filter by the entire tree. """ parts = Part.objects.filter(tree_id=self.tree_id) stock = StockModels.StockItem.objects.filter(part__in=parts).exclude(serial=None) # There are no matchin StockItem objects (skip further tests) if not stock.exists(): return None # Attempt to coerce the returned serial numbers to integers # If *any* are not integers, fail! try: ordered = sorted(stock.all(), reverse=True, key=lambda n: int(n.serial)) if len(ordered) > 0: return ordered[0].serial # One or more of the serial numbers was non-numeric # In this case, the "best" we can do is return the most recent except ValueError: return stock.last().serial # No serial numbers found return None def getSerialNumberString(self, quantity=1): """ Return a formatted string representing the next available serial numbers, given a certain quantity of items. """ latest = self.getLatestSerialNumber() quantity = int(quantity) # No serial numbers can be found, assume 1 as the first serial if latest is None: latest = 0 # Attempt to turn into an integer try: latest = int(latest) except: pass if type(latest) is int: if quantity >= 2: text = '{n} - {m}'.format(n=latest + 1, m=latest + 1 + quantity) return _('Next available serial numbers are') + ' ' + text else: text = str(latest + 1) return _('Next available serial number is') + ' ' + text else: # Non-integer values, no option but to return latest return _('Most recent serial number is') + ' ' + str(latest) @property def full_name(self): """ Format a 'full name' for this Part. - IPN (if not null) - Part name - Part variant (if not null) Elements are joined by the | character """ elements = [] if self.IPN: elements.append(self.IPN) elements.append(self.name) if self.revision: elements.append(self.revision) return ' | '.join(elements) def set_category(self, category): # Ignore if the category is already the same if self.category == category: return self.category = category self.save() def get_absolute_url(self): """ Return the web URL for viewing this part """ return reverse('part-detail', kwargs={'pk': self.id}) def get_image_url(self): """ Return the URL of the image for this part """ if self.image: return helpers.getMediaUrl(self.image.url) else: return helpers.getBlankImage() def get_thumbnail_url(self): """ Return the URL of the image thumbnail for this part """ if self.image: return helpers.getMediaUrl(self.image.thumbnail.url) else: return helpers.getBlankThumbnail() def validate_unique(self, exclude=None): """ Validate that a part is 'unique'. Uniqueness is checked across the following (case insensitive) fields: * Name * IPN * Revision e.g. there can exist multiple parts with the same name, but only if they have a different revision or internal part number. """ super().validate_unique(exclude) # User can decide whether duplicate IPN (Internal Part Number) values are allowed allow_duplicate_ipn = common.models.InvenTreeSetting.get_setting('PART_ALLOW_DUPLICATE_IPN') if not allow_duplicate_ipn: parts = Part.objects.filter(IPN__iexact=self.IPN) parts = parts.exclude(pk=self.pk) if parts.exists(): raise ValidationError({ 'IPN': _('Duplicate IPN not allowed in part settings'), }) # Part name uniqueness should be case insensitive try: parts = Part.objects.exclude(id=self.id).filter( name__iexact=self.name, IPN__iexact=self.IPN, revision__iexact=self.revision) if parts.exists(): msg = _("Part must be unique for name, IPN and revision") raise ValidationError({ "name": msg, "IPN": msg, "revision": msg, }) except Part.DoesNotExist: pass def clean(self): """ Perform cleaning operations for the Part model Update trackable status: If this part is trackable, and it is used in the BOM for a parent part which is *not* trackable, then we will force the parent part to be trackable. """ super().clean() if self.trackable: for item in self.used_in.all(): parent_part = item.part if not parent_part.trackable: parent_part.trackable = True parent_part.clean() parent_part.save() name = models.CharField(max_length=100, blank=False, help_text=_('Part name'), validators=[validators.validate_part_name] ) is_template = models.BooleanField(default=False, help_text=_('Is this part a template part?')) variant_of = models.ForeignKey('part.Part', related_name='variants', null=True, blank=True, limit_choices_to={ 'is_template': True, 'active': True, }, on_delete=models.SET_NULL, help_text=_('Is this part a variant of another part?')) description = models.CharField(max_length=250, blank=False, help_text=_('Part description')) keywords = models.CharField(max_length=250, blank=True, null=True, help_text=_('Part keywords to improve visibility in search results')) category = TreeForeignKey(PartCategory, related_name='parts', null=True, blank=True, on_delete=models.DO_NOTHING, help_text=_('Part category')) IPN = models.CharField(max_length=100, blank=True, null=True, help_text=_('Internal Part Number'), validators=[validators.validate_part_ipn]) revision = models.CharField(max_length=100, blank=True, null=True, help_text=_('Part revision or version number')) link = InvenTreeURLField(blank=True, null=True, help_text=_('Link to external URL')) image = StdImageField( upload_to=rename_part_image, null=True, blank=True, variations={'thumbnail': (128, 128)}, delete_orphans=True, ) default_location = TreeForeignKey('stock.StockLocation', on_delete=models.SET_NULL, blank=True, null=True, help_text=_('Where is this item normally stored?'), related_name='default_parts') def get_default_location(self): """ Get the default location for a Part (may be None). If the Part does not specify a default location, look at the Category this part is in. The PartCategory object may also specify a default stock location """ if self.default_location: return self.default_location elif self.category: # Traverse up the category tree until we find a default location cats = self.category.get_ancestors(ascending=True, include_self=True) for cat in cats: if cat.default_location: return cat.default_location # Default case - no default category found return None def get_default_supplier(self): """ Get the default supplier part for this part (may be None). - If the part specifies a default_supplier, return that - If there is only one supplier part available, return that - Else, return None """ if self.default_supplier: return self.default_supplier if self.supplier_count == 1: return self.supplier_parts.first() # Default to None if there are multiple suppliers to choose from return None default_supplier = models.ForeignKey(SupplierPart, on_delete=models.SET_NULL, blank=True, null=True, help_text=_('Default supplier part'), related_name='default_parts') minimum_stock = models.PositiveIntegerField(default=0, validators=[MinValueValidator(0)], help_text=_('Minimum allowed stock level')) units = models.CharField(max_length=20, default="", blank=True, null=True, help_text=_('Stock keeping units for this part')) assembly = models.BooleanField( default=False, verbose_name=_('Assembly'), help_text=_('Can this part be built from other parts?') ) component = models.BooleanField( default=part_settings.part_component_default, verbose_name=_('Component'), help_text=_('Can this part be used to build other parts?') ) trackable = models.BooleanField( default=part_settings.part_trackable_default, verbose_name=_('Trackable'), help_text=_('Does this part have tracking for unique items?')) purchaseable = models.BooleanField( default=part_settings.part_purchaseable_default, verbose_name=_('Purchaseable'), help_text=_('Can this part be purchased from external suppliers?')) salable = models.BooleanField( default=part_settings.part_salable_default, verbose_name=_('Salable'), help_text=_("Can this part be sold to customers?")) active = models.BooleanField( default=True, verbose_name=_('Active'), help_text=_('Is this part active?')) virtual = models.BooleanField( default=False, verbose_name=_('Virtual'), help_text=_('Is this a virtual part, such as a software product or license?')) notes = MarkdownxField(blank=True, null=True, help_text=_('Part notes - supports Markdown formatting')) bom_checksum = models.CharField(max_length=128, blank=True, help_text=_('Stored BOM checksum')) bom_checked_by = models.ForeignKey(User, on_delete=models.SET_NULL, blank=True, null=True, related_name='boms_checked') bom_checked_date = models.DateField(blank=True, null=True) creation_date = models.DateField(auto_now_add=True, editable=False, blank=True, null=True) creation_user = models.ForeignKey(User, on_delete=models.SET_NULL, blank=True, null=True, related_name='parts_created') responsible = models.ForeignKey(User, on_delete=models.SET_NULL, blank=True, null=True, related_name='parts_responible') def format_barcode(self, **kwargs): """ Return a JSON string for formatting a barcode for this Part object """ return helpers.MakeBarcode( "part", self.id, { "name": self.full_name, "url": reverse('api-part-detail', kwargs={'pk': self.id}), }, **kwargs ) @property def category_path(self): if self.category: return self.category.pathstring return '' @property def available_stock(self): """ Return the total available stock. - This subtracts stock which is already allocated to builds """ total = self.total_stock total -= self.allocation_count() return max(total, 0) @property def quantity_to_order(self): """ Return the quantity needing to be ordered for this part. """ # How many do we need to have "on hand" at any point? required = self.net_stock - self.minimum_stock if required < 0: return abs(required) # Do not need to order any return 0 required = self.net_stock return max(required, 0) @property def net_stock(self): """ Return the 'net' stock. It takes into account: - Stock on hand (total_stock) - Stock on order (on_order) - Stock allocated (allocation_count) This number (unlike 'available_stock') can be negative. """ return self.total_stock - self.allocation_count() + self.on_order def isStarredBy(self, user): """ Return True if this part has been starred by a particular user """ try: PartStar.objects.get(part=self, user=user) return True except PartStar.DoesNotExist: return False def need_to_restock(self): """ Return True if this part needs to be restocked (either by purchasing or building). If the allocated_stock exceeds the total_stock, then we need to restock. """ return (self.total_stock + self.on_order - self.allocation_count) < self.minimum_stock @property def can_build(self): """ Return the number of units that can be build with available stock """ # If this part does NOT have a BOM, result is simply the currently available stock if not self.has_bom: return 0 total = None # Calculate the minimum number of parts that can be built using each sub-part for item in self.bom_items.all().prefetch_related('sub_part__stock_items'): stock = item.sub_part.available_stock # If (by some chance) we get here but the BOM item quantity is invalid, # ignore! if item.quantity <= 0: continue n = int(stock / item.quantity) if total is None or n < total: total = n if total is None: total = 0 return max(total, 0) @property def active_builds(self): """ Return a list of outstanding builds. Builds marked as 'complete' or 'cancelled' are ignored """ return self.builds.filter(status__in=BuildStatus.ACTIVE_CODES) @property def inactive_builds(self): """ Return a list of inactive builds """ return self.builds.exclude(status__in=BuildStatus.ACTIVE_CODES) @property def quantity_being_built(self): """ Return the current number of parts currently being built """ stock_items = self.stock_items.filter(is_building=True) query = stock_items.aggregate( quantity=Coalesce(Sum('quantity'), Decimal(0)) ) return query['quantity'] def build_order_allocations(self): """ Return all 'BuildItem' objects which allocate this part to Build objects """ return BuildModels.BuildItem.objects.filter(stock_item__part__id=self.id) def build_order_allocation_count(self): """ Return the total amount of this part allocated to build orders """ query = self.build_order_allocations().aggregate(total=Coalesce(Sum('quantity'), 0)) return query['total'] def sales_order_allocations(self): """ Return all sales-order-allocation objects which allocate this part to a SalesOrder """ return OrderModels.SalesOrderAllocation.objects.filter(item__part__id=self.id) def sales_order_allocation_count(self): """ Return the tutal quantity of this part allocated to sales orders """ query = self.sales_order_allocations().aggregate(total=Coalesce(Sum('quantity'), 0)) return query['total'] def allocation_count(self): """ Return the total quantity of stock allocated for this part, against both build orders and sales orders. """ return sum([ self.build_order_allocation_count(), self.sales_order_allocation_count(), ]) def stock_entries(self, include_variants=True, in_stock=None): """ Return all stock entries for this Part. - If this is a template part, include variants underneath this. Note: To return all stock-entries for all part variants under this one, we need to be creative with the filtering. """ if include_variants: query = StockModels.StockItem.objects.filter(part__in=self.get_descendants(include_self=True)) else: query = self.stock_items if in_stock is True: query = query.filter(StockModels.StockItem.IN_STOCK_FILTER) elif in_stock is False: query = query.exclude(StockModels.StockItem.IN_STOCK_FILTER) return query @property def total_stock(self): """ Return the total stock quantity for this part. - Part may be stored in multiple locations - If this part is a "template" (variants exist) then these are counted too """ entries = self.stock_entries(in_stock=True) query = entries.aggregate(t=Coalesce(Sum('quantity'), Decimal(0))) return query['t'] @property def has_bom(self): return self.bom_count > 0 @property def has_trackable_parts(self): """ Return True if any parts linked in the Bill of Materials are trackable. This is important when building the part. """ for bom_item in self.bom_items.all(): if bom_item.sub_part.trackable: return True return False @property def bom_count(self): """ Return the number of items contained in the BOM for this part """ return self.bom_items.count() @property def used_in_count(self): """ Return the number of part BOMs that this part appears in """ return self.used_in.count() def get_bom_hash(self): """ Return a checksum hash for the BOM for this part. Used to determine if the BOM has changed (and needs to be signed off!) The hash is calculated by hashing each line item in the BOM. returns a string representation of a hash object which can be compared with a stored value """ hash = hashlib.md5(str(self.id).encode()) for item in self.bom_items.all().prefetch_related('sub_part'): hash.update(str(item.get_item_hash()).encode()) return str(hash.digest()) def is_bom_valid(self): """ Check if the BOM is 'valid' - if the calculated checksum matches the stored value """ return self.get_bom_hash() == self.bom_checksum @transaction.atomic def validate_bom(self, user): """ Validate the BOM (mark the BOM as validated by the given User. - Calculates and stores the hash for the BOM - Saves the current date and the checking user """ # Validate each line item too for item in self.bom_items.all(): item.validate_hash() self.bom_checksum = self.get_bom_hash() self.bom_checked_by = user self.bom_checked_date = datetime.now().date() self.save() @transaction.atomic def clear_bom(self): """ Clear the BOM items for the part (delete all BOM lines). """ self.bom_items.all().delete() def getRequiredParts(self, recursive=False, parts=set()): """ Return a list of parts required to make this part (i.e. BOM items). Args: recursive: If True iterate down through sub-assemblies parts: Set of parts already found (to prevent recursion issues) """ for bom_item in self.bom_items.all().select_related('sub_part'): sub_part = bom_item.sub_part if sub_part not in parts: parts.add(sub_part) if recursive: sub_part.getRequiredParts(recursive=True, parts=parts) return parts def get_allowed_bom_items(self): """ Return a list of parts which can be added to a BOM for this part. - Exclude parts which are not 'component' parts - Exclude parts which this part is in the BOM for """ # Start with a list of all parts designated as 'sub components' parts = Part.objects.filter(component=True) # Exclude this part parts = parts.exclude(id=self.id) # Exclude any parts that this part is used *in* (to prevent recursive BOMs) used_in = self.used_in.all() parts = parts.exclude(id__in=[item.part.id for item in used_in]) return parts @property def supplier_count(self): """ Return the number of supplier parts available for this part """ return self.supplier_parts.count() @property def has_pricing_info(self): """ Return true if there is pricing information for this part """ return self.get_price_range() is not None @property def has_complete_bom_pricing(self): """ Return true if there is pricing information for each item in the BOM. """ for item in self.bom_items.all().select_related('sub_part'): if not item.sub_part.has_pricing_info: return False return True def get_price_info(self, quantity=1, buy=True, bom=True): """ Return a simplified pricing string for this part Args: quantity: Number of units to calculate price for buy: Include supplier pricing (default = True) bom: Include BOM pricing (default = True) """ price_range = self.get_price_range(quantity, buy, bom) if price_range is None: return None min_price, max_price = price_range if min_price == max_price: return min_price min_price = normalize(min_price) max_price = normalize(max_price) return "{a} - {b}".format(a=min_price, b=max_price) def get_supplier_price_range(self, quantity=1): min_price = None max_price = None for supplier in self.supplier_parts.all(): price = supplier.get_price(quantity) if price is None: continue if min_price is None or price < min_price: min_price = price if max_price is None or price > max_price: max_price = price if min_price is None or max_price is None: return None min_price = normalize(min_price) max_price = normalize(max_price) return (min_price, max_price) def get_bom_price_range(self, quantity=1): """ Return the price range of the BOM for this part. Adds the minimum price for all components in the BOM. Note: If the BOM contains items without pricing information, these items cannot be included in the BOM! """ min_price = None max_price = None for item in self.bom_items.all().select_related('sub_part'): if item.sub_part.pk == self.pk: print("Warning: Item contains itself in BOM") continue prices = item.sub_part.get_price_range(quantity * item.quantity) if prices is None: continue low, high = prices if min_price is None: min_price = 0 if max_price is None: max_price = 0 min_price += low max_price += high if min_price is None or max_price is None: return None min_price = normalize(min_price) max_price = normalize(max_price) return (min_price, max_price) def get_price_range(self, quantity=1, buy=True, bom=True): """ Return the price range for this part. This price can be either: - Supplier price (if purchased from suppliers) - BOM price (if built from other parts) Returns: Minimum of the supplier price or BOM price. If no pricing available, returns None """ buy_price_range = self.get_supplier_price_range(quantity) if buy else None bom_price_range = self.get_bom_price_range(quantity) if bom else None if buy_price_range is None: return bom_price_range elif bom_price_range is None: return buy_price_range else: return ( min(buy_price_range[0], bom_price_range[0]), max(buy_price_range[1], bom_price_range[1]) ) @transaction.atomic def copy_bom_from(self, other, clear=True, **kwargs): """ Copy the BOM from another part. args: other - The part to copy the BOM from clear - Remove existing BOM items first (default=True) """ if clear: # Remove existing BOM items self.bom_items.all().delete() for bom_item in other.bom_items.all(): # If this part already has a BomItem pointing to the same sub-part, # delete that BomItem from this part first! try: existing = BomItem.objects.get(part=self, sub_part=bom_item.sub_part) existing.delete() except (BomItem.DoesNotExist): pass bom_item.part = self bom_item.pk = None bom_item.save() @transaction.atomic def copy_parameters_from(self, other, **kwargs): clear = kwargs.get('clear', True) if clear: self.get_parameters().delete() for parameter in other.get_parameters(): # If this part already has a parameter pointing to the same template, # delete that parameter from this part first! try: existing = PartParameter.objects.get(part=self, template=parameter.template) existing.delete() except (PartParameter.DoesNotExist): pass parameter.part = self parameter.pk = None parameter.save() @transaction.atomic def deep_copy(self, other, **kwargs): """ Duplicates non-field data from another part. Does not alter the normal fields of this part, but can be used to copy other data linked by ForeignKey refernce. Keyword Args: image: If True, copies Part image (default = True) bom: If True, copies BOM data (default = False) parameters: If True, copies Parameters data (default = True) """ # Copy the part image if kwargs.get('image', True): if other.image: # Reference the other image from this Part self.image = other.image # Copy the BOM data if kwargs.get('bom', False): self.copy_bom_from(other) # Copy the parameters data if kwargs.get('parameters', True): self.copy_parameters_from(other) # Copy the fields that aren't available in the duplicate form self.salable = other.salable self.assembly = other.assembly self.component = other.component self.purchaseable = other.purchaseable self.trackable = other.trackable self.virtual = other.virtual self.save() def getTestTemplates(self, required=None, include_parent=True): """ Return a list of all test templates associated with this Part. These are used for validation of a StockItem. args: required: Set to True or False to filter by "required" status include_parent: Set to True to traverse upwards """ if include_parent: tests = PartTestTemplate.objects.filter(part__in=self.get_ancestors(include_self=True)) else: tests = self.test_templates if required is not None: tests = tests.filter(required=required) return tests def getRequiredTests(self): # Return the tests which are required by this part return self.getTestTemplates(required=True) def requiredTestCount(self): return self.getRequiredTests().count() @property def attachment_count(self): """ Count the number of attachments for this part. If the part is a variant of a template part, include the number of attachments for the template part. """ return self.part_attachments.count() @property def part_attachments(self): """ Return *all* attachments for this part, potentially including attachments for template parts above this one. """ ancestors = self.get_ancestors(include_self=True) attachments = PartAttachment.objects.filter(part__in=ancestors) return attachments def sales_orders(self): """ Return a list of sales orders which reference this part """ orders = [] for line in self.sales_order_line_items.all().prefetch_related('order'): if line.order not in orders: orders.append(line.order) return orders def purchase_orders(self): """ Return a list of purchase orders which reference this part """ orders = [] for part in self.supplier_parts.all().prefetch_related('purchase_order_line_items'): for order in part.purchase_orders(): if order not in orders: orders.append(order) return orders def open_purchase_orders(self): """ Return a list of open purchase orders against this part """ return [order for order in self.purchase_orders() if order.status in PurchaseOrderStatus.OPEN] def closed_purchase_orders(self): """ Return a list of closed purchase orders against this part """ return [order for order in self.purchase_orders() if order.status not in PurchaseOrderStatus.OPEN] @property def on_order(self): """ Return the total number of items on order for this part. """ orders = self.supplier_parts.filter(purchase_order_line_items__order__status__in=PurchaseOrderStatus.OPEN).aggregate( quantity=Sum('purchase_order_line_items__quantity'), received=Sum('purchase_order_line_items__received') ) quantity = orders['quantity'] received = orders['received'] if quantity is None: quantity = 0 if received is None: received = 0 return quantity - received def get_parameters(self): """ Return all parameters for this part, ordered by name """ return self.parameters.order_by('template__name') @property def has_variants(self): """ Check if this Part object has variants underneath it. """ return self.get_all_variants().count() > 0 def get_all_variants(self): """ Return all Part object which exist as a variant under this part. """ return self.get_descendants(include_self=False) def get_related_parts(self): """ Return list of tuples for all related parts: - first value is PartRelated object - second value is matching Part object """ related_parts = [] related_parts_1 = self.related_parts_1.filter(part_1__id=self.pk) related_parts_2 = self.related_parts_2.filter(part_2__id=self.pk) for related_part in related_parts_1: # Add to related parts list related_parts.append((related_part, related_part.part_2)) for related_part in related_parts_2: # Add to related parts list related_parts.append((related_part, related_part.part_1)) return related_parts @property def related_count(self): return len(self.get_related_parts()) def attach_file(instance, filename): """ Function for storing a file for a PartAttachment Args: instance: Instance of a PartAttachment object filename: name of uploaded file Returns: path to store file, format: 'part_file__filename' """ # Construct a path to store a file attachment return os.path.join('part_files', str(instance.part.id), filename) class PartAttachment(InvenTreeAttachment): """ Model for storing file attachments against a Part object """ def getSubdir(self): return os.path.join("part_files", str(self.part.id)) part = models.ForeignKey(Part, on_delete=models.CASCADE, related_name='attachments') class PartSellPriceBreak(common.models.PriceBreak): """ Represents a price break for selling this part """ part = models.ForeignKey( Part, on_delete=models.CASCADE, related_name='salepricebreaks', limit_choices_to={'salable': True} ) class Meta: unique_together = ('part', 'quantity') class PartStar(models.Model): """ A PartStar object creates a relationship between a User and a Part. It is used to designate a Part as 'starred' (or favourited) for a given User, so that the user can track a list of their favourite parts. Attributes: part: Link to a Part object user: Link to a User object """ part = models.ForeignKey(Part, on_delete=models.CASCADE, related_name='starred_users') user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='starred_parts') class Meta: unique_together = ['part', 'user'] class PartTestTemplate(models.Model): """ A PartTestTemplate defines a 'template' for a test which is required to be run against a StockItem (an instance of the Part). The test template applies "recursively" to part variants, allowing tests to be defined in a heirarchy. Test names are simply strings, rather than enforcing any sort of structure or pattern. It is up to the user to determine what tests are defined (and how they are run). To enable generation of unique lookup-keys for each test, there are some validation tests run on the model (refer to the validate_unique function). """ def save(self, *args, **kwargs): self.clean() super().save(*args, **kwargs) def clean(self): self.test_name = self.test_name.strip() self.validate_unique() super().clean() def validate_unique(self, exclude=None): """ Test that this test template is 'unique' within this part tree. """ if not self.part.trackable: raise ValidationError({ 'part': _('Test templates can only be created for trackable parts') }) # Get a list of all tests "above" this one tests = PartTestTemplate.objects.filter( part__in=self.part.get_ancestors(include_self=True) ) # If this item is already in the database, exclude it from comparison! if self.pk is not None: tests = tests.exclude(pk=self.pk) key = self.key for test in tests: if test.key == key: raise ValidationError({ 'test_name': _("Test with this name already exists for this part") }) super().validate_unique(exclude) @property def key(self): """ Generate a key for this test """ return helpers.generateTestKey(self.test_name) part = models.ForeignKey( Part, on_delete=models.CASCADE, related_name='test_templates', limit_choices_to={'trackable': True}, ) test_name = models.CharField( blank=False, max_length=100, verbose_name=_("Test Name"), help_text=_("Enter a name for the test") ) description = models.CharField( blank=False, null=True, max_length=100, verbose_name=_("Test Description"), help_text=_("Enter description for this test") ) required = models.BooleanField( default=True, verbose_name=_("Required"), help_text=_("Is this test required to pass?") ) requires_value = models.BooleanField( default=False, verbose_name=_("Requires Value"), help_text=_("Does this test require a value when adding a test result?") ) requires_attachment = models.BooleanField( default=False, verbose_name=_("Requires Attachment"), help_text=_("Does this test require a file attachment when adding a test result?") ) class PartParameterTemplate(models.Model): """ A PartParameterTemplate provides a template for key:value pairs for extra parameters fields/values to be added to a Part. This allows users to arbitrarily assign data fields to a Part beyond the built-in attributes. Attributes: name: The name (key) of the Parameter [string] units: The units of the Parameter [string] """ def __str__(self): s = str(self.name) if self.units: s += " ({units})".format(units=self.units) return s def validate_unique(self, exclude=None): """ Ensure that PartParameterTemplates cannot be created with the same name. This test should be case-insensitive (which the unique caveat does not cover). """ super().validate_unique(exclude) try: others = PartParameterTemplate.objects.filter(name__iexact=self.name).exclude(pk=self.pk) if others.exists(): msg = _("Parameter template name must be unique") raise ValidationError({"name": msg}) except PartParameterTemplate.DoesNotExist: pass name = models.CharField(max_length=100, help_text=_('Parameter Name'), unique=True) units = models.CharField(max_length=25, help_text=_('Parameter Units'), blank=True) class PartParameter(models.Model): """ A PartParameter is a specific instance of a PartParameterTemplate. It assigns a particular parameter pair to a part. Attributes: part: Reference to a single Part object template: Reference to a single PartParameterTemplate object data: The data (value) of the Parameter [string] """ def __str__(self): # String representation of a PartParameter (used in the admin interface) return "{part} : {param} = {data}{units}".format( part=str(self.part.full_name), param=str(self.template.name), data=str(self.data), units=str(self.template.units) ) class Meta: # Prevent multiple instances of a parameter for a single part unique_together = ('part', 'template') part = models.ForeignKey(Part, on_delete=models.CASCADE, related_name='parameters', help_text=_('Parent Part')) template = models.ForeignKey(PartParameterTemplate, on_delete=models.CASCADE, related_name='instances', help_text=_('Parameter Template')) data = models.CharField(max_length=500, help_text=_('Parameter Value')) @classmethod def create(cls, part, template, data, save=False): part_parameter = cls(part=part, template=template, data=data) if save: part_parameter.save() return part_parameter class PartCategoryParameterTemplate(models.Model): """ A PartCategoryParameterTemplate creates a unique relationship between a PartCategory and a PartParameterTemplate. Multiple PartParameterTemplate instances can be associated to a PartCategory to drive a default list of parameter templates attached to a Part instance upon creation. Attributes: category: Reference to a single PartCategory object parameter_template: Reference to a single PartParameterTemplate object default_value: The default value for the parameter in the context of the selected category """ class Meta: constraints = [ UniqueConstraint(fields=['category', 'parameter_template'], name='unique_category_parameter_template_pair') ] def __str__(self): """ String representation of a PartCategoryParameterTemplate (admin interface) """ if self.default_value: return f'{self.category.name} | {self.parameter_template.name} | {self.default_value}' else: return f'{self.category.name} | {self.parameter_template.name}' category = models.ForeignKey(PartCategory, on_delete=models.CASCADE, related_name='parameter_templates', help_text=_('Part Category')) parameter_template = models.ForeignKey(PartParameterTemplate, on_delete=models.CASCADE, related_name='part_categories', help_text=_('Parameter Template')) default_value = models.CharField(max_length=500, blank=True, help_text=_('Default Parameter Value')) class BomItem(models.Model): """ A BomItem links a part to its component items. A part can have a BOM (bill of materials) which defines which parts are required (and in what quantity) to make it. Attributes: part: Link to the parent part (the part that will be produced) sub_part: Link to the child part (the part that will be consumed) quantity: Number of 'sub_parts' consumed to produce one 'part' optional: Boolean field describing if this BomItem is optional reference: BOM reference field (e.g. part designators) overage: Estimated losses for a Build. Can be expressed as absolute value (e.g. '7') or a percentage (e.g. '2%') note: Note field for this BOM item checksum: Validation checksum for the particular BOM line item """ def save(self, *args, **kwargs): self.clean() super().save(*args, **kwargs) def get_absolute_url(self): return reverse('bom-item-detail', kwargs={'pk': self.id}) # A link to the parent part # Each part will get a reverse lookup field 'bom_items' part = models.ForeignKey(Part, on_delete=models.CASCADE, related_name='bom_items', help_text=_('Select parent part'), limit_choices_to={ 'assembly': True, }) # A link to the child item (sub-part) # Each part will get a reverse lookup field 'used_in' sub_part = models.ForeignKey(Part, on_delete=models.CASCADE, related_name='used_in', help_text=_('Select part to be used in BOM'), limit_choices_to={ 'component': True, }) # Quantity required quantity = models.DecimalField(default=1.0, max_digits=15, decimal_places=5, validators=[MinValueValidator(0)], help_text=_('BOM quantity for this BOM item')) optional = models.BooleanField(default=False, help_text=_("This BOM item is optional")) overage = models.CharField(max_length=24, blank=True, validators=[validators.validate_overage], help_text=_('Estimated build wastage quantity (absolute or percentage)') ) reference = models.CharField(max_length=500, blank=True, help_text=_('BOM item reference')) # Note attached to this BOM line item note = models.CharField(max_length=500, blank=True, help_text=_('BOM item notes')) checksum = models.CharField(max_length=128, blank=True, help_text=_('BOM line checksum')) def get_item_hash(self): """ Calculate the checksum hash of this BOM line item: The hash is calculated from the following fields: - Part.full_name (if the part name changes, the BOM checksum is invalidated) - Quantity - Reference field - Note field """ # Seed the hash with the ID of this BOM item hash = hashlib.md5(str(self.id).encode()) # Update the hash based on line information hash.update(str(self.sub_part.id).encode()) hash.update(str(self.sub_part.full_name).encode()) hash.update(str(self.quantity).encode()) hash.update(str(self.note).encode()) hash.update(str(self.reference).encode()) return str(hash.digest()) def validate_hash(self, valid=True): """ Mark this item as 'valid' (store the checksum hash). Args: valid: If true, validate the hash, otherwise invalidate it (default = True) """ if valid: self.checksum = str(self.get_item_hash()) else: self.checksum = '' self.save() @property def is_line_valid(self): """ Check if this line item has been validated by the user """ # Ensure an empty checksum returns False if len(self.checksum) == 0: return False return self.get_item_hash() == self.checksum def clean(self): """ Check validity of the BomItem model. Performs model checks beyond simple field validation. - A part cannot refer to itself in its BOM - A part cannot refer to a part which refers to it - If the "sub_part" is trackable, then the "part" must be trackable too! """ # If the sub_part is 'trackable' then the 'quantity' field must be an integer try: if self.sub_part.trackable: if not self.quantity == int(self.quantity): raise ValidationError({ "quantity": _("Quantity must be integer value for trackable parts") }) # Force the upstream part to be trackable if the sub_part is trackable if not self.part.trackable: self.part.trackable = True self.part.clean() self.part.save() except Part.DoesNotExist: pass # Check for circular BOM references self.sub_part.checkAddToBOM(self.part) class Meta: verbose_name = _("BOM Item") # Prevent duplication of parent/child rows unique_together = ('part', 'sub_part') def __str__(self): return "{n} x {child} to make {parent}".format( parent=self.part.full_name, child=self.sub_part.full_name, n=helpers.decimal2string(self.quantity)) def available_stock(self): """ Return the available stock items for the referenced sub_part """ query = self.sub_part.stock_items.filter(StockModels.StockItem.IN_STOCK_FILTER).aggregate( available=Coalesce(Sum('quantity'), 0) ) return query['available'] def get_overage_quantity(self, quantity): """ Calculate overage quantity """ # Most of the time overage string will be empty if len(self.overage) == 0: return 0 overage = str(self.overage).strip() # Is the overage a numerical value? try: ovg = float(overage) if ovg < 0: ovg = 0 return ovg except ValueError: pass # Is the overage a percentage? if overage.endswith('%'): overage = overage[:-1].strip() try: percent = float(overage) / 100.0 if percent > 1: percent = 1 if percent < 0: percent = 0 # Must be represented as a decimal percent = Decimal(percent) return float(percent * quantity) except ValueError: pass # Default = No overage return 0 def get_required_quantity(self, build_quantity): """ Calculate the required part quantity, based on the supplier build_quantity. Includes overage estimate in the returned value. Args: build_quantity: Number of parts to build Returns: Quantity required for this build (including overage) """ # Base quantity requirement base_quantity = self.quantity * build_quantity # Overage requiremet ovrg_quantity = self.get_overage_quantity(base_quantity) required = float(base_quantity) + float(ovrg_quantity) return required @property def price_range(self): """ Return the price-range for this BOM item. """ prange = self.sub_part.get_price_range(self.quantity) if prange is None: return prange pmin, pmax = prange if pmin == pmax: return decimal2string(pmin) # Convert to better string representation pmin = decimal2string(pmin) pmax = decimal2string(pmax) return "{pmin} to {pmax}".format(pmin=pmin, pmax=pmax) class PartRelated(models.Model): """ Store and handle related parts (eg. mating connector, crimps, etc.) """ part_1 = models.ForeignKey(Part, related_name='related_parts_1', on_delete=models.DO_NOTHING) part_2 = models.ForeignKey(Part, related_name='related_parts_2', on_delete=models.DO_NOTHING, help_text=_('Select Related Part')) def __str__(self): return f'{self.part_1} <--> {self.part_2}' def validate(self, part_1, part_2): ''' Validate that the two parts relationship is unique ''' validate = True parts = Part.objects.all() related_parts = PartRelated.objects.all() # Check if part exist and there are not the same part if (part_1 in parts and part_2 in parts) and (part_1.pk != part_2.pk): # Check if relation exists already for relation in related_parts: if (part_1 == relation.part_1 and part_2 == relation.part_2) \ or (part_1 == relation.part_2 and part_2 == relation.part_1): validate = False break else: validate = False return validate def clean(self): ''' Overwrite clean method to check that relation is unique ''' validate = self.validate(self.part_1, self.part_2) if not validate: error_message = _('Error creating relationship: check that ' 'the part is not related to itself ' 'and that the relationship is unique') raise ValidationError(error_message)