qmk_firmware

QMK firmware for my keyboards (Corne, Sweep Ferris) and trackball (Ploopy Adept)
Log | Files | Refs | Submodules | LICENSE

json_schema.py (4631B)

      1 """Functions that help us generate and use info.json files.
      2 """
      3 import json
      4 import hjson
      5 import jsonschema
      6 from collections.abc import Mapping
      7 from functools import lru_cache
      8 from typing import OrderedDict
      9 from pathlib import Path
     10 from copy import deepcopy
     11 
     12 from milc import cli
     13 
     14 from qmk.util import maybe_exit
     15 
     16 
     17 def _dict_raise_on_duplicates(ordered_pairs):
     18     """Reject duplicate keys."""
     19     d = {}
     20     for k, v in ordered_pairs:
     21         if k in d:
     22             raise ValueError("duplicate key: %r" % (k,))
     23         else:
     24             d[k] = v
     25     return d
     26 
     27 
     28 @lru_cache(maxsize=20)
     29 def _json_load_impl(json_file, strict=True):
     30     """Load a json file from disk.
     31 
     32     Note: file must be a Path object.
     33     """
     34     try:
     35         # Get the IO Stream for Path objects
     36         # Not necessary if the data is provided via stdin
     37         if isinstance(json_file, Path):
     38             json_file = json_file.open(encoding='utf-8')
     39         return hjson.load(json_file, object_pairs_hook=_dict_raise_on_duplicates if strict else None)
     40 
     41     except (json.decoder.JSONDecodeError, hjson.HjsonDecodeError) as e:
     42         cli.log.error('Invalid JSON encountered attempting to load {fg_cyan}%s{fg_reset}:\n\t{fg_red}%s', json_file, e)
     43         maybe_exit(1)
     44     except Exception as e:
     45         cli.log.error('Unknown error attempting to load {fg_cyan}%s{fg_reset}:\n\t{fg_red}%s', json_file, e)
     46         maybe_exit(1)
     47 
     48 
     49 def json_load(json_file, strict=True):
     50     return deepcopy(_json_load_impl(json_file=json_file, strict=strict))
     51 
     52 
     53 @lru_cache(maxsize=20)
     54 def load_jsonschema(schema_name):
     55     """Read a jsonschema file from disk.
     56     """
     57     if Path(schema_name).exists():
     58         return json_load(schema_name)
     59 
     60     schema_path = Path(f'data/schemas/{schema_name}.jsonschema')
     61 
     62     if not schema_path.exists():
     63         schema_path = Path('data/schemas/false.jsonschema')
     64 
     65     return json_load(schema_path)
     66 
     67 
     68 @lru_cache(maxsize=1)
     69 def compile_schema_store():
     70     """Compile all our schemas into a schema store.
     71     """
     72     schema_store = {}
     73 
     74     for schema_file in Path('data/schemas').glob('*.jsonschema'):
     75         schema_data = load_jsonschema(schema_file)
     76         if not isinstance(schema_data, dict):
     77             cli.log.debug('Skipping schema file %s', schema_file)
     78             continue
     79 
     80         # `$id`-based references
     81         schema_store[schema_data['$id']] = schema_data
     82 
     83         # Path-based references
     84         schema_store[Path(schema_file).name] = schema_data
     85 
     86     return schema_store
     87 
     88 
     89 @lru_cache(maxsize=20)
     90 def create_validator(schema):
     91     """Creates a validator for the given schema id.
     92     """
     93     schema_store = compile_schema_store()
     94     resolver = jsonschema.RefResolver.from_schema(schema_store[schema], store=schema_store)
     95 
     96     return jsonschema.Draft202012Validator(schema_store[schema], resolver=resolver).validate
     97 
     98 
     99 def validate(data, schema):
    100     """Validates data against a schema.
    101     """
    102     validator = create_validator(schema)
    103 
    104     return validator(data)
    105 
    106 
    107 def deep_update(origdict, newdict):
    108     """Update a dictionary in place, recursing to do a depth-first deep copy.
    109     """
    110     for key, value in newdict.items():
    111         if isinstance(value, Mapping):
    112             origdict[key] = deep_update(origdict.get(key, {}), value)
    113 
    114         else:
    115             origdict[key] = value
    116 
    117     return origdict
    118 
    119 
    120 def merge_ordered_dicts(dicts):
    121     """Merges nested OrderedDict objects resulting from reading a hjson file.
    122     Later input dicts overrides earlier dicts for plain values.
    123     If any value is "!delete!", the existing value will be removed from its parent.
    124     Arrays will be appended. If the first entry of an array is "!reset!", the contents of the array will be cleared and replaced with RHS.
    125     Dictionaries will be recursively merged. If any entry is "!reset!", the contents of the dictionary will be cleared and replaced with RHS.
    126     """
    127     result = OrderedDict()
    128 
    129     def add_entry(target, k, v):
    130         if k in target and isinstance(v, (OrderedDict, dict)):
    131             if "!reset!" in v:
    132                 target[k] = v
    133             else:
    134                 target[k] = merge_ordered_dicts([target[k], v])
    135             if "!reset!" in target[k]:
    136                 del target[k]["!reset!"]
    137         elif k in target and isinstance(v, list):
    138             if v[0] == '!reset!':
    139                 target[k] = v[1:]
    140             else:
    141                 target[k] = target[k] + v
    142         elif v == "!delete!" and isinstance(target, (OrderedDict, dict)):
    143             del target[k]
    144         else:
    145             target[k] = v
    146 
    147     for d in dicts:
    148         for (k, v) in d.items():
    149             add_entry(result, k, v)
    150 
    151     return result