ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into wilm

Browse Source
This commit is contained in:
benjamin.baker
2026-09-01 15:46:45 -04:00
parent a016753bb6 a0ba4dfc6a
commit 8cc8b0c9b2
6 changed files with 537 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

295
plugins/computers/plugin.py
View File

@@ -1,5 +1,6 @@
"""Computers plugin main class."""
import ipaddress
import json
import logging
from pathlib import Path
@@ -59,6 +60,115 @@ PARTMARKER_LINK_ORIGIN = 'collector:partmarker'
MACHINE_CLAIM_QUIET_HOURS = 24
def _corporate_ranges():
"""CIDRs this site considers its routable network, from settings.
Empty means the site has not said, and the fallback below applies. Kept
server-side rather than pushed to the bay: a bay cannot know a site's
addressing, and baking a range into the collector script would be an
ADR-015 violation the moment a second site ran it.
"""
from shopdb.api import Setting
raw = ''
try:
raw = Setting.get('collector_corporate_ranges') or ''
except Exception:
return []
ranges = []
for cidr in str(raw).split(','):
cidr = cidr.strip()
if not cidr:
continue
try:
ranges.append(ipaddress.ip_network(cidr, strict=False))
except ValueError:
continue
return ranges
def _pick_primary(interfaces, ranges):
"""Which reported NIC is the PC's address.
In order: inside a configured corporate range; else the one the bay says
carries the default route; else the first. The last is a guess and is the
reason configuring the ranges is worth doing - it is what a site sets so
the answer stops being a guess.
"""
def addr(entry):
try:
return ipaddress.ip_address((entry.get('ipaddress') or '').strip())
except ValueError:
return None
if ranges:
for entry in interfaces:
parsed = addr(entry)
if parsed and any(parsed in net for net in ranges):
return entry
for entry in interfaces:
if entry.get('isdefaultroute'):
return entry
return interfaces[0] if interfaces else None
def _record_interfaces(comp, interfaces):
"""Write one Communication row per reported NIC, flagging exactly one.
Keyed on the MAC where the bay sent one, because an address moves between
adapters and a MAC does not. Falls back to the address for a bay that
reports no MAC, which is what the single-ipaddress payload looks like.
Rows this PC has that the bay did NOT report are left alone: a NIC that is
down at check-in time is not evidence the record is wrong, and hand-entered
detail (gateway, DNS, serial settings) lives on these same rows.
"""
# Imported here for the same reason the caller does: these come through the
# shopdb.api contract surface, not from core internals.
from shopdb.api import Communication, CommunicationType, db
warnings = []
ip_comtype = CommunicationType.query.filter_by(comtype='IP').first()
if not ip_comtype:
return ['no IP communication type seeded; network not recorded']
usable = [entry for entry in interfaces if (entry.get('ipaddress') or '').strip()]
if not usable:
return warnings
primary = _pick_primary(usable, _corporate_ranges())
existing = Communication.query.filter_by(assetid=comp.assetid).all()
bymac = {(c.macaddress or '').strip().lower(): c for c in existing if c.macaddress}
byip = {(c.ipaddress or '').strip(): c for c in existing if c.ipaddress}
touched = []
for entry in usable:
addr = (entry.get('ipaddress') or '').strip()
mac = (entry.get('macaddress') or '').strip()
row = bymac.get(mac.lower()) if mac else None
if row is None:
row = byip.get(addr)
if row is None:
row = Communication(assetid=comp.assetid, comtypeid=ip_comtype.comtypeid)
db.session.add(row)
row.ipaddress = addr
if mac:
row.macaddress = mac
touched.append((row, entry is primary))
# Exactly one primary, and only among rows this report touched. An untouched
# row keeps its flag only if nothing here claimed it, which is what stops a
# check-in from silently demoting an address somebody set by hand.
for row, isprimary in touched:
row.isprimary = isprimary
if any(isprimary for _, isprimary in touched):
claimed = {id(row) for row, _ in touched}
for row in existing:
if id(row) not in claimed:
row.isprimary = False
return warnings
class ComputersPlugin(BasePlugin):
"""
Computers plugin - manages PC, server, and workstation assets.
@@ -362,7 +472,27 @@ class ComputersPlugin(BasePlugin):
'loggedinuser': {'type': 'string'},
'lastboottime': {'type': 'string', 'format': 'date-time'},
'lastcheckin': {'type': 'string', 'format': 'date-time'},
# Kept for bays on an older collector. A payload carrying
# both wins on `interfaces`.
'ipaddress': {'type': 'string'},
# EVERY routable IPv4 the bay can see, and no opinion about
# which one matters. Choosing was the bay's job and it was the
# wrong place for it: the bay cannot know a site's addressing,
# so a CMM with a 10Gbit link to its controller reported the
# machine-side address as the PC's, and the corporate address
# was never sent at all. The server picks the primary now,
# using collector_corporate_ranges, and records the rest.
'interfaces': {
'type': 'array',
'items': {
'ipaddress': 'string',
'macaddress': 'string',
'adaptername': 'string',
'isdefaultroute': 'boolean',
},
'description': ('All connected hardware NICs with a '
'routable IPv4. Order carries no meaning.'),
},
'vendorname': {'type': 'string'},
'modelnumber': {'type': 'string'},
'osname': {'type': 'string'},
@@ -446,6 +576,26 @@ class ComputersPlugin(BasePlugin):
site that installed an earlier one.
"""
return [
{
# Which addresses are THIS site's routable network. The
# collector reports every NIC and takes no view; this decides
# which one is the PC's address.
#
# Empty falls back to the NIC the bay says carries the default
# route, then to the first reported - both guesses. A CMM with
# a 10Gbit link to its controller is the case that shows why:
# that link can carry a default route too, and the machine-side
# address then wins over the corporate one.
#
# Comma-separated CIDRs, e.g. '10.48.0.0/16,10.20.0.0/23'.
'key': 'collector_corporate_ranges',
'value': '',
'valuetype': 'string',
'category': 'computers',
'description': ('Comma-separated CIDRs naming this site routable '
'network. The collector reports every NIC; this '
'picks which is the primary address.'),
},
{
'key': 'computers_quietreporthours',
'value': '24',
@@ -646,16 +796,18 @@ class ComputersPlugin(BasePlugin):
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))
# NETWORK. A bay reports every NIC it can see; THIS decides which one
# is the PC's address, because only the server knows the site's
# addressing. A CMM whose 10Gbit link to its controller sits on
# 100.0.0.1 used to report that as the PC's address and never send the
# corporate one at all - the bay had no way to tell them apart.
interfaces = payload.get('interfaces') or []
if not interfaces and payload.get('ipaddress'):
# An older bay sends one address and no opinion beyond it.
interfaces = [{'ipaddress': payload['ipaddress']}]
if interfaces:
warnings.extend(_record_interfaces(comp, interfaces))
for app_data in payload.get('installedsoftware', []) or []:
name = app_data.get('name')
@@ -1875,6 +2027,123 @@ class ComputersPlugin(BasePlugin):
for t in types:
click.echo(f" [{t.computertypeid}] {t.computertype}")
@computerscli.command('retype-by-role')
@click.option('--apply', 'apply_changes', is_flag=True,
help='Actually write. Without this the command only reports.')
@click.option('--typename', default='Metrology PC', show_default=True,
help='ComputerType to assign to PCs that drive a measuring tool.')
def retype_by_role(apply_changes, typename):
"""Type PCs that drive a measuring tool as metrology machines.
WHY THE RELATIONSHIP AND NOT THE NAME. A PC's role is already
recorded in what it controls: a metrology box has an active
relationship to a measuring_tool asset. Matching on hostnames or on
the tool's name would guess; this reads the fact.
The imaging scope (gea-shopfloor-keyence and friends) is the OTHER
source, and it is the authoritative one - but only 17 of ~292 PCs
have ever reported an enforcement scope, so it cannot answer for the
fleet today. Relationships can.
"""
from shopdb.api import Asset
target = ComputerType.query.filter_by(computertype=typename).first()
if not target:
if not apply_changes:
click.echo(click.style(
"Type '%s' does not exist yet; it would be created."
% typename, fg='yellow'))
else:
target = ComputerType(computertype=typename, isactive=True)
db.session.add(target)
db.session.flush()
click.echo("Created ComputerType '%s'." % typename)
rows = db.session.execute(db.text("""
SELECT DISTINCT c.computerid, a.assetnumber, c.hostname,
ct.computertype
FROM computers c
JOIN assets a ON a.assetid = c.assetid
LEFT JOIN computertypes ct ON ct.computertypeid = c.computertypeid
JOIN assetrelationships r ON r.sourceassetid = c.assetid
JOIN assets ta ON ta.assetid = r.targetassetid
JOIN assettypes at ON at.assettypeid = ta.assettypeid
WHERE at.assettype = 'measuring_tool' AND r.isactive = 1
ORDER BY ct.computertype, a.assetnumber
""")).fetchall()
moving = [r for r in rows if (r[3] or '') != typename]
click.echo('%d PC(s) drive a measuring tool; %d would change type.'
% (len(rows), len(moving)))
if not moving:
return
for r in moving:
click.echo(' %-14s %-14s %-16s -> %s'
% (r[1], r[2] or '(no hostname)', r[3] or '(none)', typename))
if not apply_changes:
click.echo(click.style(
'Dry run. Nothing was written. Re-run with --apply.', fg='yellow'))
return
for r in moving:
db.session.get(Computer, r[0]).computertypeid = target.computertypeid
db.session.commit()
click.echo(click.style('Retyped %d PC(s).' % len(moving), fg='green'))
@computerscli.command('prune-types')
@click.option('--apply', 'apply_changes', is_flag=True,
help='Actually delete. Without this the command only reports.')
def prune_types(apply_changes):
"""Delete computer types nothing uses and nothing maps to.
Two guards, because a type is cheap to keep and expensive to delete
by mistake. A type is KEPT when any computer is on it, or when the
live pctype map names it - the map is what the collector assigns
from at check-in, so deleting a mapped type would make every future
check-in log 'mapped ComputerType not found' instead of typing the
machine.
"""
from .pctypemap import pctype_mapping
mapped = {str(v).strip().lower() for v in pctype_mapping().values()}
counts = dict(db.session.execute(db.text("""
SELECT ct.computertypeid, COUNT(c.computerid)
FROM computertypes ct
LEFT JOIN computers c ON c.computertypeid = ct.computertypeid
GROUP BY ct.computertypeid
""")).fetchall())
removable, kept = [], []
for ctype in ComputerType.query.order_by(ComputerType.computertype).all():
inuse = counts.get(ctype.computertypeid, 0)
ismapped = ctype.computertype.strip().lower() in mapped
if inuse:
kept.append((ctype.computertype, '%d computer(s)' % inuse))
elif ismapped:
kept.append((ctype.computertype, 'named in the pctype map'))
else:
removable.append(ctype)
click.echo('KEEPING %d:' % len(kept))
for name, why in kept:
click.echo(' %-24s %s' % (name, why))
click.echo('')
click.echo('DELETING %d:' % len(removable))
for ctype in removable:
click.echo(' %s' % ctype.computertype)
if not removable:
return
if not apply_changes:
click.echo(click.style(
'Dry run. Nothing was deleted. Re-run with --apply.', fg='yellow'))
return
for ctype in removable:
db.session.delete(ctype)
db.session.commit()
click.echo(click.style('Deleted %d type(s).' % len(removable), fg='green'))
@computerscli.command('stats')
def stats():
"""Show computer statistics."""
@@ -1888,8 +2157,10 @@ class ComputersPlugin(BasePlugin):
click.echo(f"Total active computers: {total}")
# Shopfloor count (by the Shopfloor computer type)
sf = ComputerType.query.filter_by(computertype='Shopfloor').first()
# 'Shopfloor PC' - NOT 'Shopfloor', which is an empty
# duplicate this counted for as long as both existed, so the
# number reported was always 0.
sf = ComputerType.query.filter_by(computertype='Shopfloor PC').first()
shopfloor = db.session.query(Computer).join(Asset).filter(
Asset.isactive == True,
Computer.computertypeid == (sf.computertypeid if sf else -1)