"""Computers plugin main class.""" import json import logging from pathlib import Path from typing import List, Dict, Optional, Type from flask import Flask, Blueprint import click from shopdb.plugins.base import BasePlugin, PluginMeta from shopdb.api import db, AssetType from .models import Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess from .api import computers_bp logger = logging.getLogger(__name__) # Marker stamped on PC->printer links this collector creates. The stale-link # archive only touches rows carrying this label, so manually-created printer # relationships are never removed by a collector push. Stored in the # assetrelationships.label column (no origin column exists; see BUILD notes). PRINTER_LINK_ORIGIN = 'collector:printers' # Marker stamped on PC->measuringtool "controls" links this collector creates # for metrology PCs (CMM, Keyence, Genspect, wax-and-trace). Same stale-link # discipline as PRINTER_LINK_ORIGIN: only rows carrying this label are archived # by a collector push, so hand-made tool links survive. MEASURINGTOOL_LINK_ORIGIN = 'collector:measuringtool' # Marker stamped on the PC->machine "controls" link built from the reported # machine number. Same discipline as the two above: only rows carrying this # label are archived by a collector push, so a link made by hand is never # touched. MACHINE_LINK_ORIGIN = 'collector:machine' # Marker stamped on both links a part-marker PC produces: PC controls marker, # and marker partof the operation it serves. Same archive discipline again. PARTMARKER_LINK_ORIGIN = 'collector:partmarker' # How long a PC holding a machine may go without reporting before a second PC # claiming that machine is treated as its replacement. A swap resolves itself # within a day; a PC off overnight or behind a network outage keeps its bay. # Shorter than this and a spare imaged on the bench could steal a live machine. MACHINE_CLAIM_QUIET_HOURS = 24 class ComputersPlugin(BasePlugin): """ Computers plugin - manages PC, server, and workstation assets. Computers include shopfloor PCs, engineer workstations, servers, etc. Uses the new Asset architecture with Computer extension table. """ def __init__(self): self._manifest = self._load_manifest() def _load_manifest(self) -> Dict: """Load plugin manifest from JSON file.""" manifestpath = Path(__file__).parent / 'manifest.json' if manifestpath.exists(): with open(manifestpath, 'r') as f: return json.load(f) return {} @property def meta(self) -> PluginMeta: """Return plugin metadata.""" return PluginMeta( name=self._manifest.get('name', 'computers'), version=self._manifest.get('version', '1.0.0'), description=self._manifest.get( 'description', 'Computer management for PCs, servers, and workstations' ), author=self._manifest.get('author', 'ShopDB Team'), dependencies=self._manifest.get('dependencies', []), core_version=self._manifest.get('core_version', '>=1.0.0'), api_prefix=self._manifest.get('api_prefix', '/api/computers'), ) def get_blueprint(self) -> Optional[Blueprint]: """Return Flask Blueprint with API routes.""" return computers_bp def get_models(self) -> List[Type]: """Return list of SQLAlchemy model classes.""" return [Computer, ComputerType, ComputerInstalledApp, AccessProtocol, ComputerAccess] def init_app(self, app: Flask, db_instance) -> None: """Initialize plugin with Flask app.""" logger.info(f"Computers plugin initialized (v{self.meta.version})") # -- ADR-006 collector contract ----------------------------------------- def get_collector_schema(self) -> Optional[Dict]: """Schema for the PC collector payload (matched by hostname). Aligns with the GE-Enforce status shape (transport may change), using the project naming convention (lowercase concatenated). The caller maps its own field names to these. """ return { 'identityfield': 'hostname', 'fields': { 'hostname': {'type': 'string', 'required': True}, 'machinenumber': {'type': 'string'}, # The instrument this PC drives, named by the enrollment file # measuringtool-id.txt. SEPARATE from machinenumber on purpose: # "which bay is this" and "which instrument is this" are # different facts, and machinenumber is what GE-Enforce # TargetMachineNumbers gates on - repointing it at a tool would # silently stop every bay-gated manifest entry matching. 'measuringtoolid': {'type': 'string'}, 'pctype': {'type': 'string'}, 'pcsubtype': {'type': 'string'}, 'serialnumber': {'type': 'string'}, 'loggedinuser': {'type': 'string'}, 'lastboottime': {'type': 'string', 'format': 'date-time'}, 'lastcheckin': {'type': 'string', 'format': 'date-time'}, 'ipaddress': {'type': 'string'}, 'vendorname': {'type': 'string'}, 'modelnumber': {'type': 'string'}, 'osname': {'type': 'string'}, 'installedsoftware': { 'type': 'array', 'items': {'name': 'string', 'version': 'string'}, }, # Printer identifiers reported by the GE-Enforce side, which # runs Get-CimInstance Win32_Printer and marks the default with # the Default flag. An identifier is the printer's windows name, # share name, hostname, or port/IP; the collector resolves it # flexibly to a printer asset. Both optional. Presence of either # key drives the PC->printer relationship sync (and stale-link # archive); absence leaves existing printer links untouched. 'defaultprinter': { 'type': 'string', 'description': ('Default printer identifier (Win32_Printer ' 'with Default=true): windows name, share ' 'name, hostname, or port IP.'), }, 'printers': { 'type': 'array', 'items': {'type': 'string'}, 'description': ('All installed network printer identifiers ' '(Win32_Printer): windows name / share / ' 'hostname / IP. Unresolved -> warning.'), }, 'accessprotocols': { 'type': 'array', 'items': {'type': 'string'}, 'description': ('Remote-access protocols this PC actually ' 'exposes, by catalog name (VNC, WinRM, ' 'RDP). Presence of the key drives the sync: ' 'reported protocols are activated and ' 'catalogued ones not reported are ' 'deactivated. Omit the key entirely to ' 'leave existing rows alone - most came from ' 'the legacy isvnc/iswinrm migration.'), }, }, } def get_settings_defaults(self) -> List[dict]: """Settings this plugin owns. The framework seeds these at install, at enable, and on every `flask plugin upgrade-all`, so a key added in a later version reaches a site that installed an earlier one. """ return [ { 'key': 'computers_quietreporthours', 'value': '24', 'valuetype': 'integer', 'category': 'computers', 'description': 'Hours without a collector report before a PC ' 'is listed as not reporting on the dashboard.', }, { 'key': 'computers_machinelink_alerts', 'value': 'false', 'valuetype': 'boolean', 'category': 'computers', 'description': 'Email and webhook alerts when a PC takes over ' 'a machine or claims one another PC still runs. ' 'Off while machine numbers are shared between ' 'PCs; the collector response still warns.', }, ] def _machinelink_alerts_enabled(self): """Whether the PC-to-machine alerts may be sent. Off by default, deliberately. A site may legitimately run several PCs on one machine number - part markers do at the reference site - and there both the handover and the contested case fire on normal, correct data, which is noise rather than news. Turn it on at a site where a machine number means exactly one PC. The links, the warnings in the collector response, and the archived history all continue regardless. Only the sending is gated. """ from shopdb.api import Setting setting = Setting.query.filter_by( key='computers_machinelink_alerts').first() if not setting: return False return (setting.value or '').strip().lower() in ('true', '1', 'yes') def apply_collector_payload(self, payload: Dict) -> Dict: """Idempotent upsert of a PC from a collector payload (by hostname).""" from datetime import datetime, timezone from shopdb.api import ( Asset, Application, Communication, CommunicationType, Vendor, Model, OperatingSystem, ) from .pctypemap import pctype_mapping warnings = [] hostname = (payload.get('hostname') or '').strip() if not hostname: raise ValueError('hostname is required') # The machine number identifies the MACHINE, not this PC. It is # reported so the PC can be related to its machine; it is deliberately # NOT used as the PC's assetnumber. assets.assetnumber is uniquely # indexed and the machine already owns that value, so assigning it here # raised "Duplicate entry '3015' for key 'ix_assets_assetnumber'" and # returned 500 to the bay - forever, since every retry did the same # thing. The convention this restores is what the data already shows: # of 289 computers, none has a numeric assetnumber and 214 use their # hostname. machinenumber = (payload.get('machinenumber') or '').strip() if machinenumber in ('', '9999'): machinenumber = None comp = Computer.query.filter(Computer.hostname.ilike(hostname)).first() if not comp: comp = (Computer.query.join(Asset, Asset.assetid == Computer.assetid) .filter(Asset.assetnumber.ilike(hostname)).first()) action = 'updated' if not comp: atype = AssetType.query.filter_by(assettype='computer').first() # statusid=1 is the first seeded asset status ("In Use"); a # collector-discovered PC is by definition in use. asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid, statusid=1) db.session.add(asset) db.session.flush() comp = Computer(assetid=asset.assetid, hostname=hostname) db.session.add(comp) db.session.flush() action = 'created' # NOTE: an existing PC's assetnumber is left alone. Overwriting it with # the machine number renamed the PC onto the machine's identifier, which # either collided with the unique index or silently changed how that PC # is identified everywhere else. comp.lastreporteddate = datetime.now(timezone.utc).replace(tzinfo=None) if payload.get('lastboottime'): try: comp.lastboottime = datetime.fromisoformat( payload['lastboottime'].replace('Z', '+00:00')) except (ValueError, AttributeError): warnings.append('lastboottime not parseable') loggedinuser = payload.get('loggedinuser') or payload.get('currentuser') if loggedinuser: comp.loggedinuser = loggedinuser if payload.get('serialnumber') and comp.asset: comp.asset.serialnumber = payload['serialnumber'] # pc-type -> ComputerType via the configurable settings mapping. pctype = (payload.get('pctype') or '').strip() if pctype: from .models import ComputerType mapped = pctype_mapping().get(pctype) if not mapped: warnings.append(f'no ComputerType mapping for pctype: {pctype}') else: ctype = ComputerType.query.filter_by(computertype=mapped).first() if ctype: comp.computertypeid = ctype.computertypeid else: warnings.append(f'mapped ComputerType not found: {mapped}') # Vendor / model are free vocab - create if missing. vendorname = (payload.get('vendorname') or '').strip() vendor = None if vendorname: vendor = Vendor.query.filter(Vendor.vendor.ilike(vendorname)).first() if not vendor: vendor = Vendor(vendor=vendorname) db.session.add(vendor) db.session.flush() comp.vendorid = vendor.vendorid modelnumber = (payload.get('modelnumber') or '').strip() if modelnumber: model_query = Model.query.filter(Model.modelnumber.ilike(modelnumber)) if vendor: model_query = model_query.filter(Model.vendorid == vendor.vendorid) model = model_query.first() if not model: model = Model(modelnumber=modelnumber, vendorid=vendor.vendorid if vendor else None) db.session.add(model) db.session.flush() comp.modelnumberid = model.modelnumberid # OS is a controlled vocab - look up only, warn if unknown. osname = (payload.get('osname') or '').strip() if osname: os_row = OperatingSystem.query.filter( OperatingSystem.osname.ilike(osname)).first() if os_row: comp.osid = os_row.osid else: warnings.append(f'unknown operating system: {osname}') if payload.get('pcsubtype'): warnings.append('pcsubtype received but not stored (no model field)') if payload.get('ipaddress'): ip_comtype = CommunicationType.query.filter_by(comtype='IP').first() primary = Communication.query.filter_by( assetid=comp.assetid, isprimary=True).first() if primary: primary.ipaddress = payload['ipaddress'] elif ip_comtype: db.session.add(Communication( assetid=comp.assetid, comtypeid=ip_comtype.comtypeid, ipaddress=payload['ipaddress'], isprimary=True)) for app_data in payload.get('installedsoftware', []) or []: name = app_data.get('name') if not name: continue app = Application.query.filter(Application.appname.ilike(name)).first() if not app: warnings.append(f'unknown application: {name}') continue installed = ComputerInstalledApp.query.filter_by( computerid=comp.computerid, appid=app.appid).first() version = app_data.get('version') if installed: installed.installedversion = version installed.isactive = True else: db.session.add(ComputerInstalledApp( computerid=comp.computerid, appid=app.appid, installedversion=version)) # Remote-access protocol sync (only when the payload carried the key). accessprotocols = self._sync_access_protocols(comp, payload, warnings) # Part-marker PCs get a marker asset of their own, which is what the # machine number then hangs off. Done BEFORE the machine link because a # marker PC must not also claim the operation directly: several markers # serve one operation number, so direct claims would fight over it. The # marker is partof the operation and control propagates along that rail. partmarkers = self._sync_partmarker(comp, pctype, machinenumber, warnings) # PC -> machine link from the reported machine number. if partmarkers: machinelinks = [] else: machinelinks = self._sync_machine_link(comp, machinenumber, warnings) # Printer relationship sync (only when the payload carried printer data). printerlinks = self._sync_printer_links(comp.asset, payload, warnings) # Measuring-tool sync: metrology PCs (CMM/Keyence/Genspect/wax-trace) # get an attached MeasuringTool asset auto-created and linked. measuringtoollinks = self._sync_measuringtool_link( comp.asset, pctype, hostname, warnings, measuringtoolid=payload.get('measuringtoolid'), machinenumber=machinenumber) db.session.commit() return { 'action': action, 'assetid': comp.assetid, 'identityvalue': hostname, 'warnings': warnings, 'extra': { 'printerlinks': printerlinks, 'printerlinkcount': len(printerlinks), 'measuringtoollinks': measuringtoollinks, 'measuringtoollinkcount': len(measuringtoollinks), 'accessprotocols': accessprotocols, 'machinelinks': machinelinks, 'partmarkers': partmarkers, }, } # -- printer relationship sync ----------------------------------------- def _sync_access_protocols(self, comp, payload, warnings): """Idempotently sync a PC's remote-access protocols from the collector. Payload key 'accessprotocols' is a list of protocol NAMES as they appear in the accessprotocols catalog ('VNC', 'WinRM', 'RDP'), matched case-insensitively. An unknown name warns and is skipped rather than creating a protocol: the catalog is admin-managed on purpose, so a typo on one bay must not invent a protocol for the whole site. Presence of the key drives the sync, exactly like the printer links: a reported protocol is activated, and a catalogued protocol the PC did NOT report is deactivated (not deleted, so a port override survives a temporary outage). A payload with no 'accessprotocols' key leaves every existing row untouched - most PCs' rows came from the legacy isvnc/iswinrm migration and must not be wiped by a collector that simply does not report them yet. Returns the list of active protocol names after the sync. """ from plugins.computers.models import AccessProtocol, ComputerAccess if 'accessprotocols' not in payload: return [] reported = payload.get('accessprotocols') or [] if not isinstance(reported, list): warnings.append('accessprotocols must be a list of protocol names') return [] wanted = set() for name in reported: name = str(name or '').strip() if not name: continue protocol = AccessProtocol.query.filter( AccessProtocol.name.ilike(name)).first() if not protocol: warnings.append('unknown access protocol: {}'.format(name)) continue wanted.add(protocol.protocolid) existing = {row.protocolid: row for row in ComputerAccess.query.filter_by(computerid=comp.computerid).all()} for protocolid in wanted: row = existing.get(protocolid) if row: row.isactive = True else: db.session.add(ComputerAccess( computerid=comp.computerid, protocolid=protocolid, isactive=True)) # Deactivate what the PC no longer exposes. Kept as rows so a manual # portoverride is not lost the first time a service is briefly down. for protocolid, row in existing.items(): if protocolid not in wanted: row.isactive = False names = [p.name for p in AccessProtocol.query.filter( AccessProtocol.protocolid.in_(wanted)).all()] if wanted else [] return sorted(names) def _sync_printer_links(self, pcasset, payload, warnings): """Idempotently sync PC->printer relationships from collector printer data. Default printer -> 'defaultprinter' (directional). Other reported printers -> 'connectedto' (symmetric). Resolves each identifier to a printer asset by windows name / share / hostname / asset number / name or a communications IP. Unresolved identifiers add a warning and never fail the push. Stale-link archive: on each push, collector-tagged links (label == PRINTER_LINK_ORIGIN) whose (target, type) pair is not in the reported desired set are set inactive. Only tagged rows are touched, so manual links survive. Runs only when the payload carried a printer key ('defaultprinter' or 'printers'); a PC that reports without printer data keeps its existing links. Returns the desired-link list (created + kept). """ from shopdb.api import ( AssetRelationship, RelationshipType, Asset, Communication) has_default = 'defaultprinter' in payload has_list = 'printers' in payload if not has_default and not has_list: return [] try: from plugins.printers.models import Printer except ImportError: warnings.append('printers plugin unavailable; printer links skipped') return [] dp_type = RelationshipType.query.filter_by( relationshiptype='defaultprinter').first() ct_type = RelationshipType.query.filter_by( relationshiptype='connectedto').first() if not dp_type or not ct_type: warnings.append('printer relationship types missing; ' 'run flask seed reference-data') return [] def resolve(identifier): # first match wins: printer text identity, then a printer IP. ident = (identifier or '').strip() if not ident: return None printer = ( Printer.query.join(Asset, Asset.assetid == Printer.assetid) .filter(Asset.isactive == True) .filter(db.or_( Printer.windowsname.ilike(ident), Printer.hostname.ilike(ident), Printer.sharename.ilike(ident), Asset.assetnumber.ilike(ident), Asset.name.ilike(ident), )).first()) if printer: return printer.asset comm = ( db.session.query(Communication) .join(Printer, Printer.assetid == Communication.assetid) .filter(Communication.ipaddress == ident) .first()) if comm: return db.session.get(Asset, comm.assetid) return None pcid = pcasset.assetid desired = set() # (targetassetid, relationshiptypeid) to keep printerlinks = [] default_id = None default_ident = (payload.get('defaultprinter') or '').strip() if default_ident: target = resolve(default_ident) if target: default_id = target.assetid desired.add((default_id, dp_type.relationshiptypeid)) self._sync_one(pcid, default_id, dp_type) printerlinks.append({'assetid': default_id, 'relationshiptype': 'defaultprinter'}) else: warnings.append(f'unresolved default printer: {default_ident}') for ident in payload.get('printers') or []: target = resolve(ident) if not target: warnings.append(f'unresolved printer: {ident}') continue if target.assetid == default_id: continue # already the default link desired.add((target.assetid, ct_type.relationshiptypeid)) self._sync_one(pcid, target.assetid, ct_type) printerlinks.append({'assetid': target.assetid, 'relationshiptype': 'connectedto'}) # Archive collector-tagged links no longer reported (manual links, with # a NULL/other label, are never matched here). collector_rows = AssetRelationship.query.filter( AssetRelationship.sourceassetid == pcid, AssetRelationship.relationshiptypeid.in_( [dp_type.relationshiptypeid, ct_type.relationshiptypeid]), AssetRelationship.isactive == True, AssetRelationship.label == PRINTER_LINK_ORIGIN, ).all() for rel in collector_rows: if (rel.targetassetid, rel.relationshiptypeid) not in desired: rel.isactive = False return printerlinks def _sync_one(self, pcid, printerid, reltype): """Reactivate or create one collector PC->printer link (idempotent).""" from shopdb.api import AssetRelationship existing = AssetRelationship.query.filter_by( sourceassetid=pcid, targetassetid=printerid, relationshiptypeid=reltype.relationshiptypeid).first() if existing: # do not re-stamp label: a pre-existing manual row stays manual. if not existing.isactive: existing.isactive = True return existing rel = AssetRelationship( sourceassetid=pcid, targetassetid=printerid, relationshiptypeid=reltype.relationshiptypeid, label=PRINTER_LINK_ORIGIN) db.session.add(rel) return rel # -- measuring-tool sync ----------------------------------------------- def _sync_machine_link(self, comp, machinenumber, warnings): """Link a PC to the machine it drives, from the reported machine number. Until this existed the machine number was collected and then discarded, so a replaced PC never took over its bay: the retired PC kept the link and the new one got none. Everything that walks PC->machine (the warranty machine column, the DNC info card) therefore pointed at hardware that had been pulled out. ARCHIVES, never deletes. A superseded link stays with isactive=False so "which PC ran 3015 in June" is still answerable. Status on the old PC is deliberately NOT changed: the collector cannot tell whether it was shelved, broken or re-imaged for another bay, and guessing would overwrite whatever a person deliberately set. A second PC reporting a machine another PC already holds is a CLAIM, not proof of replacement. A PC imaged on the bench for machine 3010 carries that number before it ever reaches the floor, and treating the claim as a handover made the two PCs trade the link back and forth at collector cadence, alerting on every pass. The incumbent therefore keeps the machine while it is still alive, and the challenger is recorded as a dormant link. See _incumbent_has_yielded for what alive means. """ from shopdb.api import AssetRelationship, RelationshipType, Asset pcasset = comp.asset if comp else None if not machinenumber or not pcasset: return [] controls = RelationshipType.query.filter_by( relationshiptype='controls').first() if not controls: warnings.append("'controls' relationship type missing; " 'run flask seed reference-data') return [] machine = self._asset_by_number(machinenumber) if not machine: # Reported a machine ShopDB does not know. Warn rather than invent # an asset: a mistyped number would create a machine nobody can # account for. warnings.append( 'no asset for machine number {!r}; PC not linked'.format( machinenumber)) return [] if machine.assetid == pcasset.assetid: return [] # This PC's own collector links: the one for the reported machine is # settled below, any other is archived (the PC moved bays). mine = AssetRelationship.query.filter( AssetRelationship.sourceassetid == pcasset.assetid, AssetRelationship.relationshiptypeid == controls.relationshiptypeid, AssetRelationship.label == MACHINE_LINK_ORIGIN, ).all() link = None for rel in mine: if rel.targetassetid == machine.assetid: link = rel else: rel.isactive = False # Whoever actively holds this machine now, if it is not this PC. held = AssetRelationship.query.filter( AssetRelationship.targetassetid == machine.assetid, AssetRelationship.relationshiptypeid == controls.relationshiptypeid, AssetRelationship.label == MACHINE_LINK_ORIGIN, AssetRelationship.sourceassetid != pcasset.assetid, AssetRelationship.isactive.is_(True), ).all() blocking = [rel for rel in held if not self._incumbent_has_yielded(rel.sourceassetid)] if blocking: # Contested. The incumbent is still reporting and still In Use, so # this is a claim on a bay it has not taken over yet. Record the # claim dormant and leave the live link where it is; the dormant row # is also the marker that says this was already announced, which is # what stops an alert on every report. names = ', '.join(self._assetname(rel.sourceassetid) for rel in blocking) warnings.append( 'machine {} is still held by {}; {} claim recorded but not ' 'linked'.format(machine.assetnumber, names, comp.hostname)) if link is None: db.session.add(AssetRelationship( sourceassetid=pcasset.assetid, targetassetid=machine.assetid, relationshiptypeid=controls.relationshiptypeid, label=MACHINE_LINK_ORIGIN, isactive=False)) self._alert_machine_contested(names, comp, machine) else: link.isactive = False return [{'assetid': machine.assetid, 'machinenumber': machine.assetnumber, 'contestedby': names, 'superseded': 0}] # Uncontested: take the machine. if link is None: link = AssetRelationship( sourceassetid=pcasset.assetid, targetassetid=machine.assetid, relationshiptypeid=controls.relationshiptypeid, label=MACHINE_LINK_ORIGIN, isactive=True) db.session.add(link) else: link.isactive = True # Anything still holding it has yielded: gone quiet, or taken off In # Use by a person. That is a replacement, so archive and say so. for rel in held: rel.isactive = False oldname = self._assetname(rel.sourceassetid) warnings.append( 'machine {} was taken over from {}; check that PC'.format( machine.assetnumber, oldname)) self._alert_pc_superseded(oldname, comp, machine) return [{'assetid': machine.assetid, 'machinenumber': machine.assetnumber, 'superseded': len(held)}] def _sync_partmarker(self, comp, pctype, machinenumber, warnings): """Give a part-marker PC a marker asset of its own, under its operation. Several Telesis markers serve one operation number - 0613, 0615 and WJPRT each have more than one - so treating the operation as the marker collapsed separate devices into one record. Their configs, which differ by COM port, then overwrote each other in the backup history, and no question about an individual marker (how many are there, which port, which one failed) could be asked at all. One marker per PC, so the PC identifies the marker and the collector can mint it the same way it already mints a CMM or a Keyence unit for a metrology PC. The marker is a machine asset of type Part Marker, the PC `controls` it, and the marker is `partof` the operation it serves. That last rail is why the PC does not also claim the operation directly: `controls` propagates through `partof` (seeded in reference-data), so control of the operation follows from controlling its marker, and two markers on one operation no longer contest a link that can only have one holder. Returns [] for any PC that does not drive a marker, which leaves the ordinary machine link to run. """ from shopdb.api import AssetRelationship, RelationshipType, Asset from .pctypemap import subordinate_device_for spec = subordinate_device_for(pctype) if not spec or not spec.get('partof') or not comp or not comp.asset: # Only a device that FILES UNDER an operation goes through here. # A measuring tool is a subordinate device too, but it does not # share a machine number, so it keeps the simpler path. return [] pcasset = comp.asset controls = RelationshipType.query.filter_by( relationshiptype='controls').first() if not controls: warnings.append("'controls' relationship type missing; " 'run flask seed reference-data') return [] try: from plugins.machines.models import Machine, MachineType except ImportError: warnings.append('machines plugin unavailable; {} device skipped' .format(spec['typename'])) return [] label = spec['label'] # Reuse this PC's existing device before minting one, so a re-image # never leaves a second device behind for the same physical unit. existing = AssetRelationship.query.filter( AssetRelationship.sourceassetid == pcasset.assetid, AssetRelationship.relationshiptypeid == controls.relationshiptypeid, AssetRelationship.label == label, ).all() reuse = next((rel for rel in existing if rel.isactive), None) \ or (existing[0] if existing else None) if reuse: reuse.isactive = True markerasset = db.session.get(Asset, reuse.targetassetid) else: coretype = AssetType.query.filter_by( assettype=spec['assettype']).first() if not coretype: warnings.append('{} asset type missing; {} skipped'.format( spec['assettype'], spec['typename'])) return [] devicetype = MachineType.query.filter_by( machinetype=spec['typename']).first() if not devicetype: devicetype = MachineType(machinetype=spec['typename'], description=spec.get('description')) db.session.add(devicetype) db.session.flush() hostname = comp.hostname assetnumber = '{}-{}'.format( pcasset.assetnumber or hostname, spec['suffix']) # ADOPT AN EXISTING ASSET OF THAT NUMBER before minting. The number # is derived from the PC, so it is entirely predictable and may # already exist - left by an earlier run whose link was archived, or # created by a person. assetnumber is unique, so inserting a second # one raised and the whole report 500'd, on every cycle, forever. # # Deliberately NOT filtered on isactive: an inactive asset still # holds the number, and it is the constraint that decides this. markerasset = Asset.query.filter( db.func.lower(Asset.assetnumber) == assetnumber.lower()).first() if markerasset is None: markerasset = Asset( assetnumber=assetnumber, name='{} ({})'.format(spec['typename'], hostname), assettypeid=coretype.assettypeid, statusid=1) db.session.add(markerasset) db.session.flush() # The extension row may be missing on an adopted asset, and the link # may already exist under another label - both get-or-create. if not Machine.query.filter_by(assetid=markerasset.assetid).first(): db.session.add(Machine(assetid=markerasset.assetid, machinetypeid=devicetype.machinetypeid)) controlrow = AssetRelationship.query.filter_by( sourceassetid=pcasset.assetid, targetassetid=markerasset.assetid, relationshiptypeid=controls.relationshiptypeid).first() if controlrow is not None: controlrow.isactive = True if not controlrow.label: controlrow.label = label else: db.session.add(AssetRelationship( sourceassetid=pcasset.assetid, targetassetid=markerasset.assetid, relationshiptypeid=controls.relationshiptypeid, label=label, isactive=True)) # One marker per PC: archive any other collector marker link. for rel in existing: if rel is not reuse and rel.isactive: rel.isactive = False operation = self._link_marker_to_operation( markerasset, machinenumber, pcasset, label, warnings) return [{'assetid': markerasset.assetid, 'assetnumber': markerasset.assetnumber, 'operationassetid': operation}] def _link_marker_to_operation(self, markerasset, machinenumber, pcasset, label, warnings): """Make a marker `partof` the operation whose number its PC reports. Unlike the PC-to-machine link this does NOT contest: an operation can hold any number of markers, which is the whole point. Moving a marker to another operation archives the old membership rather than deleting it, so where a marker used to live stays answerable. Refuses to file a marker under the reporting PC. A PC first seen before the machine-number fix was created with the machine number as its OWN asset number, and that is deliberately never overwritten, so looking up the number can return the PC itself. Filing the marker partof its own PC would read, on the machine page, as the PC being the operation. """ from shopdb.api import AssetRelationship, RelationshipType, Asset if not machinenumber: return None partof = RelationshipType.query.filter_by( relationshiptype='partof').first() if not partof: warnings.append("'partof' relationship type missing; " 'run flask seed reference-data') return None operation = self._asset_by_number(machinenumber) if not operation: warnings.append( 'no asset for machine number {!r}; marker not filed under an ' 'operation'.format(machinenumber)) return None if operation.assetid == markerasset.assetid: return None if pcasset is not None and operation.assetid == pcasset.assetid: warnings.append( 'machine number {!r} is this PC\'s own asset number; marker ' 'not filed under an operation. Rename the PC asset to its ' 'hostname, or create the operation asset.'.format( machinenumber)) return None # Our own rows first: a marker that moved to another operation has its # old membership archived, never deleted, so where it used to live stays # answerable. for rel in AssetRelationship.query.filter( AssetRelationship.sourceassetid == markerasset.assetid, AssetRelationship.relationshiptypeid == partof.relationshiptypeid, AssetRelationship.label == label).all(): if rel.targetassetid != operation.assetid and rel.isactive: rel.isactive = False # Then get-or-create KEYED ON THE UNIQUE TRIPLE, not on the label. # assetrelationships is unique on (source, target, type), so a row made # by hand or by the legacy import - carrying a different label, or none # at all - was invisible to the label-filtered lookup this used to do, # and the insert then violated that constraint. The PC reported 200 on # the cycle that minted the marker and 500 on every cycle after. found = AssetRelationship.query.filter_by( sourceassetid=markerasset.assetid, targetassetid=operation.assetid, relationshiptypeid=partof.relationshiptypeid).first() if found is not None: found.isactive = True if not found.label: # Unlabelled means the import or a person made it. Stamping ours # is what makes the next cycle reuse it. A row carrying somebody # else's label is left as theirs. found.label = label else: db.session.add(AssetRelationship( sourceassetid=markerasset.assetid, targetassetid=operation.assetid, relationshiptypeid=partof.relationshiptypeid, label=label, isactive=True)) return operation.assetid def _assetname(self, assetid): """Readable name for an asset id, for warnings and alerts.""" from shopdb.api import Asset asset = db.session.get(Asset, assetid) return asset.assetnumber if asset else str(assetid) def _incumbent_has_yielded(self, assetid): """True when the PC currently holding a machine has given it up. Two ways to yield, and both are evidence rather than a guess: It went quiet. A PC pulled off a machine stops reporting, so silence past MACHINE_CLAIM_QUIET_HOURS is the handover signal. The window is long enough that a PC switched off overnight, or one behind a network outage, never loses its bay to a spare sitting on the bench. Or a person moved it off In Use. Setting the old PC to Retired, Inventory or In Repair is a deliberate statement that it no longer runs the machine, and it is the one-step way to force a handover the moment the swap happens instead of waiting out the window. An asset with no computer extension row cannot report at all, so it cannot be alive; it yields. """ from datetime import datetime, timedelta, timezone from shopdb.api import Asset computer = Computer.query.filter_by(assetid=assetid).first() if not computer: return True asset = db.session.get(Asset, assetid) status = asset.status.status if asset and asset.status else None if status and status != 'In Use': return True reported = computer.lastreporteddate if not reported: return True quiet = datetime.now(timezone.utc).replace(tzinfo=None) - reported return quiet > timedelta(hours=MACHINE_CLAIM_QUIET_HOURS) def _alert_machine_contested(self, holdername, newcomp, machine): """Say that a PC claims a machine another PC is still running. Fires ONCE, on the report that first records the dormant claim. The dormant relationship row is the marker: while it exists the claim is already known, so a PC sitting on the bench for a fortnight does not alert every collector cycle. Nothing is changed in the data by this. It exists because the two legitimate readings - a swap in progress, or a machine number typed onto the wrong PC at imaging - look identical to the collector, and only a person can tell them apart. """ import logging as _logging if not self._machinelink_alerts_enabled(): return try: from shopdb.api import send_alert, Setting newname = newcomp.hostname if newcomp else 'a new PC' machinename = machine.assetnumber or machine.name subject = 'Machine {} claimed by {}, still run by {}'.format( machinename, newname, holdername) base = (Setting.get('site_base_url') or '').rstrip('/') link = '{}/pcs/{}'.format(base, newcomp.computerid) if base else '' linkhtml = ('

View {1}

'.format(link, newname) if link else '') html = ( '

{0} reports that it runs machine ' '{1}, but {2} is still ' 'reporting on that machine and is still In Use, so the link ' 'has been left where it is.

' '

If this is a swap in progress, nothing needs doing: {0} ' 'takes the machine once {2} stops reporting for a day, or ' 'straight away if you set {2} to Retired, Inventory or In ' 'Repair. If instead {0} was imaged with the wrong machine ' 'number, correct it on {0}.

{3}'.format( newname, machinename, holdername, linkhtml)) send_alert(subject, html) except Exception: _logging.getLogger(__name__).exception( 'machine-contested alert failed for machine %s', getattr(machine, 'assetnumber', '?')) def _alert_pc_superseded(self, oldname, newcomp, machine): """Tell a human a PC was replaced on a machine. Best effort, never raises. Deliberately an ALERT and not a status change: the collector cannot tell whether the old PC was shelved, sent for repair or re-imaged for another bay, so it says what happened and lets a person decide. Goes through send_alert, which is the site's configured alert fan-out: email to the SMTP settings' alert recipients plus the alert webhook. Resolving those recipients by hand would have missed the SMTP_ALERT_RECIPIENTS environment fallback, so a site that configures SMTP by environment rather than in the UI would have got the webhook and no email. NOT a shopfloor notification: that board is for operators (General, Recertification, Recognition), and an IT asset message does not belong in front of the floor. """ import logging as _logging if not self._machinelink_alerts_enabled(): return try: from shopdb.api import send_alert, Setting newname = newcomp.hostname if newcomp else 'a new PC' machinename = machine.assetnumber or machine.name subject = 'PC replaced on machine {}: {} to {}'.format( machinename, oldname, newname) # /pcs/:id is keyed on computerid, not assetid - an assetid here # opens someone else's PC or a 404. base = (Setting.get('site_base_url') or '').rstrip('/') link = '{}/pcs/{}'.format(base, newcomp.computerid) if base else '' linkhtml = ('

View {1}

'.format(link, newname) if link else '') html = ( '

{0} is now reporting machine ' '{1}, which was previously driven by ' '{2}.

' '

The old link has been archived. {2} has NOT had its status ' 'changed - set it to Inventory, In Repair or Retired as ' 'appropriate.

{3}'.format(newname, machinename, oldname, linkhtml)) send_alert(subject, html) except Exception: _logging.getLogger(__name__).exception( 'PC-superseded alert failed for machine %s', getattr(machine, 'assetnumber', '?')) def _sync_measuringtool_link(self, pcasset, pctype, hostname, warnings, measuringtoolid=None, machinenumber=None): """Link the MeasuringTool a metrology PC drives. ADOPT before minting. A CMM / Keyence / Genspect / wax-and-trace pc-type means the PC controls an attached instrument. This links it with a directional PC->tool 'controls' relationship tagged MEASURINGTOOL_LINK_ORIGIN. RESOLUTION ORDER, most stable identity first: 1. measuringtoolid - the instrument named by the enrollment file measuringtool-id.txt. Explicit, survives a PC swap, and it is what a Keyence or Genspect bay has instead of a bay id. 2. a prior collector link on this PC (reactivate + retype). 3. an EXISTING measuring tool this PC already controls that the collector did not create - a legacy MT-#### row, or one somebody made by hand. Adopting stamps the label so it is ours from then on. 4. the reported machine number, when it resolves to a measuring tool. This is the CMM case: cmmid.txt already reports CMM4. 5. mint, only when none of the above found anything. WHY THIS ORDER. Minting derives an asset number from the HOSTNAME, so a permanent instrument inherited the identity of whichever PC drove it that week - replace the PC and either the number lies or a second tool appears for the same physical unit. It also could not see a tool it had not created itself, so on prod 43 legacy MT-#### tools were shadowed by minted -CMM twins, three records deep in places. Minting is now the last resort, and what it produces should be treated as a placeholder until a real identifier is recorded. A non-metrology pc-type archives any collector-created tool link (a PC re-imaged to another type) but never deletes the tool asset, which may carry calibration history. Returns the desired-link list. """ from shopdb.api import AssetRelationship, RelationshipType, Asset from .pctypemap import metrology_tool_for tool_spec = metrology_tool_for(pctype) controls = RelationshipType.query.filter_by( relationshiptype='controls').first() if not controls: # No 'controls' type => no tool links can exist. Only a problem for a # metrology PC that needs one; stay quiet for ordinary PCs. if tool_spec: warnings.append("'controls' relationship type missing; " 'run flask seed reference-data') return [] # Collector-created tool links already on this PC (active or archived). existing = AssetRelationship.query.filter( AssetRelationship.sourceassetid == pcasset.assetid, AssetRelationship.relationshiptypeid == controls.relationshiptypeid, AssetRelationship.label == MEASURINGTOOL_LINK_ORIGIN, ).all() if not tool_spec: # Not a metrology PC: archive any collector-created tool link. for rel in existing: if rel.isactive: rel.isactive = False return [] try: from plugins.measuringtools.models import MeasuringTool except ImportError: warnings.append('measuringtools plugin unavailable; ' 'tool link skipped') return [] typename, typedescription = tool_spec tooltype = self._ensure_measuringtool_type(typename, typedescription) # --- 1. an explicitly named instrument wins over everything --------- adopted = None named = (measuringtoolid or '').strip() if named: candidate = self._asset_by_number(named) if candidate is None: # Warn rather than invent: a typo in the file must not mint a # phantom instrument that nobody can account for. warnings.append( 'no asset for measuring tool {!r}; not linked'.format(named)) elif candidate.assetid not in self._measuringtool_assetids(): # The name resolved, but not to an instrument. measuringtool-id # .txt holding a machine number would otherwise link the PC to # that MACHINE under a measuring-tool label - a link that reads # as an instrument everywhere downstream, on a row the machine # sync also owns. Refuse and say which asset it hit. warnings.append( 'asset {!r} is not a measuring tool; not linked'.format( candidate.assetnumber)) else: adopted = candidate # --- 2. a prior collector link (reactivate + retype) ----------------- reuse = next((rel for rel in existing if rel.isactive), None) \ or (existing[0] if existing else None) # --- 3/4. adopt a tool this PC already controls that we did not make -- # Ordered after the collector's own link so a settled PC keeps what it # has, and before minting so an existing instrument is never twinned. if adopted is None and reuse is None: adopted = self._adoptable_measuringtool( pcasset, controls, machinenumber) if adopted is not None: # CLAIM the existing link rather than adding a second one. A row for # (pc, tool, controls) usually already exists - that is how the tool # was found - and assetrelationships is unique on exactly that # triple, so inserting would raise. claimed = AssetRelationship.query.filter_by( sourceassetid=pcasset.assetid, targetassetid=adopted.assetid, relationshiptypeid=controls.relationshiptypeid).first() if claimed is not None and claimed.label == MACHINE_LINK_ORIGIN: # LEAVE IT ALONE. On a CMM the instrument IS the reported bay, # so the machine sync has already made the row for this exact # triple - and it finds its own rows BY that label. Relabelling # hid the link from it, so the next cycle built a second row for # the same triple and MySQL rejected it: the PC reported 200, # then 500 forever after. The row's lifecycle (including staying # dormant while a bay is contested) belongs to the machine sync; # adoption only needs the identity, which it has. pass elif claimed is not None: # Relabelling is what adoption MEANS for any other row: the link # is ours from here, so a later cycle reuses it instead of # minting. claimed.label = MEASURINGTOOL_LINK_ORIGIN claimed.isactive = True else: db.session.add(AssetRelationship( sourceassetid=pcasset.assetid, targetassetid=adopted.assetid, relationshiptypeid=controls.relationshiptypeid, label=MEASURINGTOOL_LINK_ORIGIN)) if adopted.measuringtool and tooltype: adopted.measuringtool.measuringtooltypeid = \ tooltype.measuringtooltypeid targetid = adopted.assetid elif reuse: reuse.isactive = True toolasset = db.session.get(Asset, reuse.targetassetid) if toolasset and toolasset.measuringtool and tooltype: toolasset.measuringtool.measuringtooltypeid = \ tooltype.measuringtooltypeid targetid = reuse.targetassetid else: mt_assettype = AssetType.query.filter_by( assettype='measuring_tool').first() if not mt_assettype: warnings.append('measuring_tool asset type missing; ' 'tool link skipped') return [] suffix = (pctype or '').split('-')[-1].upper() baseasset = pcasset.assetnumber or hostname toolasset = Asset( assetnumber=f'{baseasset}-{suffix}', name=f'{typename} ({hostname})', assettypeid=mt_assettype.assettypeid, statusid=1) db.session.add(toolasset) db.session.flush() db.session.add(MeasuringTool( assetid=toolasset.assetid, measuringtooltypeid=tooltype.measuringtooltypeid if tooltype else None)) db.session.add(AssetRelationship( sourceassetid=pcasset.assetid, targetassetid=toolasset.assetid, relationshiptypeid=controls.relationshiptypeid, label=MEASURINGTOOL_LINK_ORIGIN)) targetid = toolasset.assetid # Only one tool link is desired; archive any other collector row. # # Keyed on the TARGET, not on object identity. `rel is not reuse` spared # the row reuse pointed at even when adoption had just resolved a # different instrument - so a bay whose measuringtool-id.txt named the # real tool kept its minted twin link alive alongside the adopted one, # and the PC read as controlling two instruments. That is the exact # duplicate this whole adoption path exists to remove. for rel in existing: if rel.targetassetid != targetid and rel.isactive: rel.isactive = False return [{'assetid': targetid, 'relationshiptype': 'controls', 'measuringtooltype': typename}] @staticmethod def _asset_by_number(assetnumber): """The active asset with exactly this asset number, case-insensitively. NOT `ilike`. Every identifier resolved here arrives from a file on a shopfloor PC - cmmid.txt, measuringtool-id.txt, the reported machine number - and ilike reads `_` and `%` in that value as WILDCARDS. A reported '%' matched whatever active asset happened to come first, and 'MT-600_' matched MT-6001: an identifier silently resolving to an asset it does not name, then linked to as though it did. """ from shopdb.api import Asset number = (assetnumber or '').strip() if not number: return None return Asset.query.filter( db.func.lower(Asset.assetnumber) == number.lower(), Asset.isactive.is_(True)).first() def _measuringtool_assetids(self): """Asset ids that really are measuring tools. Empty when the measuringtools plugin is not installed, which makes both callers fall through rather than guess: an asset number alone does not say what kind of thing it names. """ try: from plugins.measuringtools.models import MeasuringTool except ImportError: return set() return {row.assetid for row in MeasuringTool.query.with_entities(MeasuringTool.assetid)} def _adoptable_measuringtool(self, pcasset, controls, machinenumber): """A measuring tool this PC already controls that we did not create. Two sources, in order: a. any ACTIVE controls link from this PC to a measuring_tool asset that carries a different label - the legacy import's MT-#### rows, or one somebody added by hand. These were invisible to the collector, which keyed idempotency on its OWN label and so minted a twin for the same physical instrument on every fresh PC. b. the reported machine number, when it names a measuring tool. That is the CMM case: cmmid.txt reports CMM4, which already resolves to a real asset through the machine-link path. Returns the Asset to adopt, or None. Never creates anything. """ from shopdb.api import AssetRelationship, Asset toolassetids = self._measuringtool_assetids() if not toolassetids: return None # a. already linked, by someone else for rel in (AssetRelationship.query .filter(AssetRelationship.sourceassetid == pcasset.assetid, AssetRelationship.relationshiptypeid == controls.relationshiptypeid, AssetRelationship.isactive.is_(True)) .order_by(AssetRelationship.relationshipid.asc()).all()): if rel.label == MEASURINGTOOL_LINK_ORIGIN: continue if rel.targetassetid in toolassetids: return db.session.get(Asset, rel.targetassetid) # b. the reported machine number naming a tool number = (machinenumber or '').strip() if number: candidate = self._asset_by_number(number) if candidate is not None and candidate.assetid in toolassetids: return candidate return None def _ensure_measuringtool_type(self, name, description): """Find or create a MeasuringToolType (metrology types are not in the measuringtools starter seed).""" from plugins.measuringtools.models import MeasuringToolType tooltype = MeasuringToolType.query.filter_by(name=name).first() if not tooltype: tooltype = MeasuringToolType(name=name, description=description) db.session.add(tooltype) db.session.flush() return tooltype def on_install(self, app: Flask) -> None: """Called when plugin is installed.""" with app.app_context(): self._ensure_asset_type() self._ensure_computer_types() from .pctypemap import seed_pctype_settings seed_pctype_settings() db.session.commit() logger.info("Computers plugin installed") def _ensure_asset_type(self) -> None: """Ensure computer asset type exists.""" existing = AssetType.query.filter_by(assettype='computer').first() if not existing: at = AssetType( assettype='computer', pluginname='computers', tablename='computers', description='PCs, servers, and workstations', icon='desktop' ) db.session.add(at) logger.debug("Created asset type: computer") db.session.commit() def _ensure_computer_types(self) -> None: """Ensure basic computer types exist.""" computer_types = [ ('Shopfloor PC', 'PC located on the shop floor for machine operation', 'desktop'), ('Engineer Workstation', 'Engineering workstation for CAD/CAM work', 'laptop'), ('CMM PC', 'PC dedicated to CMM operation', 'desktop'), ('Server', 'Server system', 'server'), ('Kiosk', 'Kiosk or info display PC', 'tv'), ('Laptop', 'Laptop computer', 'laptop'), ('Virtual Machine', 'Virtual machine', 'cloud'), ('Other', 'Other computer type', 'desktop'), ] for name, description, icon in computer_types: existing = ComputerType.query.filter_by(computertype=name).first() if not existing: ct = ComputerType( computertype=name, description=description, icon=icon ) db.session.add(ct) logger.debug(f"Created computer type: {name}") db.session.commit() def on_uninstall(self, app: Flask) -> None: """Called when plugin is uninstalled.""" logger.info("Computers plugin uninstalled") def get_cli_commands(self) -> List: """Return CLI commands for this plugin.""" @click.group('computers') def computerscli(): """Computers plugin commands.""" pass @computerscli.command('list-types') def list_types(): """List all computer types.""" from flask import current_app with current_app.app_context(): types = ComputerType.query.filter_by(isactive=True).all() if not types: click.echo('No computer types found.') return click.echo('Computer Types:') for t in types: click.echo(f" [{t.computertypeid}] {t.computertype}") @computerscli.command('stats') def stats(): """Show computer statistics.""" from flask import current_app from shopdb.api import Asset with current_app.app_context(): total = db.session.query(Computer).join(Asset).filter( Asset.isactive == True ).count() click.echo(f"Total active computers: {total}") # Shopfloor count (by the Shopfloor computer type) sf = ComputerType.query.filter_by(computertype='Shopfloor').first() shopfloor = db.session.query(Computer).join(Asset).filter( Asset.isactive == True, Computer.computertypeid == (sf.computertypeid if sf else -1) ).count() click.echo(f" Shopfloor PCs: {shopfloor}") click.echo(f" Other: {total - shopfloor}") @computerscli.command('find') @click.argument('hostname') def find_by_hostname(hostname): """Find a computer by hostname.""" from flask import current_app with current_app.app_context(): comp = Computer.query.filter( Computer.hostname.ilike(f'%{hostname}%') ).first() if not comp: click.echo(f'No computer found matching hostname: {hostname}') return click.echo(f'Found: {comp.hostname}') click.echo(f' Asset: {comp.asset.assetnumber}') click.echo(f' Type: {comp.computertype.computertype if comp.computertype else "N/A"}') click.echo(f' OS: {comp.operatingsystem.osname if comp.operatingsystem else "N/A"}') click.echo(f' Logged in: {comp.loggedinuser or "N/A"}') return [computerscli] def get_dashboard_widgets(self) -> List[Dict]: """Dashboard cards this plugin contributes. REPLACES a declaration that named a Vue component which was never written - as did four other plugins - so nothing rendered and nobody noticed, because the frontend never called the widgets endpoint either. The contract now carries data and a renderer; core draws it. """ return [ { 'id': 'computers-quiet', 'viewall': '/pcs', 'title': 'PCs gone silent', 'endpoint': '/api/computers/dashboard/quiet', 'render': 'exceptions', 'severity': 'warning', 'permission': 'computers.view', 'empty': 'hide', 'position': 20, 'map': { 'title': 'hostname', 'detail': '', 'meta': [{'key': 'quietdays', 'label': 'silent', 'suffix': ' days'}], 'link': '/pcs/{computerid}', }, }, { 'id': 'computers-sharedmachines', 'viewall': '/machines', 'title': 'Machine numbers on two PCs', 'endpoint': '/api/computers/dashboard/sharedmachines', 'render': 'exceptions', 'severity': 'critical', 'permission': 'computers.view', 'empty': 'hide', 'position': 15, 'map': { 'title': 'machinenumber', 'detail': 'pcs', 'meta': [{'key': 'pccount', 'suffix': ' PCs'}], 'link': '/machines/{assetid}', }, }, ] def get_navigation_items(self) -> List[Dict]: """Return navigation menu items.""" return [ { 'name': 'PCs', 'icon': 'desktop', 'route': '/pcs', 'position': 15, }, ] def get_permissions(self) -> List: """Return the RBAC permissions this plugin owns.""" return [ ('computers.view', 'View computers', 'computers'), ('computers.create', 'Create computers', 'computers'), ('computers.edit', 'Edit computers', 'computers'), ('computers.delete', 'Delete computers', 'computers'), ]