diff --git a/pyproject.toml b/pyproject.toml index 0ebcff8..dad0968 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ "colored", "h5py", "hdf5plugin", + "lxml", "numpy", "packaging", "pygments", @@ -52,6 +53,8 @@ Changelog = "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/nexpy/nexusformat/releases" [project.scripts] nexusformat = "nexusformat.scripts.nexusformat:main" nxcheck = "nexusformat.scripts.nxcheck:main" +nxlint = "nexusformat.scripts.nxlint:main" +nxvalidate = "nexusformat.scripts.nxvalidate:main" nxconsolidate = "nexusformat.scripts.nxconsolidate:main" nxdir = "nexusformat.scripts.nxdir:main" nxduplicate = "nexusformat.scripts.nxduplicate:main" @@ -62,6 +65,7 @@ nxstack = "nexusformat.scripts.nxstack:main" version_file = "src/nexusformat/_version.py" [tool.setuptools.package-data] +"nexusformat.definitions" = ["*.xsd"] "nexusformat.definitions.applications" = ["*.nxdl.xml"] "nexusformat.definitions.base_classes" = ["*.nxdl.xml"] "nexusformat.definitions.contributed_definitions" = ["*.nxdl.xml"] diff --git a/src/nexusformat/nexus/tree.py b/src/nexusformat/nexus/tree.py index edac8ca..79bce50 100644 --- a/src/nexusformat/nexus/tree.py +++ b/src/nexusformat/nexus/tree.py @@ -212,6 +212,7 @@ import re import sys import warnings +import zlib from copy import copy, deepcopy from pathlib import Path from pathlib import PurePosixPath as PurePath @@ -563,6 +564,23 @@ def mtime(self): """Return the modification time of the NeXus file.""" return Path(self._filename).stat().st_mtime + @property + def size(self): + """Return the size of the NeXus file in bytes.""" + return Path(self._filename).stat().st_size + + def file_hash(self, nbytes=65536): + """Return a fast hash of the leading bytes of the NeXus file. + + Computes an Adler-32 checksum over the first `nbytes` of the file + (default 64 KB). This is used as a secondary modification check when + the filesystem mtime has changed but the file size has not, which + can happen when a file is opened in write mode without any data being + written. + """ + with open(self._filename, 'rb') as f: + return zlib.adler32(f.read(nbytes)) + @property def lock(self): """ @@ -713,6 +731,7 @@ def open(self, **kwargs): self._file = self.h5.File(self._filename, self._mode, **kwargs) if self._root: self._root._mtime = self.mtime + self._root._file_size = self.size self.nxpath = '/' def close(self): @@ -730,6 +749,8 @@ def close(self): self.release_lock() try: self._root._mtime = self.mtime + self._root._file_size = self.size + self._root._file_hash = self.file_hash() except Exception: pass @@ -1424,6 +1445,7 @@ def reload(self): self._root._changed = True self._root._file_modified = False self._root._mtime = self.mtime + self._root._file_size = self.size def rename(self, old_path, new_path): """ @@ -6351,6 +6373,8 @@ def __init__(self, *args, **kwargs): self._class = 'NXroot' self._backup = None self._mtime = None + self._file_size = None + self._file_hash = None self._file_modified = False NXgroup.__init__(self, *args, **kwargs) @@ -6386,6 +6410,7 @@ def serialize(self): """Serialize the root group to a dictionary.""" serialized_root = super().serialize() serialized_root['mtime'] = self.mtime + serialized_root['file_size'] = self._file_size return serialized_root @classmethod @@ -6393,6 +6418,7 @@ def deserialize(self, serialized_root): """Deserialize the root group from a dictionary.""" obj = NXgroup.deserialize(serialized_root) obj._mtime = serialized_root['mtime'] + obj._file_size = serialized_root.get('file_size') if ('filename' in serialized_root and Path(serialized_root['filename']).is_file()): obj._file = NXFile(obj._filename, 'r') @@ -6409,13 +6435,40 @@ def reload(self): f"'{self.nxname}' has no associated file to reload") def is_modified(self): - """True if the file has been modified by an external process.""" + """True if the file has been modified by an external process. + + Uses a three-stage check to reduce false positives from operations + that update the filesystem mtime without changing the file content + (e.g., opening the file in write mode without writing anything): + + 1. If the filesystem mtime has not advanced, the file is not modified. + 2. If the mtime *has* advanced but the file size is unchanged *and* + the leading 64 KB of the file hash to the same value as when the + file was last opened or closed, the mtime change is treated as a + false positive and the file is considered unmodified. + 3. Only if the mtime advanced *and* either the size changed or the + leading-bytes hash differs is the file reported as modified. + """ if self._file is None: self._file_modified = False else: - _mtime = self._file.mtime + _stat = Path(self._file.filename).stat() + _mtime = _stat.st_mtime if self._mtime and _mtime > self._mtime: - self._file_modified = True + _size = _stat.st_size + if self._file_size is not None and _size == self._file_size: + try: + _hash = self._file.file_hash() + except PermissionError: + self._file_modified = True + return self._file_modified + if (self._file_hash is not None and + _hash == self._file_hash): + self._file_modified = False + else: + self._file_modified = True + else: + self._file_modified = True else: self._file_modified = False return self._file_modified @@ -6576,6 +6629,8 @@ def nxfile(self, filename): self._attrs._setattrs(root.attrs) self._file = NXFile(self._filename, self._mode) self._mtime = self._file.mtime + self._file_size = self._file.size + self._file_hash = self._file.file_hash() self.set_changed() else: raise NeXusError(f"'{Path(filename).resolve()}' does not exist") diff --git a/src/nexusformat/nexus/validate.py b/src/nexusformat/nexus/validate.py index 867dbe7..9a722e5 100644 --- a/src/nexusformat/nexus/validate.py +++ b/src/nexusformat/nexus/validate.py @@ -6,6 +6,7 @@ # The full license is in the file COPYING, distributed with this software. # ----------------------------------------------------------------------------- import logging +import re import xml.etree.ElementTree as ET from pathlib import Path @@ -123,6 +124,19 @@ def is_valid_link(self, item): else: target = item._target if item.exists(): + if not item.is_external() and isinstance(item, NXlink): + try: + resolved = item.nxlink + if isinstance(resolved, NXlink) and not resolved.exists(): + chain_target = ( + f'{resolved._filename}[{resolved._target}]' + if resolved.is_external() else resolved._target) + self.log(f'This is a broken link to "{target}", ' + f'which itself links to "{chain_target}"', + level='error') + return False + except Exception: + pass if isinstance(item, NXfield): self.log(f'This field is linked to "{target}"', level='info') elif isinstance(item, NXgroup): @@ -232,6 +246,7 @@ def __init__(self, nxclass, definitions=None): """ super().__init__(definitions=definitions) self.nxclass = nxclass + self.symbols = {} if self.nxclass is None or self.nxclass == 'NXgroup': self.xml_dict = None self.valid_class = False @@ -293,7 +308,7 @@ def get_xml_dict(self): xml_extended_dict = parent_validator.get_xml_dict() xml_dict = merge_dicts(xml_dict, xml_extended_dict) if 'symbols' in xml_dict: - self.symbols = xml_dict['symbols']['symbol'] + self.symbols = xml_dict['symbols'].get('symbol', {}) else: self.symbols = {} else: @@ -561,8 +576,8 @@ def validate(self, group, parent=None, indent=0, level=None): definitions=self.definitions) parsed = False if group.nxname in parent_validator.valid_groups: - cls = parent_validator.valid_groups[group.nxname]['@type'] - if group.nxclass != cls: + cls = parent_validator.valid_groups[group.nxname].get('@type') + if cls is not None and group.nxclass != cls: self.log(f'{group.nxname} should have a class of ' f'{cls}, not {group.nxclass}', level='error') parsed = True @@ -996,7 +1011,11 @@ def validate_file(filename, path=None, definitions=None): log_header(validator, filename=filename, path=path) - validator.validate(path) + try: + validator.validate(path) + except NeXusError as e: + logger.error(e) + return return log_summary() @@ -1057,7 +1076,7 @@ def load_application(self, application): 'does not contain the correct root tag.') symbols = xml_root.find('symbols') if symbols is not None: - self.symbols.update(xml_to_dict(symbols)['symbol']) + self.symbols.update(xml_to_dict(symbols).get('symbol', {})) xml_dict = xml_to_dict(xml_root.find('group')) if xml_root.attrib['extends'] != 'NXobject': xml_extended_dict = self.load_application( @@ -1198,36 +1217,150 @@ def validate_application(filename, path=None, application=None, A tuple containing the total number of warnings and errors encountered while validating the file. """ - with nxopen(filename) as root: - if path is None: - nxpath = root.NXentry[0].nxpath - else: - nxpath = path - entry = root[nxpath] - if not (isinstance(entry, NXentry) or isinstance(entry, NXsubentry)): - logger.error( - f'Path "{nxpath}" is not a NXentry or NXsubentry group') - return - elif application is None and 'definition' in entry: - application = entry['definition'].nxvalue - elif application is None: - logger.error(f'No application definition is defined in "{nxpath}"') - return + if not Path(filename).exists(): + logger.error(f'File "{filename}" does not exist') + return + try: + with nxopen(filename) as root: + if path is None: + nxpath = root.NXentry[0].nxpath + else: + nxpath = path + entry = root[nxpath] + if not (isinstance(entry, NXentry) + or isinstance(entry, NXsubentry)): + logger.error( + f'Path "{nxpath}" is not a NXentry or NXsubentry group') + return + elif application is None and 'definition' in entry: + application = entry['definition'].nxvalue + elif application is None: + logger.error( + f'No application definition is defined in "{nxpath}"') + return - try: - validator = ApplicationValidator(application, - definitions=definitions) - except NeXusError as e: - logger.error(e) - return + try: + validator = ApplicationValidator(application, + definitions=definitions) + except NeXusError as e: + logger.error(e) + return - log_header(validator, filename, nxpath, application) + log_header(validator, filename, nxpath, application) + + lint_results = lint_nxdl(validator.filepath, + definitions=definitions) + if lint_results: + n_errors = sum( + 1 for _, _, sev in lint_results if sev == 'error') + n_warnings = sum( + 1 for _, _, sev in lint_results if sev == 'warning') + issues = [] + if n_errors: + issues.append(f'{n_errors} error(s)') + if n_warnings: + issues.append(f'{n_warnings} warning(s)') + if Path(application).exists(): + nxlint_cmd = f'nxlint {application}' + elif definitions: + nxlint_cmd = f'nxlint -d {definitions} {application}' + else: + nxlint_cmd = f'nxlint {application}' + logger.warning( + f'The application definition has ' + f'{" and ".join(issues)} that may affect validation. ' + f'Run \'{nxlint_cmd}\' for details.') - validator.validate(entry) + validator.validate(entry) + except NeXusError as e: + logger.error(e) + return return log_summary() +def lint_nxdl(filepath, definitions=None): + """ + Checks an NXDL file for structural errors. + + Performs validation using lxml against the nxdl.xsd schema file, if + it is present in the definitions directory, and also checks for the + most common NXDL authoring errors: nested elements. + + Parameters + ---------- + filepath : str or Path + Path to the NXDL file to lint. + definitions : str or Path, optional + Path to the NeXus definitions directory (used to locate nxdl.xsd). + Defaults to the bundled definitions. + + Returns + ------- + list of tuple + A list of (message, location_hint, severity) tuples where severity is + 'error' or 'warning'. An empty list means no issues were found. + """ + filepath = Path(filepath) + if not filepath.exists(): + stem = filepath.stem if filepath.suffix else filepath.name + defs = get_definitions(definitions=definitions) + for search_dir in [defs / 'applications', + defs / 'contributed_definitions', + defs / 'base_classes']: + candidate = search_dir / f'{stem}.nxdl.xml' + if candidate.exists(): + filepath = candidate + break + else: + return [(f'NXDL definition "{stem}" not found in the definitions ' + f'directory', str(defs), 'error')] + from lxml import etree as lxml_etree + + filepath = filepath.resolve() + results = [] + + try: + doc = lxml_etree.parse(str(filepath)) + except lxml_etree.XMLSyntaxError as e: + return [(f'XML syntax error: {e}', str(filepath), 'error')] + + def walk(elem): + if callable(elem.tag): # skip comments, PIs, and other non-elements + return + local = elem.tag.split('}')[-1] if '}' in elem.tag else elem.tag + if local == 'field': + field_name = elem.get('name', '(unnamed)') + for child in elem: + if callable(child.tag): + continue + child_local = (child.tag.split('}')[-1] + if '}' in child.tag else child.tag) + if child_local == 'field': + child_name = child.get('name', '(unnamed)') + results.append(( + f'Nested inside ' + f' is not allowed; ' + f'use instead', + f'line {child.sourceline}', 'error')) + for child in elem: + walk(child) + + walk(doc.getroot()) + + definitions_path = get_definitions(definitions=definitions) + xsd_path = definitions_path / 'nxdl.xsd' + if xsd_path.exists(): + schema = lxml_etree.XMLSchema(lxml_etree.parse(str(xsd_path))) + if not schema.validate(doc): + for err in schema.error_log: + msg = re.sub(r'\{[^}]+\}', '', err.message) + results.append((msg, f'line {err.line}', 'error')) + + results.sort(key=lambda r: int(r[1].split()[-1])) + return results + + def inspect_base_class(base_class, definitions=None): """ Prints the valid components of a NeXus base class. @@ -1249,7 +1382,7 @@ def inspect_base_class(base_class, definitions=None): log(f"NXDL File: {truncate_path(validator.filepath)}\n") else: log(f'NXDL file for "{base_class}" does not exist') - log(f"Definitions: {truncate_path(validator.filepath)}\n") + log(f"Definitions: {truncate_path(validator.definitions)}\n") return tree = ET.parse(validator.filepath) diff --git a/src/nexusformat/scripts/nxlint.py b/src/nexusformat/scripts/nxlint.py new file mode 100644 index 0000000..fe18427 --- /dev/null +++ b/src/nexusformat/scripts/nxlint.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python +# ----------------------------------------------------------------------------- +# Copyright (c) 2025-2026, NeXpy Development Team. +# +# Distributed under the terms of the Modified BSD License. +# +# The full license is in the file COPYING, distributed with this software. +# ----------------------------------------------------------------------------- + +import argparse +import logging + +import nexusformat +from nexusformat.nexus.validate import lint_nxdl, log, log_summary, logger + + +def main(): + parser = argparse.ArgumentParser( + prog="nxlint", + description="Check NXDL application definition files for structural " + "errors.") + parser.add_argument("filename", nargs='+', + help="NXDL file(s) to lint") + parser.add_argument("-d", "--definitions", nargs=1, + help="path to the directory containing NeXus definitions") + parser.add_argument("-i", "--info", action='store_true', + help="output info messages in addition to warnings and errors") + parser.add_argument("-w", "--warning", action='store_true', + help="output warning and error messages (default)") + parser.add_argument("-e", "--error", action='store_true', + help="output errors only") + parser.add_argument('-v', '--version', action='version', + version='%(prog)s v'+nexusformat.__version__) + args = parser.parse_args() + + if args.info: + logger.setLevel(logging.INFO) + elif args.warning: + logger.setLevel(logging.WARNING) + elif args.error: + logger.setLevel(logging.ERROR) + else: + logger.setLevel(logging.WARNING) + + if args.definitions: + definitions = args.definitions[0] + else: + definitions = None + + for filename in args.filename: + log("\n", level='all') + log(f"NXDL file: {filename}", level='all') + if definitions: + log(f"Definitions: {definitions}", level='all') + log("\n", level='all') + logger.total = {'warning': 0, 'error': 0} + + results = lint_nxdl(filename, definitions=definitions) + if results: + for message, location, severity in results: + log(f'[{location}] {message}', level=severity) + log('\nFor help interpreting these errors, consult the NXDL ' + 'reference at https://manual.nexusformat.org/nxdl.html', + level='all') + else: + log(f'No structural errors found in "{filename}"', level='all') + + log_summary() + + +if __name__ == "__main__": + main() diff --git a/src/nexusformat/scripts/nxvalidate.py b/src/nexusformat/scripts/nxvalidate.py new file mode 100644 index 0000000..4f80224 --- /dev/null +++ b/src/nexusformat/scripts/nxvalidate.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python +# ----------------------------------------------------------------------------- +# Copyright (c) 2025-2026, NeXpy Development Team. +# +# Distributed under the terms of the Modified BSD License. +# +# The full license is in the file COPYING, distributed with this software. +# ----------------------------------------------------------------------------- + +import argparse +import logging + +import nexusformat +from nexusformat.nexus.validate import logger, validate_application + + +def main(): + parser = argparse.ArgumentParser( + prog="nxvalidate", + description="Validate NeXus files against an application definition.") + parser.add_argument("filename", nargs=1, + help="name of the NeXus file to be validated") + parser.add_argument("-p", "--path", nargs=1, + help="path to NXentry group to be validated in the NeXus file") + parser.add_argument("-a", "--application", nargs='?', const=True, + help="application definition to validate against; if omitted, " + "the definition is read from the file") + parser.add_argument("-d", "--definitions", nargs=1, + help="path to the directory containing NeXus definitions") + parser.add_argument("-i", "--info", action='store_true', + help="output info messages in addition to warnings and errors") + parser.add_argument("-w", "--warning", action='store_true', + help="output warning and error messages (default)") + parser.add_argument("-e", "--error", action='store_true', + help="output errors only") + parser.add_argument('-v', '--version', action='version', + version='%(prog)s v'+nexusformat.__version__) + args = parser.parse_args() + + if args.info: + logger.setLevel(logging.INFO) + elif args.warning: + logger.setLevel(logging.WARNING) + elif args.error: + logger.setLevel(logging.ERROR) + else: + logger.setLevel(logging.WARNING) + + if args.definitions: + definitions = args.definitions[0] + else: + definitions = None + + if args.filename: + filename = args.filename[0] + if args.path: + path = args.path[0] + else: + path = None + if args.application is True or args.application is None: + application = None + else: + application = args.application + validate_application(filename, path=path, application=application, + definitions=definitions) + else: + parser.print_help() + + +if __name__ == "__main__": + main()