a.williamson/BAT
1
0
Fork 0
forked from b.peck/BAT
Code Pull Requests Activity
Files
main
BAT/exchangeResources/exchange_lint.py
a.williamson 58610c1e1a exchange work
2026-09-15 13:06:45 -05:00

1244 lines
52 KiB
Python
Executable File
Raw Permalink Blame History

#!/usr/bin/env python3
"""exchange_lint.py — conformance checker for the Ignition Exchange Resources Style Guide.
Tests whether an Ignition project on disk (8.1 project folder layout) follows the
naming, structure, and storage conventions in "Exchange Resources Style Guide"
v1.1.0, and whether it is ready to upload to the Exchange.
Every finding carries a rule id (E-...) so it can be traced back to a section of
the guide; run with --list-rules to print the full table.
Severity follows the guide's own language:
FAIL the guide says "must", or the rule protects importability into someone
else's project (resource namespacing, session/client-tag storage).
warn a naming convention -- the guide "encourages" these but states they are
not required. --strict promotes every warning to a failure.
Targets: a project folder, any folder containing projects (a gateway's
projects/ directory, an unpacked gateway backup, a source repo), or a
project-export .zip. The resource name is inferred from the namespace the
project actually uses, so no per-project configuration is needed.
Usage:
python3 exchange_lint.py # scan the current directory
python3 exchange_lint.py /path/to/MyProject
python3 exchange_lint.py MyResource.zip --strict
python3 exchange_lint.py /gateway/projects # every project found
python3 exchange_lint.py --list-rules
Exit status: 0 clean, 1 findings, 2 bad invocation.
Stdlib only; no third-party dependencies.
"""
import argparse
import ast
import io
import json
import os
import re
import sys
import tokenize
PERSPECTIVE = "com.inductiveautomation.perspective"
VISION = "com.inductiveautomation.vision"
# ---------------------------------------------------------------- rule table
RULES = [
# id, section, description
("E-PROJ-NAME", "Projects", "project name lowercase, dashes, URL-friendly, no version suffix"),
("E-PROJ-TITLE", "Projects", "project title human-friendly Title Case"),
("E-PROJ-DESC", "Upload", "project description filled in (feeds the Exchange overview)"),
("E-NS-ROOT", "Resource Structure", "every resource under Exchange/<ResourceName>/ or exchange/<resource-name>/"),
("E-NS-CASE", "Resource Structure", "correct case of the namespace root for the resource type"),
("E-NS-NAME", "Resource Structure", "second path segment is the resource name"),
("E-VIEW-NAME", "Perspective", "view names Title Case or PascalCase"),
("E-COMP-NAME", "Perspective", "component names PascalCase"),
("E-PROP-NAME", "Perspective", "custom/param property names camelCase"),
("E-PAGE-NAME", "Perspective", "page names lowercase with dashes"),
("E-PAGE-ROOT", "Perspective", "page '/' claims the root URL of any importing project"),
("E-STYLE-NAME", "Perspective", "style class names lowercase with dashes"),
("E-STYLE-HIER", "Perspective", "style class placement mirrors the views that use it"),
("E-MSG-NAME", "Perspective", "message handlers namespaced exchange.resourceName.handlerName"),
("E-STORE-SESS", "Variable Storage", "no Perspective session custom properties (blocks clean import)"),
("E-STORE-CTAG", "Variable Storage", "no Vision client tags (blocks clean import)"),
("E-STORE-GLOB", "Variable Storage", "getGlobals() keyed ['exchange']['resourceName']"),
("E-TAG-NAME", "Tags", "tag folders/names PascalCase or Title Case under Exchange/<Resource>"),
("E-VIS-NAME", "Vision", "window/template names Title Case or PascalCase"),
("E-PY-TABS", "Code", "Python indented with tabs, not spaces"),
("E-PY-LIB", "Code", "project library packages lowercase, single words"),
("E-PY-FUNC", "Code", "function names camelCase"),
("E-PY-CLASS", "Code", "class names PascalCase"),
("E-PY-VAR", "Code", "variable names camelCase"),
("E-PY-DOC", "Code", "public functions/classes carry a docstring"),
("E-PY-DOCFMT", "Code", "docstring summary ends with . ? or !"),
("E-PY-COMMENT", "Code", "comments start with '# ' (hash then space)"),
("E-PY-PARSE", "Code", "source parses (checks below it are skipped otherwise)"),
("E-LOG-NAME", "Loggers", "logger names exchange.resourceName[...].PascalCaseName"),
("E-DB-TABLE", "Database", "table names ex_<abbrev>_snake_case"),
("E-DB-COL", "Database", "column names snake_case"),
("E-DB-ID", "Database", "every table has an 'id' primary key"),
("E-DB-FK", "Database", "foreign key columns named <foreign_table>_id"),
("E-NQ-NAME", "Named Queries", "named query names Title Case or PascalCase"),
("E-NQ-PARAM", "Named Queries", "named query parameters camelCase"),
("E-WEB-NAME", "WebDev", "WebDev sources lowercase with dashes"),
("E-I18N-KEY", "Translation", "translation keys 'word' or '_translation_phrase'"),
("E-UP-DOCS", "Upload", "installation/README documentation present in the resource"),
]
RULE_SECTIONS = {r[0]: r[1] for r in RULES}
VALID_RULES = set(RULE_SECTIONS)
# Resource types and the case their namespace root uses, per the guide's
# Project Browser screenshots (uppercase Exchange/ for designer resources that
# are themselves capitalised; lowercase exchange/ for URL- and code-facing ones).
UPPER = "Exchange"
LOWER = "exchange"
# ------------------------------------------------------------ name predicates
RX_PASCAL = re.compile(r"^[A-Z][A-Za-z0-9]*$")
RX_CAMEL = re.compile(r"^[a-z][A-Za-z0-9]*$")
RX_KEBAB = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$")
RX_SNAKE = re.compile(r"^[a-z][a-z0-9_]*$")
RX_CONST = re.compile(r"^[A-Z][A-Z0-9_]*$")
RX_PROJECT = re.compile(r"^[a-z0-9]+([-_][a-z0-9]+)*$")
RX_VERSION_SUFFIX = re.compile(r"[-_. ]v?\d+([._]\d+)*$", re.IGNORECASE)
# Title Case tolerates lowercase joining words in non-initial position.
SMALL_WORDS = {"a", "an", "and", "as", "at", "by", "for", "in", "of", "on",
"or", "the", "to", "vs", "with"}
def is_pascal(s):
return bool(RX_PASCAL.match(s))
def is_camel(s):
return bool(RX_CAMEL.match(s))
def is_kebab(s):
return bool(RX_KEBAB.match(s))
def is_title(s):
words = s.split(" ")
if len(words) < 2 or any(not w for w in words):
return False
for i, w in enumerate(words):
if RX_PASCAL.match(w):
continue
if i > 0 and w.lower() in SMALL_WORDS:
continue
return False
return True
def is_title_or_pascal(s):
"""The guide lets top level resources be either Title Case or PascalCase."""
return is_pascal(s) or is_title(s)
def norm(s):
"""Fold a name to bare lowercase alphanumerics for cross-style comparison."""
return re.sub(r"[^a-z0-9]", "", (s or "").lower())
def pascalize(s):
return "".join(p[:1].upper() + p[1:] for p in re.split(r"[-_ ]+", s or "") if p)
def kebabize(s):
s = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "-", s or "")
return re.sub(r"[-_ ]+", "-", s).lower()
# ------------------------------------------------------------------- findings
class Report(object):
def __init__(self, ignore=None, strict=False):
self.findings = [] # (severity, rule, where, message)
self.ignore = ignore or set()
self.strict = strict
self.checked = set()
def _add(self, sev, rule, where, msg):
self.checked.add(rule)
if rule in self.ignore:
return
if sev == "warn" and self.strict:
sev = "FAIL"
self.findings.append((sev, rule, where, msg))
def fail(self, rule, where, msg):
self._add("FAIL", rule, where, msg)
def warn(self, rule, where, msg):
self._add("warn", rule, where, msg)
def touch(self, *rules):
"""Record that a rule ran even though it produced no finding."""
self.checked.update(rules)
@property
def fails(self):
return [f for f in self.findings if f[0] == "FAIL"]
@property
def warns(self):
return [f for f in self.findings if f[0] == "warn"]
# --------------------------------------------------------------- file helpers
def load_json(rep, path, where):
try:
with open(path) as f:
return json.load(f)
except Exception as e:
rep.fail("E-PY-PARSE", where, "invalid JSON: %s" % e)
return None
def read_text(path):
with open(path, "r", encoding="utf-8", errors="replace") as f:
return f.read()
def find_payload_dirs(root, payload):
"""Yield path-segment tuples for every directory under root holding payload."""
out = []
if not os.path.isdir(root):
return out
for dirpath, _dirs, files in os.walk(root):
if payload in files:
rel = os.path.relpath(dirpath, root)
if rel != ".":
out.append(tuple(rel.split(os.sep)))
return sorted(out)
def find_resource_dirs(root):
"""Yield path-segment tuples for every leaf directory holding a resource.json."""
out = []
if not os.path.isdir(root):
return out
for dirpath, _dirs, files in os.walk(root):
if "resource.json" in files:
rel = os.path.relpath(dirpath, root)
if rel != ".":
out.append(tuple(rel.split(os.sep)))
return sorted(out)
def walk_components(node, visit, path="root"):
"""Depth-first over a Perspective view component tree; visit(node, path)."""
if not isinstance(node, dict):
return
visit(node, path)
for i, child in enumerate(node.get("children") or []):
name = "?"
if isinstance(child, dict):
name = (child.get("meta") or {}).get("name", "?")
walk_components(child, visit, "%s/%s[%d]" % (path, name, i))
# ------------------------------------------------------------ namespace check
def check_namespace(rep, kind, parts, where, want_case, resource, name_style):
"""Check <Exchange|exchange>/<ResourceName>/... on one resource path.
parts path segments below the resource-type root
want_case UPPER or LOWER -- the expected spelling of the first segment
name_style "title-pascal" or "kebab" -- expected style of the resource segment
Returns the segments whose names still need style checking: those below
the namespace when it is correct, or below the wrong root when it is not.
"""
if len(parts) < 2:
rep.fail("E-NS-ROOT", where,
"%s is not inside a %s/<resource>/ folder (imports will collide "
"with the target project)" % (kind, want_case))
return list(parts)
root, name = parts[0], parts[1]
if norm(root) != "exchange":
rep.fail("E-NS-ROOT", where,
"%s namespace root is %r, expected %r" % (kind, root, want_case))
# Keep checking the names below the (wrong) root so a project that has
# not been namespaced yet still gets its naming findings.
return list(parts[1:])
rep.touch("E-NS-ROOT")
if root != want_case:
rep.warn("E-NS-CASE", where,
"namespace root %r should be %r for %s" % (root, want_case, kind))
else:
rep.touch("E-NS-CASE")
if resource and norm(name) != norm(resource):
rep.warn("E-NS-NAME", where,
"resource folder %r does not match resource name %r" % (name, resource))
else:
rep.touch("E-NS-NAME")
if name_style == "kebab" and not is_kebab(name):
rep.warn("E-NS-NAME", where,
"resource folder %r should be lowercase-with-dashes" % name)
elif name_style == "title-pascal" and not is_title_or_pascal(name):
rep.warn("E-NS-NAME", where,
"resource folder %r should be Title Case or PascalCase" % name)
elif name_style == "camel" and not is_camel(name):
rep.warn("E-NS-NAME", where,
"resource folder %r should be lowercase or camelCase "
"(exchange.resourceName.scriptName)" % name)
return parts[2:]
# ------------------------------------------------------------------- sections
def check_project(rep, proj_dir, proj_name):
"""Projects: name, title, description."""
where = "project.json"
if not RX_PROJECT.match(proj_name):
rep.fail("E-PROJ-NAME", where,
"project name %r must be lowercase letters, numbers, hyphens and "
"underscores only" % proj_name)
elif RX_VERSION_SUFFIX.search(proj_name):
rep.warn("E-PROJ-NAME", where,
"project name %r looks version-suffixed; the Exchange manages "
"versions on upload" % proj_name)
else:
rep.touch("E-PROJ-NAME")
pj = os.path.join(proj_dir, "project.json")
if not os.path.exists(pj):
rep.warn("E-PROJ-TITLE", where, "no project.json found")
return
meta = load_json(rep, pj, where) or {}
title = (meta.get("title") or "").strip()
if not title:
rep.warn("E-PROJ-TITLE", where, "project title is empty")
elif not (is_title(title) or is_pascal(title)):
rep.warn("E-PROJ-TITLE", where,
"project title %r should be human-friendly Title Case" % title)
else:
rep.touch("E-PROJ-TITLE")
if not (meta.get("description") or "").strip():
rep.warn("E-PROJ-DESC", where,
"project description is empty; the Exchange builds the resource "
"readme from it")
else:
rep.touch("E-PROJ-DESC")
def check_perspective(rep, proj_dir, resource):
"""Perspective: views, components, properties, pages, style classes."""
views_root = os.path.join(proj_dir, PERSPECTIVE, "views")
styles_root = os.path.join(proj_dir, PERSPECTIVE, "style-classes")
view_paths = find_payload_dirs(views_root, "view.json")
style_paths = find_payload_dirs(styles_root, "style.json")
# ---- views -----------------------------------------------------------
for parts in view_paths:
where = "views/" + "/".join(parts)
tail = check_namespace(rep, "view", parts, where, UPPER, resource, "title-pascal")
for seg in (tail if tail is not None else []):
if not is_title_or_pascal(seg):
rep.warn("E-VIEW-NAME", where,
"view path segment %r should be Title Case or PascalCase" % seg)
else:
rep.touch("E-VIEW-NAME")
check_view_file(rep, os.path.join(views_root, *parts), where, resource)
# ---- style classes ---------------------------------------------------
for parts in style_paths:
where = "style-classes/" + "/".join(parts)
tail = check_namespace(rep, "style class", parts, where, LOWER, resource, "kebab")
for seg in (tail if tail is not None else []):
if not is_kebab(seg):
rep.warn("E-STYLE-NAME", where,
"style class segment %r should be lowercase-with-dashes" % seg)
else:
rep.touch("E-STYLE-NAME")
check_style_hierarchy(rep, views_root, view_paths, style_paths)
# ---- pages -----------------------------------------------------------
pc_path = os.path.join(proj_dir, PERSPECTIVE, "page-config", "config.json")
if os.path.exists(pc_path):
pc = load_json(rep, pc_path, "page-config/config.json") or {}
for url in (pc.get("pages") or {}):
where = "page-config/config.json"
if url == "/":
rep.warn("E-PAGE-ROOT", where,
"page '/' takes over the root URL of any project this "
"resource is imported into")
continue
rep.touch("E-PAGE-ROOT")
for seg in [s for s in url.split("/") if s]:
if seg.startswith(":"): # path parameter
continue
if not is_kebab(seg):
rep.warn("E-PAGE-NAME", where,
"page url segment %r should be lowercase-with-dashes" % seg)
else:
rep.touch("E-PAGE-NAME")
# ---- session properties ---------------------------------------------
sp_path = os.path.join(proj_dir, PERSPECTIVE, "session-props", "props.json")
if os.path.exists(sp_path):
sp = load_json(rep, sp_path, "session-props/props.json") or {}
custom = sp.get("custom") or {}
leaves = _leaf_paths(custom)
if leaves:
rep.fail("E-STORE-SESS", "session-props/props.json",
"%d custom session propert%s (%s%s) -- the guide directs "
"resources to use view properties and system.util.getGlobals() "
"instead, so importing does not merge into the host project's "
"session props"
% (len(leaves), "y" if len(leaves) == 1 else "ies",
", ".join(leaves[:3]), ", ..." if len(leaves) > 3 else ""))
else:
rep.touch("E-STORE-SESS")
def _leaf_paths(node, prefix=""):
"""Flatten a nested property dict to dotted leaf paths."""
out = []
if isinstance(node, dict) and node:
for k in sorted(node):
out.extend(_leaf_paths(node[k], "%s.%s" % (prefix, k) if prefix else k))
elif prefix:
out.append(prefix)
return out
def check_view_file(rep, view_dir, where, resource):
"""Component names, custom/param property names, message handlers, globals."""
vfile = os.path.join(view_dir, "view.json")
view = load_json(rep, vfile, where)
if view is None:
return
raw = read_text(vfile)
for group in ("params", "custom"):
for key in (view.get(group) or {}):
if not is_camel(key):
rep.warn("E-PROP-NAME", where,
"view %s property %r should be camelCase" % (group, key))
else:
rep.touch("E-PROP-NAME")
for key in (view.get("propConfig") or {}):
if key.startswith(("params.", "custom.")):
leaf = key.split(".", 1)[1].split(".")[0].split("[")[0]
if leaf and not is_camel(leaf):
rep.warn("E-PROP-NAME", where,
"view property %r should be camelCase" % key)
else:
rep.touch("E-PROP-NAME")
def visit(node, cpath):
name = (node.get("meta") or {}).get("name")
if cpath == "root" or not name:
return
if not is_pascal(name):
rep.warn("E-COMP-NAME", where, "component %r should be PascalCase" % name)
else:
rep.touch("E-COMP-NAME")
for key in (node.get("custom") or {}):
if not is_camel(key):
rep.warn("E-PROP-NAME", where,
"custom property %r on %r should be camelCase" % (key, name))
else:
rep.touch("E-PROP-NAME")
walk_components(view.get("root") or {}, visit)
check_message_names(rep, raw, where, resource)
check_globals(rep, raw, where, resource)
def check_message_names(rep, text, where, resource):
"""Message handler / sendMessage types namespaced to the resource."""
seen = set()
for rx in (r'"messageType"\s*:\s*"([^"]+)"',
r'sendMessage\(\s*[\'"]([^\'"]+)[\'"]',
r'messageType\s*=\s*[\'"]([^\'"]+)[\'"]'):
for m in re.finditer(rx, text):
seen.add(m.group(1))
for mt in sorted(seen):
parts = mt.split(".")
if len(parts) >= 3 and parts[0] == "exchange" and norm(parts[1]) == norm(resource):
rep.touch("E-MSG-NAME")
continue
rep.warn("E-MSG-NAME", where,
"message type %r should be exchange.%s.handlerName to avoid "
"collisions" % (mt, resource[:1].lower() + resource[1:]))
def check_globals(rep, text, where, resource):
"""getGlobals() must be keyed ['exchange']['resourceName']."""
for m in re.finditer(r"getGlobals\(\)\s*((?:\[\s*[\'\"][^\'\"]+[\'\"]\s*\])*)", text):
keys = re.findall(r"[\'\"]([^\'\"]+)[\'\"]", m.group(1))
if len(keys) >= 2 and keys[0] == "exchange" and norm(keys[1]) == norm(resource):
rep.touch("E-STORE-GLOB")
else:
rep.warn("E-STORE-GLOB", where,
"getGlobals()%s should be keyed ['exchange']['%s']"
% ("".join("[%r]" % k for k in keys) or "[...]",
resource[:1].lower() + resource[1:]))
def check_style_hierarchy(rep, views_root, view_paths, style_paths):
"""A class used by exactly one view belongs in a folder named for that view."""
if not style_paths:
return
known = {"/".join(p) for p in style_paths}
users = {}
for parts in view_paths:
vfile = os.path.join(views_root, *parts, "view.json")
if not os.path.exists(vfile):
continue
view = load_json(rep, vfile, "views/" + "/".join(parts))
if view is None:
continue
view_name = parts[-1]
def visit(node, _cpath):
classes = ((node.get("props") or {}).get("style") or {}).get("classes")
if isinstance(classes, str):
for cls in classes.split():
if cls in known:
users.setdefault(cls, set()).add(view_name)
walk_components(view.get("root") or {}, visit)
for cls, vnames in sorted(users.items()):
segs = cls.split("/")
view_segs = {norm(s) for s in segs}
if len(vnames) == 1:
vn = next(iter(vnames))
if norm(vn) not in view_segs:
rep.warn("E-STYLE-HIER", "style-classes/" + cls,
"used only by view %r; place it in a folder named for that "
"view" % vn)
else:
rep.touch("E-STYLE-HIER")
else:
shared = view_segs & {norm(v) for v in vnames}
if shared:
rep.warn("E-STYLE-HIER", "style-classes/" + cls,
"used by %d views (%s) but nested under a view-specific "
"folder; move it to the topmost folder covering them all"
% (len(vnames), ", ".join(sorted(vnames)[:3])))
else:
rep.touch("E-STYLE-HIER")
def check_vision(rep, proj_dir, resource):
"""Vision: windows, templates, and the client-tag prohibition."""
vis = os.path.join(proj_dir, VISION)
if not os.path.isdir(vis):
return
ctags = os.path.join(vis, "client-tags")
if os.path.isdir(ctags) and os.listdir(ctags):
rep.fail("E-STORE-CTAG", "%s/client-tags" % VISION,
"Vision client tags present -- the guide directs resources away "
"from client tags so they import cleanly into another project")
else:
rep.touch("E-STORE-CTAG")
for sub, kind in (("windows", "window"), ("templates", "template")):
root = os.path.join(vis, sub)
for parts in find_resource_dirs(root):
where = "%s/%s/%s" % (VISION, sub, "/".join(parts))
tail = check_namespace(rep, kind, parts, where, UPPER, resource, "title-pascal")
for seg in (tail if tail is not None else []):
if not is_title_or_pascal(seg):
rep.warn("E-VIS-NAME", where,
"%s name %r should be Title Case or PascalCase" % (kind, seg))
else:
rep.touch("E-VIS-NAME")
def check_scripts(rep, proj_dir, resource):
"""Code: library layout, indentation, names, docstrings, comments, loggers."""
sp_root = os.path.join(proj_dir, "ignition", "script-python")
if not os.path.isdir(sp_root):
return
for dirpath, _dirs, files in os.walk(sp_root):
if "code.py" not in files:
continue
parts = tuple(os.path.relpath(dirpath, sp_root).split(os.sep))
where = "script-python/" + "/".join(parts)
if norm(parts[0]) == "exchange":
tail = check_namespace(rep, "script package", parts, where, LOWER,
resource, "camel")
tail = tail if tail is not None else []
else:
rep.fail("E-NS-ROOT", where,
"project library package %r is not under exchange/<resource>/; "
"importing would land scripts in the host project's namespace"
% parts[0])
tail = list(parts[1:])
for seg in tail:
if not is_camel(seg):
rep.warn("E-PY-LIB", where,
"library package/script %r should be lowercase, a single "
"word where possible, camelCase otherwise" % seg)
else:
rep.touch("E-PY-LIB")
check_python_file(rep, os.path.join(dirpath, "code.py"), where + "/code.py",
resource)
def check_python_file(rep, path, where, resource):
src = read_text(path)
check_py_indentation(rep, src, where)
check_py_comments(rep, src, where)
check_globals(rep, src, where, resource)
check_message_names(rep, src, where, resource)
check_loggers(rep, src, where, resource)
try:
tree = ast.parse(src)
except SyntaxError as e:
rep.warn("E-PY-PARSE", where,
"could not parse (line %s): %s -- name and docstring checks skipped"
% (e.lineno, e.msg))
return
rep.touch("E-PY-PARSE")
check_py_names(rep, tree, where)
def check_py_indentation(rep, src, where):
"""Tabs, not spaces. Uses the tokenizer so continuation lines are not counted."""
space_lines, tab_lines = [], []
try:
for tok in tokenize.generate_tokens(io.StringIO(src).readline):
if tok.type == tokenize.INDENT:
(tab_lines if "\t" in tok.string else space_lines).append(tok.start[0])
except (tokenize.TokenError, IndentationError, SyntaxError):
for i, line in enumerate(src.splitlines(), 1):
if line.startswith("\t"):
tab_lines.append(i)
elif line.startswith(" "):
space_lines.append(i)
if space_lines:
preview = ", ".join(str(n) for n in space_lines[:5])
rep.warn("E-PY-TABS", where,
"%d space-indented block%s (line%s %s%s); Ignition's editor inserts "
"tabs" % (len(space_lines), "" if len(space_lines) == 1 else "s",
"" if len(space_lines) == 1 else "s", preview,
", ..." if len(space_lines) > 5 else ""))
elif tab_lines:
rep.touch("E-PY-TABS")
def check_py_comments(rep, src, where):
"""Comments start with '#' then a space."""
bad = []
try:
for tok in tokenize.generate_tokens(io.StringIO(src).readline):
if tok.type != tokenize.COMMENT:
continue
body = tok.string
if body in ("#", "#!"):
continue
if body.startswith("#!") and tok.start[0] == 1:
continue
if re.match(r"^#[-=#*]{2,}", body): # separator banners
continue
if not body.startswith("# "):
bad.append(tok.start[0])
except (tokenize.TokenError, IndentationError, SyntaxError):
return
if bad:
rep.warn("E-PY-COMMENT", where,
"%d comment%s missing the space after '#' (line%s %s%s)"
% (len(bad), "" if len(bad) == 1 else "s",
"" if len(bad) == 1 else "s",
", ".join(str(n) for n in bad[:5]),
", ..." if len(bad) > 5 else ""))
else:
rep.touch("E-PY-COMMENT")
def check_py_names(rep, tree, where):
"""Function/class/variable names and docstrings."""
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
if not is_pascal(node.name.lstrip("_")):
rep.warn("E-PY-CLASS", where,
"class %r (line %d) should be PascalCase" % (node.name, node.lineno))
else:
rep.touch("E-PY-CLASS")
check_docstring(rep, node, where, "class")
elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
bare = node.name.strip("_")
dunder = node.name.startswith("__") and node.name.endswith("__")
if not dunder and bare and not is_camel(bare):
rep.warn("E-PY-FUNC", where,
"function %r (line %d) should be camelCase (the guide "
"diverges from PEP-8 here)" % (node.name, node.lineno))
else:
rep.touch("E-PY-FUNC")
if not dunder:
check_docstring(rep, node, where, "function")
for arg in _all_args(node):
_check_var(rep, arg.arg, arg.lineno, where, "argument")
elif isinstance(node, ast.Assign):
for tgt in node.targets:
for name, lineno in _assigned_names(tgt):
_check_var(rep, name, lineno, where, "variable")
elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
_check_var(rep, node.target.id, node.lineno, where, "variable")
def _all_args(node):
a = node.args
out = list(getattr(a, "posonlyargs", [])) + list(a.args) + list(a.kwonlyargs)
if a.vararg:
out.append(a.vararg)
if a.kwarg:
out.append(a.kwarg)
return out
def _assigned_names(tgt):
if isinstance(tgt, ast.Name):
yield tgt.id, tgt.lineno
elif isinstance(tgt, (ast.Tuple, ast.List)):
for el in tgt.elts:
for pair in _assigned_names(el):
yield pair
def _check_var(rep, name, lineno, where, kind):
bare = name.strip("_")
if not bare or bare == "self" or bare == "cls":
return
# Module constants fall back to PEP-8, which the guide names as the default
# for anything it does not cover.
if RX_CONST.match(bare):
rep.touch("E-PY-VAR")
return
if is_camel(bare):
rep.touch("E-PY-VAR")
return
rep.warn("E-PY-VAR", where,
"%s %r (line %d) should be camelCase" % (kind, name, lineno))
def check_docstring(rep, node, where, kind):
doc = ast.get_docstring(node, clean=False)
if node.name.startswith("_"):
return
if not doc or not doc.strip():
rep.warn("E-PY-DOC", where,
"%s %r (line %d) has no docstring; Ignition 8.1.32+ surfaces these "
"in autocomplete" % (kind, node.name, node.lineno))
return
rep.touch("E-PY-DOC")
summary = doc.strip().splitlines()[0].strip()
if summary and summary[-1] not in ".?!":
rep.warn("E-PY-DOCFMT", where,
"%s %r (line %d) docstring summary should end with . ? or !"
% (kind, node.name, node.lineno))
else:
rep.touch("E-PY-DOCFMT")
def check_loggers(rep, src, where, resource):
"""exchange.resourceName[.path].PascalCaseName"""
want = resource[:1].lower() + resource[1:]
for m in re.finditer(r"getLogger\(\s*[\'\"]([^\'\"]+)[\'\"]", src):
name = m.group(1)
parts = name.split(".")
ok = (len(parts) >= 3 and parts[0] == "exchange"
and norm(parts[1]) == norm(resource)
and all(is_camel(p) for p in parts[2:-1])
and is_pascal(parts[-1]))
if ok:
rep.touch("E-LOG-NAME")
else:
rep.warn("E-LOG-NAME", where,
"logger %r should be exchange.%s[.path].LoggerName with a "
"PascalCase final element" % (name, want))
def check_named_queries(rep, proj_dir, resource):
root = os.path.join(proj_dir, "ignition", "named-query")
for parts in find_resource_dirs(root):
where = "named-query/" + "/".join(parts)
tail = check_namespace(rep, "named query", parts, where, UPPER, resource,
"title-pascal")
for seg in (tail if tail is not None else []):
if not is_title_or_pascal(seg):
rep.warn("E-NQ-NAME", where,
"named query %r should be Title Case or PascalCase" % seg)
else:
rep.touch("E-NQ-NAME")
qdir = os.path.join(root, *parts)
for fn in sorted(os.listdir(qdir)):
if not fn.endswith(".json") or fn == "resource.json":
continue
data = load_json(rep, os.path.join(qdir, fn), where)
for p in ((data or {}).get("parameters") or []):
pname = p.get("name") if isinstance(p, dict) else p
if isinstance(pname, str) and not is_camel(pname):
rep.warn("E-NQ-PARAM", where,
"query parameter %r should be camelCase" % pname)
elif isinstance(pname, str):
rep.touch("E-NQ-PARAM")
def check_other_resources(rep, proj_dir, resource):
"""Alarm pipelines, SFCs, transaction groups, reports, WebDev, translations."""
ign = os.path.join(proj_dir, "ignition")
upper_kinds = {
"alarm-pipeline": "alarm pipeline", "alarm-pipelines": "alarm pipeline",
"sfc": "SFC", "sfcs": "SFC",
"transaction-group": "transaction group", "transaction-groups": "transaction group",
"report": "report", "reports": "report",
}
if os.path.isdir(ign):
for entry in sorted(os.listdir(ign)):
kind = upper_kinds.get(entry)
if not kind:
continue
for parts in find_resource_dirs(os.path.join(ign, entry)):
where = "%s/%s" % (entry, "/".join(parts))
tail = check_namespace(rep, kind, parts, where, UPPER, resource,
"title-pascal")
for seg in (tail if tail is not None else []):
if not is_title_or_pascal(seg):
rep.warn("E-VIEW-NAME", where,
"%s name %r should be Title Case or PascalCase"
% (kind, seg))
for webroot in (os.path.join(proj_dir, "com.inductiveautomation.webdev"),
os.path.join(ign, "webdev")):
if not os.path.isdir(webroot):
continue
base = os.path.join(webroot, "resources")
base = base if os.path.isdir(base) else webroot
for parts in find_resource_dirs(base):
where = "webdev/" + "/".join(parts)
tail = check_namespace(rep, "WebDev source", parts, where, LOWER,
resource, "kebab")
for seg in (tail if tail is not None else []):
stem = seg.rsplit(".", 1)[0]
if not is_kebab(stem):
rep.warn("E-WEB-NAME", where,
"WebDev source %r should be lowercase-with-dashes" % seg)
else:
rep.touch("E-WEB-NAME")
check_translations(rep, proj_dir)
def check_translations(rep, proj_dir):
for dirpath, _dirs, files in os.walk(proj_dir):
for fn in files:
if "locale" not in fn and "translation" not in fn.lower():
continue
if not fn.endswith(".json"):
continue
where = os.path.relpath(os.path.join(dirpath, fn), proj_dir)
data = load_json(rep, os.path.join(dirpath, fn), where)
keys = list(data) if isinstance(data, dict) else []
for key in keys:
if not isinstance(key, str):
continue
ok = (re.match(r"^[A-Za-z0-9]+$", key)
or re.match(r"^_[a-z0-9]+(_[a-z0-9]+)*$", key))
if ok:
rep.touch("E-I18N-KEY")
else:
rep.warn("E-I18N-KEY", where,
"translation key %r should be a single word, or "
"_underscore_separated_phrase" % key)
def check_tags(rep, tag_files, resource):
"""Tag export JSON: Exchange/<Resource>/... with PascalCase or Title Case names."""
for tf in tag_files:
where = os.path.basename(tf)
data = load_json(rep, tf, where)
if data is None:
continue
def walk(node, trail):
if not isinstance(node, dict):
return
name = node.get("name")
here = trail + [name] if name else trail
if name:
if not is_title_or_pascal(name):
rep.warn("E-TAG-NAME", where,
"tag/folder %r should be PascalCase or Title Case"
% "/".join(here))
else:
rep.touch("E-TAG-NAME")
for child in (node.get("tags") or []):
walk(child, here)
roots = data.get("tags") or []
top = [t.get("name") for t in roots if isinstance(t, dict)]
if not any(norm(n) == "exchange" for n in top if n):
rep.fail("E-NS-ROOT", where,
"tag export root is %s; tags must live under Exchange/%s/"
% (top or "(empty)", resource))
for t in roots:
walk(t, [])
def check_database(rep, sql_files, resource):
"""CREATE TABLE statements: ex_ prefix, snake_case, id PK, *_id foreign keys."""
abbrev_hint = "".join(w[0] for w in re.findall(r"[A-Z][a-z]*", resource)).lower()
for sf in sql_files:
where = os.path.basename(sf)
sql = read_text(sf)
for m in re.finditer(
r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[`\"\[]?([A-Za-z0-9_.]+)[`\"\]]?\s*\((.*?)\)\s*;",
sql, re.IGNORECASE | re.DOTALL):
table = m.group(1).split(".")[-1]
body = m.group(2)
if not RX_SNAKE.match(table):
rep.warn("E-DB-TABLE", where,
"table %r should be snake_case" % table)
elif not table.startswith("ex_"):
rep.warn("E-DB-TABLE", where,
"table %r should start with 'ex_' (e.g. ex_%s_%s) to avoid "
"collisions in an existing schema"
% (table, abbrev_hint or "abbrev", table))
else:
rep.touch("E-DB-TABLE")
cols = []
for line in body.splitlines():
line = line.strip().rstrip(",")
cm = re.match(r"^[`\"\[]?([A-Za-z0-9_]+)[`\"\]]?\s+[A-Za-z]", line)
if not cm:
continue
kw = cm.group(1).upper()
if kw in ("PRIMARY", "FOREIGN", "UNIQUE", "KEY", "INDEX", "CONSTRAINT", "CHECK"):
continue
cols.append(cm.group(1))
for col in cols:
if not RX_SNAKE.match(col):
rep.warn("E-DB-COL", where,
"column %s.%s should be snake_case" % (table, col))
else:
rep.touch("E-DB-COL")
if "id" not in cols:
rep.warn("E-DB-ID", where,
"table %r has no 'id' column; the first auto-increment "
"primary key should be named 'id'" % table)
else:
rep.touch("E-DB-ID")
for fkm in re.finditer(
r"FOREIGN\s+KEY\s*\(\s*[`\"\[]?([A-Za-z0-9_]+)[`\"\]]?\s*\)\s*"
r"REFERENCES\s+[`\"\[]?([A-Za-z0-9_]+)",
body, re.IGNORECASE):
col, ref = fkm.group(1), fkm.group(2).split(".")[-1]
if col != "%s_id" % ref:
rep.warn("E-DB-FK", where,
"foreign key %s.%s references %s; name it %s_id"
% (table, col, ref, ref))
else:
rep.touch("E-DB-FK")
def check_upload_readiness(rep, proj_dir, extra_roots=()):
"""Documentation the Exchange upload form and the guide's checklist expect."""
bases = [proj_dir] + [r for r in extra_roots if r]
for base in bases:
if not os.path.isdir(base):
continue
for fn in os.listdir(base):
if re.match(r"^(readme|install|installation|docs?|instructions)",
fn, re.IGNORECASE):
rep.touch("E-UP-DOCS")
return
rep.warn("E-UP-DOCS", "(resource root)",
"no README / installation document found; the Exchange upload form "
"asks for custom installation instructions")
# ------------------------------------------------------------------ discovery
# Markers specific enough not to match a gateway's config/resources tree, which
# also carries com.inductiveautomation.* folders but is not a project.
PROJECT_MARKERS = (
"project.json",
os.path.join(PERSPECTIVE, "views"),
os.path.join(PERSPECTIVE, "session-props"),
os.path.join(PERSPECTIVE, "page-config"),
os.path.join(VISION, "windows"),
os.path.join(VISION, "templates"),
os.path.join("ignition", "script-python"),
)
MAX_SCAN_DEPTH = 6
SKIP_DIRS = {".git", ".svn", "__pycache__", "node_modules", ".resources",
".venv", "venv", ".idea", ".vscode"}
def looks_like_project(d):
"""True if d is the root of an Ignition 8.x project folder."""
if not os.path.isdir(d):
return False
return any(os.path.exists(os.path.join(d, m)) for m in PROJECT_MARKERS)
def discover_projects(target):
"""Return project directories under target.
Accepts a project folder, any directory containing one or more projects
(a gateway's projects/ dir, an unpacked gateway backup, a repo), or a
project-export .zip. Nested projects are not searched once a project root
is found, so a projects/ dir yields its children, not their subfolders.
"""
target = os.path.abspath(target)
if os.path.isfile(target):
if target.lower().endswith(".zip"):
return discover_projects(unpack_zip(target))
raise SystemExit("not a project folder or .zip export: %s" % target)
if not os.path.isdir(target):
raise SystemExit("no such path: %s" % target)
if looks_like_project(target):
return [target]
found = []
base_depth = target.rstrip(os.sep).count(os.sep)
for dirpath, dirs, _files in os.walk(target):
dirs[:] = [d for d in dirs if d not in SKIP_DIRS and not d.startswith(".")]
if dirpath.count(os.sep) - base_depth >= MAX_SCAN_DEPTH:
dirs[:] = []
continue
for d in list(dirs):
cand = os.path.join(dirpath, d)
if looks_like_project(cand):
found.append(cand)
dirs.remove(d)
return sorted(found)
def unpack_zip(path):
"""Extract a project-export zip to a temp dir and return the project root."""
import tempfile
import zipfile
dest = tempfile.mkdtemp(prefix="exchange_lint_")
with zipfile.ZipFile(path) as zf:
for member in zf.namelist():
# refuse absolute paths and traversal in an untrusted archive
norm_m = os.path.normpath(member)
if os.path.isabs(norm_m) or norm_m.startswith(".."):
raise SystemExit("unsafe path in archive: %s" % member)
zf.extractall(dest)
entries = [os.path.join(dest, e) for e in os.listdir(dest)]
if len(entries) == 1 and os.path.isdir(entries[0]):
return entries[0]
return dest
def discover_aux(proj_dir, pattern_fn, limit=200):
"""Collect auxiliary files (tag exports, .sql dumps) shipped beside a project."""
out = []
roots = [proj_dir, os.path.dirname(proj_dir.rstrip(os.sep))]
seen = set()
for base in roots:
if not os.path.isdir(base) or base in seen:
continue
seen.add(base)
for dirpath, dirs, files in os.walk(base):
dirs[:] = [d for d in dirs if d not in SKIP_DIRS and not d.startswith(".")]
for fn in sorted(files):
p = os.path.join(dirpath, fn)
if p not in out and pattern_fn(p):
out.append(p)
if len(out) >= limit:
return out
return out
def is_tag_export(path):
if not path.lower().endswith(".json"):
return False
if os.path.basename(path) in ("project.json", "resource.json", "view.json",
"style.json", "props.json", "config.json"):
return False
try:
with open(path) as f:
head = f.read(4096)
except OSError:
return False
return '"tags"' in head and ('"tagType"' in head or '"valueSource"' in head
or '"tagGroup"' in head)
def is_sql_file(path):
return path.lower().endswith(".sql")
def infer_resource(proj_dir, proj_name):
"""Best guess at the resource name from the view namespace, else the project."""
for sub, payload in ((os.path.join(PERSPECTIVE, "views"), "view.json"),
(os.path.join(VISION, "windows"), "resource.json")):
paths = find_payload_dirs(os.path.join(proj_dir, sub), payload)
if not paths:
continue
tops = {}
for p in paths:
key = p[1] if len(p) > 1 and norm(p[0]) == "exchange" else p[0]
tops[key] = tops.get(key, 0) + 1
return max(tops, key=lambda k: tops[k])
return pascalize(proj_name)
# ---------------------------------------------------------------------- main
def render(rep, proj_name, proj_dir, resource, as_json, show_header=True,
max_per_rule=8):
if as_json:
return {
"project": proj_name,
"path": proj_dir,
"resource": resource,
"failures": len(rep.fails),
"warnings": len(rep.warns),
"rules_exercised": sorted(rep.checked),
"findings": [{"severity": s, "rule": r, "section": RULE_SECTIONS.get(r, "?"),
"where": w, "message": m} for s, r, w, m in rep.findings],
}
if show_header:
print("Ignition Exchange Style Guide v1.1.0 — conformance report")
print("\nproject: %s resource: %s" % (proj_name, resource))
print("path: %s" % proj_dir)
if not rep.findings:
print(" No findings.")
sections = []
for _rid, sec, _desc in RULES:
if sec not in sections:
sections.append(sec)
for sec in sections:
rows = [f for f in rep.findings if RULE_SECTIONS.get(f[1]) == sec]
if not rows:
continue
print("\n %s" % sec)
rows.sort(key=lambda f: (f[0] != "FAIL", f[1], f[2]))
shown = {}
for sev, rule, where, msg in rows:
n = shown.get(rule, 0)
if max_per_rule and n >= max_per_rule:
shown[rule] = n + 1
continue
shown[rule] = n + 1
print(" %-4s %-13s %s: %s" % (sev, rule, where, msg))
for rule, n in shown.items():
if max_per_rule and n > max_per_rule:
print(" %-4s %-13s ... and %d more (-v for all)"
% ("", rule, n - max_per_rule))
print("\n summary: %d failure(s), %d warning(s); %d of %d rules exercised"
% (len(rep.fails), len(rep.warns), len(rep.checked), len(RULES)))
return None
def lint_project(proj_dir, resource=None, tags=None, sql=None, ignore=None,
strict=False):
"""Run every check against one project directory and return its Report."""
proj_name = os.path.basename(proj_dir.rstrip(os.sep))
resource = resource or infer_resource(proj_dir, proj_name)
tags = tags if tags is not None else discover_aux(proj_dir, is_tag_export)
sql = sql if sql is not None else discover_aux(proj_dir, is_sql_file)
rep = Report(ignore=ignore, strict=strict)
check_project(rep, proj_dir, proj_name)
check_perspective(rep, proj_dir, resource)
check_vision(rep, proj_dir, resource)
check_scripts(rep, proj_dir, resource)
check_named_queries(rep, proj_dir, resource)
check_other_resources(rep, proj_dir, resource)
check_tags(rep, tags, resource)
check_database(rep, sql, resource)
check_upload_readiness(rep, proj_dir,
[os.path.dirname(proj_dir.rstrip(os.sep))])
return rep, proj_name, resource
def main(argv=None):
ap = argparse.ArgumentParser(
description="Check an Ignition project against the Ignition Exchange "
"Resources Style Guide (v1.1.0).",
epilog="TARGET may be a project folder, a folder containing projects "
"(a gateway projects/ dir, an unpacked gateway backup, a repo), "
"or a project-export .zip. Defaults to the current directory.")
ap.add_argument("target", nargs="*", default=None,
help="project folder, containing folder, or .zip export")
ap.add_argument("--resource", help="Exchange resource name (default: inferred "
"from the resource namespace)")
ap.add_argument("--tags", nargs="*", default=None,
help="tag export .json files (default: auto-discovered)")
ap.add_argument("--sql", nargs="*", default=None,
help=".sql files to check (default: auto-discovered)")
ap.add_argument("--strict", action="store_true", help="treat warnings as failures")
ap.add_argument("--ignore", default="", help="comma-separated rule ids to suppress")
ap.add_argument("--json", action="store_true", dest="as_json",
help="machine-readable output")
ap.add_argument("--list-rules", action="store_true",
help="print the rule table and exit")
ap.add_argument("-v", "--verbose", action="store_true",
help="list every occurrence instead of collapsing per rule")
ap.add_argument("--max-per-rule", type=int, default=8,
help="occurrences shown per rule before collapsing (default 8)")
args = ap.parse_args(argv)
if args.list_rules:
width = max(len(r[0]) for r in RULES)
for rid, sec, desc in RULES:
print("%-*s %-20s %s" % (width, rid, sec, desc))
return 0
ignore = {r.strip() for r in args.ignore.split(",") if r.strip()}
unknown = ignore - VALID_RULES
if unknown:
print("unknown rule id(s) in --ignore: %s" % ", ".join(sorted(unknown)),
file=sys.stderr)
return 2
targets = args.target or [os.getcwd()]
projects = []
try:
for t in targets:
for p in discover_projects(t):
if p not in projects:
projects.append(p)
except SystemExit as e:
print(e, file=sys.stderr)
return 2
if not projects:
print("no Ignition project found under: %s" % ", ".join(targets),
file=sys.stderr)
return 2
if args.resource and len(projects) > 1:
print("--resource applies to a single project; %d found" % len(projects),
file=sys.stderr)
return 2
results, total_fails = [], 0
for i, proj_dir in enumerate(projects):
rep, proj_name, resource = lint_project(
proj_dir, resource=args.resource, tags=args.tags, sql=args.sql,
ignore=ignore, strict=args.strict)
total_fails += len(rep.fails)
payload = render(rep, proj_name, proj_dir, resource, args.as_json,
show_header=(i == 0),
max_per_rule=0 if args.verbose else args.max_per_rule)
if payload is not None:
results.append(payload)
if args.as_json:
print(json.dumps({"projects": results,
"failures": total_fails}, indent=2))
elif len(projects) > 1:
print("\n%d project(s) checked, %d total failure(s)"
% (len(projects), total_fails))
return 1 if total_fails else 0
if __name__ == "__main__":
sys.exit(main())
View Git Blame Copy Permalink