"""Polymorphic Asset models - core of the new asset architecture.""" from shopdb.extensions import db from .base import BaseModel, SoftDeleteMixin, AuditMixin class AssetType(BaseModel): """ Registry of asset categories. Each type maps to a plugin-owned extension table. Examples: machine, computer, network_device, printer """ __tablename__ = 'assettypes' assettypeid = db.Column(db.Integer, primary_key=True) assettype = db.Column( db.String(50), unique=True, nullable=False, comment='Category name: machine, computer, network_device, printer' ) pluginname = db.Column( db.String(100), nullable=True, comment='Plugin that owns this type' ) tablename = db.Column( db.String(100), nullable=True, comment='Extension table name for this type' ) description = db.Column(db.Text) icon = db.Column(db.String(50), comment='Icon name for UI') color = db.Column(db.String(20), comment='CSS color for UI/map markers') def __repr__(self): return f"" class AssetStatus(BaseModel): """Asset status options.""" __tablename__ = 'assetstatuses' statusid = db.Column(db.Integer, primary_key=True) status = db.Column(db.String(50), unique=True, nullable=False) description = db.Column(db.Text) color = db.Column(db.String(20), comment='CSS color for UI') def __repr__(self): return f"" class Asset(BaseModel, SoftDeleteMixin, AuditMixin): """ Core asset model - minimal shared fields. Category-specific data lives in plugin extension tables (machines, computers, network_devices, printers). The assetid matches original machineid for migration compatibility. """ __tablename__ = 'assets' assetid = db.Column(db.Integer, primary_key=True) # Identification assetnumber = db.Column( db.String(50), unique=True, nullable=False, index=True, comment='Business identifier (e.g., CMM01, G5QX1GT3ESF)' ) name = db.Column( db.String(100), comment='Display name/alias' ) gaugelabreference = db.Column( db.String(50), index=True, comment='Gauge lab asset reference (authoritative tag the gauge lab ' 'assigns to machines); distinct from assetnumber' ) maintenancereference = db.Column( db.String(50), index=True, comment='Maintenance system asset reference; distinct from assetnumber' ) serialnumber = db.Column( db.String(100), index=True, comment='Hardware serial number' ) # Classification assettypeid = db.Column( db.Integer, db.ForeignKey('assettypes.assettypeid'), nullable=False ) statusid = db.Column( db.Integer, db.ForeignKey('assetstatuses.statusid'), default=1, comment='In Use, Spare, Retired, etc.' ) # Location and organization locationid = db.Column( db.Integer, db.ForeignKey('locations.locationid'), nullable=True ) businessunitid = db.Column( db.Integer, db.ForeignKey('businessunits.businessunitid'), nullable=True ) # Floor map position (ADR-001: asset-specific override; nullable) mapx = db.Column(db.Integer, comment='X coordinate on floor map (ADR-001)') mapy = db.Column(db.Integer, comment='Y coordinate on floor map (ADR-001)') # Notes notes = db.Column(db.Text, nullable=True) # Relationships assettype = db.relationship('AssetType', backref='assets') status = db.relationship('AssetStatus', backref='assets') location = db.relationship('Location', backref='assets') businessunit = db.relationship('BusinessUnit', backref='assets') # Communications (one-to-many) - will be migrated to use assetid communications = db.relationship( 'Communication', foreign_keys='Communication.assetid', backref='asset', cascade='all, delete-orphan', lazy='dynamic' ) # Indexes __table_args__ = ( db.Index('idx_asset_type_bu', 'assettypeid', 'businessunitid'), db.Index('idx_asset_location', 'locationid'), db.Index('idx_asset_active', 'isactive'), db.Index('idx_asset_status', 'statusid'), ) def __repr__(self): return f"" @property def display_name(self): """Get display name (name if set, otherwise assetnumber).""" return self.name or self.assetnumber @property def primary_ip(self): """Get primary IP address from communications.""" comm = self.communications.filter_by( isprimary=True, comtypeid=1 # IP type ).first() if comm: return comm.ipaddress # Fall back to any IP comm = self.communications.filter_by(comtypeid=1).first() return comm.ipaddress if comm else None def get_inherited_location(self): """ Get location data from a related asset if this asset has none. Returns dict with locationid, location_name, mapx, mapy, and inherited_from (assetnumber of source asset) if location was inherited. Returns None if no location data available. """ if self.locationid is not None or (self.mapx is not None and self.mapy is not None): return None related_assets = [] if hasattr(self, 'incoming_relationships'): for rel in self.incoming_relationships: if rel.sourceasset and rel.isactive: related_assets.append(rel.sourceasset) if hasattr(self, 'outgoing_relationships'): for rel in self.outgoing_relationships: if rel.targetasset and rel.isactive: related_assets.append(rel.targetasset) for related in related_assets: if related.locationid is not None or (related.mapx is not None and related.mapy is not None): return { 'locationid': related.locationid, 'locationname': related.location.locationname if related.location else None, 'mapx': related.mapx, 'mapy': related.mapy, 'inheritedfrom': related.assetnumber } return None def to_dict(self, include_type_data=False, include_inherited_location=True): """ Convert model to dictionary. Args: include_type_data: If True, include category-specific data from extension table include_inherited_location: If True, include location from related assets when missing """ result = super().to_dict() # Add related object names for convenience if self.assettype: result['assettypename'] = self.assettype.assettype result['assettypecolor'] = getattr(self.assettype, 'color', None) if self.status: result['statusname'] = self.status.status result['statuscolor'] = self.status.color if self.location: result['locationname'] = self.location.locationname if self.businessunit: result['businessunitname'] = self.businessunit.businessunit # Add plugin-specific ID for navigation purposes if hasattr(self, 'machine') and self.machine: result['pluginid'] = self.machine.machineid elif hasattr(self, 'computer') and self.computer: result['pluginid'] = self.computer.computerid elif hasattr(self, 'network_device') and self.network_device: result['pluginid'] = self.network_device.networkdeviceid elif hasattr(self, 'printer') and self.printer: result['pluginid'] = self.printer.printerid elif hasattr(self, 'measuringtool') and self.measuringtool: result['pluginid'] = self.measuringtool.measuringtoolid # Include inherited location if this asset has no location data if include_inherited_location: inherited = self.get_inherited_location() if inherited: result['inheritedlocation'] = inherited # Also set the location fields if they're missing if result.get('locationid') is None: result['locationid'] = inherited['locationid'] result['locationname'] = inherited['locationname'] if result.get('mapx') is None: result['mapx'] = inherited['mapx'] if result.get('mapy') is None: result['mapy'] = inherited['mapy'] # Operation/short code of the resolved location (own or inherited). # Derived from the location name's leading token; labels can encode a # tool's inspection operation instead of the tool. None when unplaced. from .location import derive_locationcode result['locationcode'] = derive_locationcode(result.get('locationname')) # Include extension data if requested if include_type_data: ext_data = self._get_extension_data() if ext_data: result['typedata'] = ext_data return result def _get_extension_data(self): """Get category-specific data from extension table.""" # Check for machine extension if hasattr(self, 'machine') and self.machine: return self.machine.to_dict() # Check for computer extension if hasattr(self, 'computer') and self.computer: return self.computer.to_dict() # Check for network_device extension if hasattr(self, 'network_device') and self.network_device: return self.network_device.to_dict() # Check for printer extension if hasattr(self, 'printer') and self.printer: return self.printer.to_dict() # Check for measuring-tool extension if hasattr(self, 'measuringtool') and self.measuringtool: return self.measuringtool.to_dict() return None