Add the measuringtools plugin (ADR-005) and the plugin-system tutorial
Some checks failed
CI / backend (push) Failing after 28s
CI / naming (push) Successful in 1s
CI / frontend (push) Successful in 43s

Gage-lab instruments as Asset extensions: measuringtooltypes (color-coded
lookup) + measuringtools (calibration interval/dates, provider, notes) with
calibration status derived at read time (overdue / due soon / current /
unknown), never stored. Full CRUD API with permission-gated writes, types
management with in-use guard, calibration report, nav/reports/config-schema
hooks, and a complete frontend (list/detail/form, types settings page,
calibration report page, gated routes per ADR-009).

First plugin whose migration chain really creates tables post-cutover
(ADR-008), and the working example for docs/PLUGIN-GUIDE.md - a 12-section
walkthrough of building a plugin on this framework, linked from
PLUGIN-QUICKSTART and PLUGINS.

Verified: full suite 323 passing, live E2E on all four pages, fresh
scratch-MySQL migration dry-run green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-07-11 10:02:01 -04:00
parent 22e623c1f6
commit 2efe17b743
18 changed files with 2649 additions and 0 deletions

View File

@@ -0,0 +1,5 @@
"""Measuring-tools plugin package."""
from .plugin import MeasuringToolsPlugin
__all__ = ['MeasuringToolsPlugin']

View File

@@ -0,0 +1,5 @@
"""Measuring-tools plugin API blueprint."""
from .routes import measuringtools_bp
__all__ = ['measuringtools_bp']

View File

@@ -0,0 +1,362 @@
"""Measuring-tools plugin API.
Two resources: the measuringtooltypes lookup (CRUD with an in-use delete guard)
and the measuring tools themselves (an Asset core row plus a measuringtools
extension row, written in one payload). Reads are jwt-optional per the app
convention; writes require the measuringtools.* permissions.
Calibration status is derived at read time in the model (see
models.derive_status), never stored, so the report and list badges are always
current.
"""
from datetime import date, datetime
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.api import (
db, Asset, AssetType, AuditLog,
success_response, error_response, paginated_response, ErrorCodes,
get_pagination_params, paginate_query,
require_permission,
)
from ..models import MeasuringTool, MeasuringToolType, derive_status, STATUS_COLORS
measuringtools_bp = Blueprint('measuringtools', __name__)
def _parse_date(value):
"""Accept 'YYYY-MM-DD' (or None/empty) -> date or None."""
if not value:
return None
try:
return datetime.strptime(str(value)[:10], '%Y-%m-%d').date()
except (ValueError, TypeError):
return None
def _merged(tool, today=None):
"""Asset core dict with the measuringtools extension nested under 'measuringtool'."""
result = tool.asset.to_dict() if tool.asset else {}
result['measuringtool'] = tool.to_dict(today)
return result
# =============================================================================
# Measuring-tool types
# =============================================================================
@measuringtools_bp.route('/types', methods=['GET'])
@jwt_required(optional=True)
def list_types():
"""List measuring-tool types."""
page, per_page = get_pagination_params(request)
query = MeasuringToolType.query
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(MeasuringToolType.isactive == True)
if search := request.args.get('search'):
query = query.filter(MeasuringToolType.name.ilike(f'%{search}%'))
query = query.order_by(MeasuringToolType.name)
items, total = paginate_query(query, page, per_page)
return paginated_response([t.to_dict() for t in items], page, per_page, total)
@measuringtools_bp.route('/types/<int:type_id>', methods=['GET'])
@jwt_required(optional=True)
def get_type(type_id: int):
"""Get one measuring-tool type."""
tool_type = db.session.get(MeasuringToolType, type_id)
if not tool_type:
return error_response(ErrorCodes.NOT_FOUND,
f'Measuring-tool type {type_id} not found', http_code=404)
return success_response(tool_type.to_dict())
@measuringtools_bp.route('/types', methods=['POST'])
@jwt_required()
@require_permission('measuringtools.create')
def create_type():
"""Create a measuring-tool type (reactivates a soft-deleted same-named one)."""
data = request.get_json() or {}
name = (data.get('name') or '').strip()
if not name:
return error_response(ErrorCodes.VALIDATION_ERROR, 'name is required')
existing = MeasuringToolType.query.filter_by(name=name).first()
if existing:
if not existing.isactive:
existing.isactive = True
for key in ('description', 'color'):
if data.get(key) is not None:
setattr(existing, key, data[key])
db.session.commit()
return success_response(existing.to_dict(), message='Reactivated existing type')
return error_response(ErrorCodes.CONFLICT,
f"Measuring-tool type '{name}' already exists", http_code=409)
tool_type = MeasuringToolType(
name=name,
description=data.get('description'),
color=data.get('color'),
)
db.session.add(tool_type)
db.session.commit()
return success_response(tool_type.to_dict(), message='Measuring-tool type created',
http_code=201)
@measuringtools_bp.route('/types/<int:type_id>', methods=['PUT'])
@jwt_required()
@require_permission('measuringtools.edit')
def update_type(type_id: int):
"""Update a measuring-tool type."""
tool_type = db.session.get(MeasuringToolType, type_id)
if not tool_type:
return error_response(ErrorCodes.NOT_FOUND,
f'Measuring-tool type {type_id} not found', http_code=404)
data = request.get_json() or {}
if 'name' in data and data['name'] != tool_type.name:
if MeasuringToolType.query.filter_by(name=data['name']).first():
return error_response(ErrorCodes.CONFLICT,
f"Measuring-tool type '{data['name']}' already exists",
http_code=409)
for key in ('name', 'description', 'color', 'isactive'):
if key in data:
setattr(tool_type, key, data[key])
db.session.commit()
return success_response(tool_type.to_dict(), message='Measuring-tool type updated')
@measuringtools_bp.route('/types/<int:type_id>', methods=['DELETE'])
@jwt_required()
@require_permission('measuringtools.delete')
def delete_type(type_id: int):
"""Delete a measuring-tool type. Refused if any tool still uses it."""
tool_type = db.session.get(MeasuringToolType, type_id)
if not tool_type:
return error_response(ErrorCodes.NOT_FOUND,
f'Measuring-tool type {type_id} not found', http_code=404)
inuse = MeasuringTool.query.filter_by(measuringtooltypeid=type_id).count()
if inuse:
return error_response(ErrorCodes.CONFLICT,
f'Cannot delete: {inuse} tool(s) still use this type',
http_code=409)
db.session.delete(tool_type)
db.session.commit()
return success_response(message='Measuring-tool type deleted')
# =============================================================================
# Measuring tools
# =============================================================================
@measuringtools_bp.route('', methods=['GET'])
@jwt_required(optional=True)
def list_tools():
"""List measuring tools with filters + pagination.
Query params: typeid, locationid, statusid, calibrationstatus, search,
active, page, perpage.
"""
page, per_page = get_pagination_params(request)
query = db.session.query(MeasuringTool).join(Asset)
if request.args.get('active', 'true').lower() != 'false':
query = query.filter(Asset.isactive == True)
if search := request.args.get('search'):
query = query.filter(db.or_(
Asset.assetnumber.ilike(f'%{search}%'),
Asset.name.ilike(f'%{search}%'),
Asset.serialnumber.ilike(f'%{search}%'),
))
if type_id := request.args.get('typeid', type=int):
query = query.filter(MeasuringTool.measuringtooltypeid == type_id)
if location_id := request.args.get('locationid', type=int):
query = query.filter(Asset.locationid == location_id)
if status_id := request.args.get('statusid', type=int):
query = query.filter(Asset.statusid == status_id)
query = query.order_by(Asset.assetnumber)
items, total = paginate_query(query, page, per_page)
today = date.today()
data = [_merged(tool, today) for tool in items]
# Calibration status is derived, so it cannot be a SQL filter - apply it
# to the built rows. Filtering post-pagination is acceptable here because
# the dataset is a gage lab's worth of tools, not a fleet.
calibrationstatus = request.args.get('calibrationstatus')
if calibrationstatus:
data = [d for d in data
if d['measuringtool']['calibrationstatus'] == calibrationstatus]
return paginated_response(data, page, per_page, total)
@measuringtools_bp.route('/<int:tool_id>', methods=['GET'])
@jwt_required(optional=True)
def get_tool(tool_id: int):
"""Get one measuring tool (asset core merged with the extension)."""
tool = db.session.get(MeasuringTool, tool_id)
if not tool:
return error_response(ErrorCodes.NOT_FOUND,
f'Measuring tool {tool_id} not found', http_code=404)
return success_response(_merged(tool))
@measuringtools_bp.route('/by-asset/<int:asset_id>', methods=['GET'])
@jwt_required(optional=True)
def get_tool_by_asset(asset_id: int):
"""Get a measuring tool by its core asset id."""
tool = MeasuringTool.query.filter_by(assetid=asset_id).first()
if not tool:
return error_response(ErrorCodes.NOT_FOUND,
f'Measuring tool for asset {asset_id} not found', http_code=404)
return success_response(_merged(tool))
def _asset_type_id():
"""Resolve the seeded 'measuring_tool' asset-type id, or None."""
asset_type = AssetType.query.filter_by(assettype='measuring_tool').first()
return asset_type.assettypeid if asset_type else None
@measuringtools_bp.route('', methods=['POST'])
@jwt_required()
@require_permission('measuringtools.create')
def create_tool():
"""Create a measuring tool: one Asset core row + one extension row."""
data = request.get_json() or {}
if not data.get('assetnumber'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'assetnumber is required')
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
return error_response(ErrorCodes.CONFLICT,
f"Asset with number '{data['assetnumber']}' already exists",
http_code=409)
assettypeid = _asset_type_id()
if not assettypeid:
return error_response(ErrorCodes.INTERNAL_ERROR,
'measuring_tool asset type not found. Plugin may not be '
'properly installed.', http_code=500)
asset = Asset(
assetnumber=data['assetnumber'],
name=data.get('name'),
gaugelabreference=data.get('gaugelabreference'),
serialnumber=data.get('serialnumber'),
assettypeid=assettypeid,
statusid=data.get('statusid', 1),
locationid=data.get('locationid'),
businessunitid=data.get('businessunitid'),
mapx=data.get('mapx'),
mapy=data.get('mapy'),
notes=data.get('notes'),
)
db.session.add(asset)
db.session.flush() # assign assetid
tool = MeasuringTool(
assetid=asset.assetid,
measuringtooltypeid=data.get('measuringtooltypeid'),
calibrationintervaldays=data.get('calibrationintervaldays'),
lastcalibrationdate=_parse_date(data.get('lastcalibrationdate')),
nextcalibrationdate=_parse_date(data.get('nextcalibrationdate')),
calibrationprovider=(data.get('calibrationprovider') or '').strip() or None,
notes=(data.get('notes') or '').strip() or None,
)
db.session.add(tool)
db.session.flush()
AuditLog.log('created', 'MeasuringTool', entityid=tool.measuringtoolid,
entityname=asset.assetnumber)
db.session.commit()
return success_response(_merged(tool), message='Measuring tool created', http_code=201)
# Asset core fields writable through this plugin's write path.
_ASSET_FIELDS = ('assetnumber', 'name', 'gaugelabreference', 'serialnumber',
'statusid', 'locationid', 'businessunitid', 'mapx', 'mapy',
'notes', 'isactive')
# Extension fields with plain assignment (dates handled separately).
_TOOL_FIELDS = ('measuringtooltypeid', 'calibrationintervaldays',
'calibrationprovider', 'notes')
@measuringtools_bp.route('/<int:tool_id>', methods=['PUT'])
@jwt_required()
@require_permission('measuringtools.edit')
def update_tool(tool_id: int):
"""Update a measuring tool (asset core fields + extension in one payload)."""
tool = db.session.get(MeasuringTool, tool_id)
if not tool:
return error_response(ErrorCodes.NOT_FOUND,
f'Measuring tool {tool_id} not found', http_code=404)
data = request.get_json() or {}
asset = tool.asset
if 'assetnumber' in data and data['assetnumber'] != asset.assetnumber:
if Asset.query.filter_by(assetnumber=data['assetnumber']).first():
return error_response(ErrorCodes.CONFLICT,
f"Asset with number '{data['assetnumber']}' already exists",
http_code=409)
changes = {}
for key in _ASSET_FIELDS:
if key in data:
if getattr(asset, key) != data[key]:
changes[key] = {'old': getattr(asset, key), 'new': data[key]}
setattr(asset, key, data[key])
for key in _TOOL_FIELDS:
if key in data:
if getattr(tool, key) != data[key]:
changes[key] = {'old': getattr(tool, key), 'new': data[key]}
setattr(tool, key, data[key])
for key in ('lastcalibrationdate', 'nextcalibrationdate'):
if key in data:
setattr(tool, key, _parse_date(data[key]))
if changes:
AuditLog.log('updated', 'MeasuringTool', entityid=tool.measuringtoolid,
entityname=asset.assetnumber, changes=changes)
db.session.commit()
return success_response(_merged(tool), message='Measuring tool updated')
@measuringtools_bp.route('/<int:tool_id>', methods=['DELETE'])
@jwt_required()
@require_permission('measuringtools.delete')
def delete_tool(tool_id: int):
"""Soft delete a measuring tool (deactivates its asset)."""
tool = db.session.get(MeasuringTool, tool_id)
if not tool:
return error_response(ErrorCodes.NOT_FOUND,
f'Measuring tool {tool_id} not found', http_code=404)
tool.asset.isactive = False
AuditLog.log('deleted', 'MeasuringTool', entityid=tool.measuringtoolid,
entityname=tool.asset.assetnumber)
db.session.commit()
return success_response(message='Measuring tool deleted')
# =============================================================================
# Calibration report (for the Reports hub)
# =============================================================================
@measuringtools_bp.route('/report/calibration', methods=['GET'])
@jwt_required(optional=True)
def calibration_report():
"""Counts + lists bucketed by derived calibration status."""
today = date.today()
buckets = {'overdue': [], 'duesoon': [], 'current': [], 'unknown': []}
query = db.session.query(MeasuringTool).join(Asset).filter(Asset.isactive == True)
for tool in query.all():
buckets.setdefault(derive_status(tool.nextcalibrationdate, today), []).append(
_merged(tool, today))
return success_response({
'counts': {key: len(value) for key, value in buckets.items()},
'buckets': buckets,
'statuscolors': STATUS_COLORS,
})

View File

@@ -0,0 +1,13 @@
{
"name": "measuringtools",
"version": "1.0.0",
"description": "Metrology and inspection instruments (gauges, calipers, thread gages, bore gages) with a calibration lifecycle and derived calibration status.",
"author": "ShopDB Team",
"dependencies": [],
"core_version": ">=0.6.0,<1.0.0",
"api_prefix": "/api/measuringtools",
"default_enabled": false,
"provides": {
"features": ["measuring-tools", "calibration-tracking"]
}
}

View File

@@ -0,0 +1,16 @@
"""Alembic environment for the measuring-tools plugin migration chain.
Delegates to the shared runner in shopdb.plugins.alembic_template, which filters
the metadata to this plugin's tables and drives Alembic against the per-plugin
version table alembic_version_measuringtools. See ADR-008 for the ownership
model. Unlike the ten cutover plugins whose 0001 is a no-op anchor, this plugin
is NEW: its 0001 baseline really CREATES its tables, because the core chain
never built them.
"""
import os
os.environ['PLUGIN_NAME'] = 'measuringtools'
from shopdb.plugins.alembic_template import run_migrations # noqa: E402
run_migrations()

View File

@@ -0,0 +1,24 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade():
${upgrades if upgrades else "pass"}
def downgrade():
${downgrades if downgrades else "pass"}

View File

@@ -0,0 +1,69 @@
"""measuringtools plugin baseline (real create).
This plugin was built AFTER the ADR-008 ownership cutover, so unlike the ten
bundled plugins whose 0001 is a stamp-only no-op anchor, this baseline actually
CREATES the plugin's tables. The core Alembic chain never knew about
measuringtooltypes / measuringtools, so this per-plugin chain is their sole
authoritative creator. It runs from `flask plugin install measuringtools`
(and `flask plugin upgrade-all`) after `flask db upgrade` builds the core schema.
Note on create_plugin_tables: the shared helper in
shopdb.plugins.alembic_template builds a per-plugin MetaData filtered to only the
plugin's own tables, which means a foreign key to a core table (assets) cannot
be resolved at CreateTable-compile time (NoReferencedTableError). Since these
tables reference assets.assetid, the baseline emits explicit Alembic ops (the
same shape Alembic autogenerate produces) instead. Tables inherit the
connection's default charset, so on a utf8mb4 database they are utf8mb4, matching
how the core chain creates its tables.
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'measuringtools0001baseline'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
op.create_table(
'measuringtooltypes',
sa.Column('measuringtooltypeid', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=100), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('color', sa.String(length=20), nullable=True),
sa.Column('createddate', sa.DateTime(), nullable=False),
sa.Column('modifieddate', sa.DateTime(), nullable=False),
sa.Column('isactive', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('measuringtooltypeid'),
sa.UniqueConstraint('name'),
)
op.create_table(
'measuringtools',
sa.Column('measuringtoolid', sa.Integer(), nullable=False),
sa.Column('assetid', sa.Integer(), nullable=False),
sa.Column('measuringtooltypeid', sa.Integer(), nullable=True),
sa.Column('calibrationintervaldays', sa.Integer(), nullable=True),
sa.Column('lastcalibrationdate', sa.Date(), nullable=True),
sa.Column('nextcalibrationdate', sa.Date(), nullable=True),
sa.Column('calibrationprovider', sa.String(length=150), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('createddate', sa.DateTime(), nullable=False),
sa.Column('modifieddate', sa.DateTime(), nullable=False),
sa.Column('isactive', sa.Boolean(), nullable=False),
sa.ForeignKeyConstraint(['assetid'], ['assets.assetid'], ondelete='CASCADE'),
sa.ForeignKeyConstraint(['measuringtooltypeid'],
['measuringtooltypes.measuringtooltypeid']),
sa.PrimaryKeyConstraint('measuringtoolid'),
sa.UniqueConstraint('assetid'),
)
op.create_index('idx_measuringtool_type', 'measuringtools',
['measuringtooltypeid'])
def downgrade():
op.drop_index('idx_measuringtool_type', table_name='measuringtools')
op.drop_table('measuringtools')
op.drop_table('measuringtooltypes')

View File

@@ -0,0 +1,17 @@
"""Measuring-tools plugin models."""
from .measuringtool import (
MeasuringTool,
MeasuringToolType,
derive_status,
STATUS_COLORS,
DUESOON_WINDOW_DAYS,
)
__all__ = [
'MeasuringTool',
'MeasuringToolType',
'derive_status',
'STATUS_COLORS',
'DUESOON_WINDOW_DAYS',
]

View File

@@ -0,0 +1,133 @@
"""Measuring-tools models.
Measuring tools are gage-lab instruments (calipers, micrometers, thread gages,
bore gages, height gages, Genspect heads, ...) that measure parts, as opposed
to equipment that makes parts (see ADR-005). Each tool is a core Asset plus a
one-to-one measuringtools extension row.
The lifecycle a measuring tool cares about is CALIBRATION, not maintenance:
an interval, a last date, and a next date. Calibration STATUS is DERIVED from
nextcalibrationdate at read time (see derive_status), never stored, so it is
always current no matter how long since the last write. This mirrors the
warranty plugin's derive-at-read pattern.
"""
from datetime import date, timedelta
from shopdb.api import db, BaseModel
# Window before nextcalibrationdate where a tool counts as "due soon".
DUESOON_WINDOW_DAYS = 30
# Derived status -> display color (hex). Reused by the frontend status badge.
STATUS_COLORS = {
'overdue': '#F44336',
'duesoon': '#FF9800',
'current': '#4CAF50',
'unknown': '#9E9E9E',
}
def derive_status(nextcalibrationdate, today=None):
"""Calibration status from a next-calibration date. Never stored.
overdue - the next date is in the past
duesoon - the next date is within DUESOON_WINDOW_DAYS from today
current - the next date is further out than the window
unknown - no next date recorded
"""
if not nextcalibrationdate:
return 'unknown'
today = today or date.today()
if nextcalibrationdate < today:
return 'overdue'
if nextcalibrationdate <= today + timedelta(days=DUESOON_WINDOW_DAYS):
return 'duesoon'
return 'current'
class MeasuringToolType(BaseModel):
"""Measuring-tool classification (Caliper, Micrometer, Thread Gage, ...).
Site-managed lookup with a display color for badges and map markers,
the same shape as the equipment/computer/printer type tables.
"""
__tablename__ = 'measuringtooltypes'
measuringtooltypeid = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), unique=True, nullable=False)
description = db.Column(db.Text)
color = db.Column(db.String(20), comment='CSS color for UI/map markers')
def __repr__(self):
return f"<MeasuringToolType {self.name}>"
class MeasuringTool(BaseModel):
"""Measuring-tool extension data, one-to-one with a core Asset.
Identity lives on the Asset (assetnumber = gage tag, serialnumber = vendor
serial, gaugelabreference identifier). This row carries only the metrology
domain fields: the tool type and the calibration lifecycle.
"""
__tablename__ = 'measuringtools'
measuringtoolid = db.Column(db.Integer, primary_key=True)
# Link to the core asset (one extension row per asset).
assetid = db.Column(
db.Integer,
db.ForeignKey('assets.assetid', ondelete='CASCADE'),
unique=True,
nullable=False,
index=True,
)
measuringtooltypeid = db.Column(
db.Integer,
db.ForeignKey('measuringtooltypes.measuringtooltypeid'),
nullable=True,
)
# Calibration lifecycle. Status is DERIVED from nextcalibrationdate, so
# none of these columns store a status.
calibrationintervaldays = db.Column(db.Integer, nullable=True)
lastcalibrationdate = db.Column(db.Date, nullable=True)
nextcalibrationdate = db.Column(db.Date, nullable=True)
calibrationprovider = db.Column(db.String(150), nullable=True)
notes = db.Column(db.Text, nullable=True)
asset = db.relationship(
'Asset',
backref=db.backref('measuringtool', uselist=False, lazy='joined'),
)
measuringtooltype = db.relationship('MeasuringToolType', backref='tools')
__table_args__ = (
db.Index('idx_measuringtool_type', 'measuringtooltypeid'),
)
def __repr__(self):
return f"<MeasuringTool {self.assetid}>"
def calibrationstatus(self, today=None):
return derive_status(self.nextcalibrationdate, today)
def to_dict(self, today=None):
"""Extension dict with the type name and derived calibration status."""
result = super().to_dict()
# BaseModel.to_dict only isoformats datetime, not plain date columns.
# Emit calibration dates as 'YYYY-MM-DD' so the frontend parses them
# the same way it parses warranty dates.
for field in ('lastcalibrationdate', 'nextcalibrationdate'):
value = getattr(self, field)
result[field] = value.isoformat() if value else None
if self.measuringtooltype:
result['measuringtooltypename'] = self.measuringtooltype.name
result['measuringtooltypecolor'] = self.measuringtooltype.color
status = self.calibrationstatus(today)
result['calibrationstatus'] = status
result['calibrationstatuscolor'] = STATUS_COLORS.get(
status, STATUS_COLORS['unknown'])
return result

View File

@@ -0,0 +1,127 @@
"""Measuring-tools plugin main class.
The first plugin built on the matured framework scaffold (ADR-005). It owns two
tables (measuringtooltypes, measuringtools), a blueprint under /api/measuringtools,
a sidebar entry, and a calibration report card. It seeds its own asset type and
a set of starter tool types on install.
"""
import json
import logging
from pathlib import Path
from typing import List, Dict, Optional, Type
from flask import Flask, Blueprint
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.api import db, AssetType
from .api import measuringtools_bp
from .models import MeasuringTool, MeasuringToolType
logger = logging.getLogger(__name__)
# Starter tool types seeded on install: (name, description, color).
# Colors are drawn from the shared frontend PALETTE (utils/colorStyle.js).
STARTER_TYPES = [
('Caliper', 'Vernier / digital caliper', '#14abef'),
('Micrometer', 'Outside / inside / depth micrometer', '#2dce89'),
('Thread Gage', 'Go / no-go thread gage', '#fb6340'),
('Bore Gage', 'Bore / hole diameter gage', '#7934f3'),
('Height Gage', 'Height / vertical measuring gage', '#ffc107'),
('Indicator', 'Dial / test indicator', '#11cdef'),
('Gage Block Set', 'Reference gage block set', '#e83e8c'),
('Other', 'Other measuring tool', '#6c757d'),
]
class MeasuringToolsPlugin(BasePlugin):
"""Metrology and inspection instruments (calibration lifecycle)."""
def __init__(self):
self._manifest = self._load_manifest()
def _load_manifest(self) -> Dict:
manifest_path = Path(__file__).parent / 'manifest.json'
if manifest_path.exists():
with open(manifest_path, 'r') as f:
return json.load(f)
return {}
@property
def meta(self) -> PluginMeta:
return PluginMeta(
name=self._manifest.get('name', 'measuringtools'),
version=self._manifest.get('version', '1.0.0'),
description=self._manifest.get(
'description', 'Metrology and inspection instruments'),
author=self._manifest.get('author', 'ShopDB Team'),
dependencies=self._manifest.get('dependencies', []),
core_version=self._manifest.get('core_version', '>=0.6.0,<1.0.0'),
api_prefix=self._manifest.get('api_prefix', '/api/measuringtools'),
)
def get_blueprint(self) -> Optional[Blueprint]:
return measuringtools_bp
def get_models(self) -> List[Type]:
return [MeasuringTool, MeasuringToolType]
def get_navigation_items(self) -> List[Dict]:
return [
{
'name': 'Measuring Tools',
'icon': 'ruler',
'route': '/measuringtools',
'position': 22,
},
]
def get_reports(self) -> List[Dict]:
return [
{
'id': 'calibration',
'name': 'Calibration Due',
'description': 'Measuring tools bucketed by calibration status',
'category': 'compliance',
'route': '/reports/calibration',
},
]
def get_config_schema(self) -> List[Dict]:
# No external credentials or endpoints: calibration is tracked by hand,
# so the setup wizard shows nothing to configure. Documented in the
# plugin guide as the intentional empty-schema case.
return []
def init_app(self, app: Flask, db_instance) -> None:
logger.info(f"Measuring-tools plugin initialized (v{self.meta.version})")
def on_install(self, app: Flask) -> None:
with app.app_context():
self._ensure_asset_type()
self._ensure_starter_types()
db.session.commit()
logger.info("Measuring-tools plugin installed")
def _ensure_asset_type(self) -> None:
existing = AssetType.query.filter_by(assettype='measuring_tool').first()
if not existing:
db.session.add(AssetType(
assettype='measuring_tool',
pluginname='measuringtools',
tablename='measuringtools',
description='Metrology and inspection instruments (gauges, '
'calipers, thread gages, bore gages, ...)',
icon='ruler',
))
logger.debug("Created asset type: measuring_tool")
db.session.commit()
def _ensure_starter_types(self) -> None:
for name, description, color in STARTER_TYPES:
if not MeasuringToolType.query.filter_by(name=name).first():
db.session.add(MeasuringToolType(
name=name, description=description, color=color))
logger.debug(f"Created measuring-tool type: {name}")
db.session.commit()