239 lines
11 KiB
Python
239 lines
11 KiB
Python
import argparse
|
|
import ast
|
|
import copy
|
|
import hashlib
|
|
import inspect
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
|
|
|
|
MANAGED = {'starspec', 'restartfile', 'runscript', 'resources', 'landmap', 'topomap', 'otherargs', 'runsteps', 'noutput'}
|
|
|
|
|
|
def _type(value):
|
|
if isinstance(value, bool):
|
|
return 'boolean'
|
|
if isinstance(value, int):
|
|
return 'integer'
|
|
if isinstance(value, float):
|
|
return 'number'
|
|
if isinstance(value, dict):
|
|
return 'object'
|
|
if isinstance(value, list):
|
|
return 'array'
|
|
return 'string'
|
|
|
|
|
|
def catalog(config):
|
|
import exoplasim
|
|
model = exoplasim.Model
|
|
signature = inspect.signature(model.configure)
|
|
documentation = inspect.getdoc(model.configure)
|
|
entries, group, active = {}, 'Parameters', None
|
|
for line in documentation.splitlines():
|
|
if line.strip() in ('Notes', 'References'):
|
|
active = None
|
|
heading = re.fullmatch(r'\s*\*\*(.+?)\*\*\s*', line)
|
|
if heading:
|
|
group, active = heading[1], None
|
|
match = re.match(r'^\s*(\w+)\s*[:;]\s*(.+)', line)
|
|
if match and match[1] in signature.parameters:
|
|
active = match[1]
|
|
entries[active] = {'declaration': match[2], 'description': '', 'group': group}
|
|
elif active and line.strip():
|
|
entries[active]['description'] += ' ' + line.strip()
|
|
source = inspect.getsource(model.configure)
|
|
tree = ast.parse(Path(inspect.getsourcefile(model)).read_text())
|
|
fields = []
|
|
for name, parameter in signature.parameters.items():
|
|
if name == 'self':
|
|
continue
|
|
default = copy.deepcopy(parameter.default)
|
|
doc = entries.get(name, {})
|
|
declaration = doc.get('declaration', '')
|
|
kind = _type(default) if default is not None else ('integer' if re.search(r'\bint(?:eger)?\b', declaration) else 'string' if ('str' in declaration or '{' in declaration) else 'number')
|
|
field = {'name': name, 'label': name, 'type': kind, 'default': default, 'nullable': default is None,
|
|
'description': doc.get('description', '').strip(), 'group': doc.get('group', 'Parameters'), 'editable': name not in MANAGED}
|
|
if declaration.startswith('{'):
|
|
choices = [s.strip().strip('\"\' }') for s in declaration.split('}', 1)[0].lstrip('{').split(',')]
|
|
choices = [s for s in choices if s and s != 'None']
|
|
if all(re.fullmatch(r'-?\d+', s) for s in choices):
|
|
field['type'], field['choices'] = 'integer', [int(s) for s in choices]
|
|
else:
|
|
field['type'], field['choices'] = 'string', choices
|
|
if isinstance(default, dict):
|
|
field['properties'] = {key: {'type': _type(value), 'default': value} for key, value in default.items()}
|
|
if name == 'gravity':
|
|
field['implicit_default'] = next(node.value.value for node in ast.walk(tree) if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == name for target in node.targets) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, (int, float)))
|
|
if name == 'ozone':
|
|
keys = sorted({node.slice.value for node in ast.walk(tree) if isinstance(node, ast.Subscript) and isinstance(node.value, ast.Name) and node.value.id == name and isinstance(node.slice, ast.Constant) and isinstance(node.slice.value, str)})
|
|
field.update(type='object', types=['boolean', 'object'], properties={key: {'type': 'number'} for key in keys}, required=keys)
|
|
if name == 'vegetation':
|
|
field.update(type='integer', types=['boolean', 'integer'], choices=[0, 1, 2])
|
|
if name == 'physicsfilter':
|
|
field['pattern'] = r'^(?:(?:gp|sp|none|cesaro|exp|lh)(?:\|(?:gp|sp|none|cesaro|exp|lh))*)?$'
|
|
if name == 'aerofile':
|
|
field['choices'] = sorted(p.name for p in (Path(exoplasim.__file__).parent / 'hazeconstants').glob('*.dat'))
|
|
if field['type'] == 'string' and not field.get('choices') and not field.get('pattern'):
|
|
field['editable'] = False
|
|
lead = field['description'].split('.', 1)[0].split(',', 1)[0]
|
|
lead = re.split(r'\s+in\s+', lead, maxsplit=1)[0]
|
|
if 3 < len(lead) < 55 and not re.match(r'^(True|False|If|Can|A |The |Whether)', lead):
|
|
field['label'] = lead
|
|
fields.append(field)
|
|
limits = config['limits']
|
|
constructor = inspect.signature(model)
|
|
model_defaults = {key: constructor.parameters[key].default for key in ('resolution', 'layers', 'ncpus')}
|
|
model_defaults['ncpus'] = min(model_defaults['ncpus'], limits['max_cpus'])
|
|
model_rules = {'resolution': {'type': 'string', 'choices': limits['resolutions']},
|
|
'layers': {'type': 'integer', 'min': 2, 'max': limits['max_layers']},
|
|
'ncpus': {'type': 'integer', 'min': 1, 'max': limits['max_cpus']}}
|
|
run_rules = {'cycles': {'type': 'integer', 'min': 1, 'max': limits['max_cycles']},
|
|
'steps': {'type': 'integer', 'nullable': True, 'min': 1, 'max': limits['max_steps']}}
|
|
if __package__:
|
|
from .terrain import surface_schema
|
|
from .mapcatalog import map_schema
|
|
else:
|
|
from terrain import surface_schema
|
|
from mapcatalog import map_schema
|
|
maps = map_schema()
|
|
maps['grids'] = {key: maps['grids'][key] for key in limits['resolutions'] if key in maps['grids']}
|
|
return {'fields': fields, 'model': model_rules, 'run': run_rules, 'model_defaults': model_defaults,
|
|
'run_defaults': config['run_defaults'], 'surface': surface_schema(), 'maps': maps,
|
|
'map_fields': maps['fields'], 'map_grids': maps['grids'], 'limits': limits,
|
|
'source_hash': hashlib.sha256(source.encode()).hexdigest()}
|
|
|
|
|
|
def validate_value(value, rule):
|
|
if value is None:
|
|
return (None, None) if rule.get('nullable') else (value, 'A value is required.')
|
|
types = rule.get('types', [rule['type']])
|
|
actual = _type(value)
|
|
if actual == 'integer' and 'number' in types:
|
|
actual = 'number'
|
|
if actual not in types:
|
|
return value, 'Expected ' + ' or '.join(types) + '.'
|
|
if isinstance(value, (float, int)) and not isinstance(value, bool):
|
|
try:
|
|
finite = math.isfinite(value)
|
|
except OverflowError:
|
|
finite = False
|
|
if not finite:
|
|
return value, 'Use a finite number.'
|
|
if 'min' in rule and value < rule['min'] or 'max' in rule and value > rule['max']:
|
|
return value, 'The value is outside the allowed range.'
|
|
if 'choices' in rule and value not in rule['choices']:
|
|
return value, 'Select an allowed value.'
|
|
if isinstance(value, str) and (len(value) > 200 or 'pattern' in rule and not re.fullmatch(rule['pattern'], value)):
|
|
return value, 'The value has an invalid format.'
|
|
if isinstance(value, dict):
|
|
properties = rule.get('properties', {})
|
|
unknown = set(value) - set(properties)
|
|
if unknown:
|
|
return value, 'Unknown keys: ' + ', '.join(sorted(unknown)) + '.'
|
|
result = copy.deepcopy(rule.get('default')) if isinstance(rule.get('default'), dict) else {}
|
|
for key, item in value.items():
|
|
normalized, error = validate_value(item, properties[key])
|
|
if error:
|
|
return value, key + ': ' + error
|
|
result[key] = normalized
|
|
missing = set(rule.get('required', [])) - set(result)
|
|
if missing:
|
|
return value, 'Required keys: ' + ', '.join(sorted(missing)) + '.'
|
|
value = result
|
|
return value, None
|
|
|
|
|
|
def validate_spec(spec, config):
|
|
schema = json.loads(Path(config['_paths']['catalog']).read_text())
|
|
errors, normalized = {}, {}
|
|
if not isinstance(spec, dict):
|
|
return {'ok': False, 'spec': spec, 'errors': {'spec': 'Expected an object.'}}
|
|
for key in set(spec) - {'model', 'run', 'parameters', 'surface', 'maps'}:
|
|
errors[key] = 'Unknown spec section.'
|
|
for section in ('model', 'run'):
|
|
values = spec.get(section, {})
|
|
if not isinstance(values, dict):
|
|
errors[section] = 'Expected an object.'
|
|
continue
|
|
normalized[section] = copy.deepcopy(schema[section + '_defaults'])
|
|
for key, value in values.items():
|
|
if key not in schema[section]:
|
|
errors[section + '.' + key] = 'Unknown field.'
|
|
continue
|
|
result, error = validate_value(value, schema[section][key])
|
|
if error:
|
|
errors[section + '.' + key] = error
|
|
else:
|
|
normalized[section][key] = result
|
|
parameters = spec.get('parameters', {})
|
|
if parameters == []:
|
|
parameters = {}
|
|
normalized['parameters'] = {}
|
|
fields = {field['name']: field for field in schema['fields']}
|
|
if not isinstance(parameters, dict):
|
|
errors['parameters'] = 'Expected an object.'
|
|
else:
|
|
for name, value in parameters.items():
|
|
field = fields.get(name)
|
|
if not field:
|
|
errors['parameters.' + name] = 'Unknown parameter.'
|
|
continue
|
|
if not field['editable']:
|
|
if value != field['default']:
|
|
errors['parameters.' + name] = 'This input is managed by the application.'
|
|
continue
|
|
result, error = validate_value(value, field)
|
|
if error:
|
|
errors['parameters.' + name] = error
|
|
else:
|
|
normalized['parameters'][name] = result
|
|
if 'maps' in spec:
|
|
maps = spec['maps']
|
|
rules = schema.get('maps', {}).get('fields', {})
|
|
if not isinstance(maps, dict):
|
|
errors['maps'] = 'Expected a map version object.'
|
|
else:
|
|
normalized['maps'] = {}
|
|
for field, version in maps.items():
|
|
if field not in rules:
|
|
errors['maps.' + field] = 'Unsupported map field.'
|
|
elif not isinstance(version, str) or not re.fullmatch(r'[a-f0-9]{32}', version):
|
|
errors['maps.' + field] = 'Expected a saved map version ID.'
|
|
else:
|
|
normalized['maps'][field] = version
|
|
if 'surface' in spec:
|
|
surface = spec['surface']
|
|
rules = schema['surface']
|
|
if not isinstance(surface, dict):
|
|
errors['surface'] = 'Expected an object.'
|
|
else:
|
|
normalized['surface'] = {}
|
|
for key in set(surface) - set(rules):
|
|
errors['surface.' + key] = 'Unknown field.'
|
|
for key, rule in rules.items():
|
|
result, error = validate_value(surface.get(key, rule.get('default')), rule)
|
|
if error:
|
|
errors['surface.' + key] = error
|
|
else:
|
|
normalized['surface'][key] = result
|
|
return {'ok': not errors, 'spec': normalized, 'errors': errors}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
from config import load_config
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--config')
|
|
parser.add_argument('--output')
|
|
args = parser.parse_args()
|
|
configuration = load_config(args.config)
|
|
output = Path(args.output or configuration['_paths']['catalog'])
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = output.with_name(output.name + '.' + str(os.getpid()) + '.tmp')
|
|
temporary.write_text(json.dumps(catalog(configuration), indent=2) + '\n')
|
|
temporary.chmod(0o644)
|
|
temporary.replace(output)
|