simulation_db.py

# -*- coding: utf-8 -*-
"""Simulation database

:copyright: Copyright (c) 2015 RadiaSoft LLC.  All Rights Reserved.
:license: http://www.apache.org/licenses/LICENSE-2.0.html
"""
from __future__ import absolute_import, division, print_function
from pykern import pkcollections
from pykern import pkconfig
from pykern import pkinspect
from pykern import pkio
from pykern.pkcollections import PKDict
from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp
from sirepo import feature_config
from sirepo import srschema
from sirepo import srtime
from sirepo import util
from sirepo.template import template_common
import contextlib
import copy
import datetime
import errno
import glob
import numconv
import os
import os.path
import random
import re
import sirepo.const
import sirepo.mpi
import sirepo.resource
import sirepo.srdb
import sirepo.template
import time

#: Names to display to use for jobRunMode

JOB_RUN_MODE_MAP = None

#: Schema common values, e.g. version
SCHEMA_COMMON = None

#: Simulation file name is globally unique to avoid collisions with simulation output
SIMULATION_DATA_FILE = "sirepo-data" + sirepo.const.JSON_SUFFIX

#: where users live under db_dir
USER_ROOT_DIR = "user"

#: Valid characters in ID
_ID_CHARS = numconv.BASE62

#: length of ID
_ID_LEN = 8

#: Relative regexp from ID_Name
_ID_PARTIAL_RE_STR = "[{}]{{{}}}".format(_ID_CHARS, _ID_LEN)

#: Verify ID
_ID_RE = re.compile("^{}$".format(_ID_PARTIAL_RE_STR))

#: where users live under db_dir
_LIB_DIR = "lib"

#: lib relative to sim_dir
_REL_LIB_DIR = "../" + _LIB_DIR

#: Older than any other version
_OLDEST_VERSION = "20140101.000001"

#: Absolute path of rsmanifest file
_RSMANIFEST_PATH = pkio.py_path("/rsmanifest" + sirepo.const.JSON_SUFFIX)

#: Cache of schemas keyed by app name
_SCHEMA_CACHE = PKDict()

#: Special field to direct pseudo-subclassing of schema objects
_SCHEMA_SUPERCLASS_FIELD = "_super"

#: created under dir
_TMP_DIR = "tmp"

#: Use to assert _serial_new result. Not perfect but good enough to avoid common problems
_serial_prev = 0

#: configuration
_cfg = None

#: version for development
_dev_version = None

_SIM_DB_FILE_PATH_RE = re.compile(r"^[a-zA-Z0-9-_\.]{1,128}$")


def app_version():
    """Force the version to be dynamic if running in dev channel

    Returns:
        str: chronological version
    """
    if pkconfig.channel_in("dev"):
        return _timestamp()
    return SCHEMA_COMMON.version


def assert_sid(sid):
    assert _ID_RE.search(sid), "invalid sid={}".format(sid)
    return sid


def cfg():
    return _cfg


def default_data(sim_type):
    """New simulation base data

    Args:
        sim_type (str): simulation type

    Returns:
        dict: simulation data
    """
    from sirepo import sim_data

    return open_json_file(
        sim_type,
        path=sim_data.get_class(sim_type).resource_path(
            f"default-data{sirepo.const.JSON_SUFFIX}",
        ),
    )


def delete_simulation(simulation_type, sid, qcall=None):
    """Deletes the simulation's directory."""
    pkio.unchecked_remove(simulation_dir(simulation_type, sid, qcall=qcall))


def delete_user(qcall):
    """Deletes a user's directory."""
    pkio.unchecked_remove(user_path(qcall=qcall))


def examples(app):
    # TODO(robnagler) Need to update examples statically before build
    # and assert on build
    # example data is not fixed-up to avoid performance problems when searching examples by name
    # fixup occurs during save_new_example()
    from sirepo import sim_data

    return [
        open_json_file(app, path=f, fixup=False)
        for f in sim_data.get_class(app).example_paths()
    ]


def find_user_simulation_copy(sim_type, sid, qcall):
    """ONLY USED BY api_simulationData"""
    rows = iterate_simulation_datafiles(
        sim_type,
        process_simulation_list,
        PKDict({"simulation.outOfSessionSimulationId": sid}),
        qcall=qcall,
    )
    if len(rows):
        return rows[0]["simulationId"]
    return None


def find_global_simulation(sim_type, sid, checked=False):
    paths = pkio.sorted_glob(user_path_root().join("*", sim_type, sid))
    if len(paths) == 1:
        return str(paths[0])
    if len(paths) == 0:
        if checked:
            util.raise_not_found(
                "{}/{}: global simulation not found",
                sim_type,
                sid,
            )
        return None
    util.raise_not_found(
        "{}: more than one path found for simulation={}/{}",
        paths,
        sim_type,
        sid,
    )


def fixup_old_data(data, force=False, path=None, qcall=None):
    """Upgrade data to latest schema and updates version.

    Args:
        data (dict): to be updated (destructively)
        force (bool): force validation
        path (py.path): simulation path, only in a few cases [None]

    Returns:
        dict: upgraded `data`
        bool: True if data changed
    """

    def _last_modified(data, path):
        if not path:
            return srtime.utc_now_as_milliseconds()
        m = 0.0
        for p in pkio.sorted_glob(
            # POSIT: same format as simulation_run_dir
            json_filename(
                template_common.INPUT_BASE_NAME, run_dir=path.dirpath().join("*")
            ),
        ):
            if p.mtime() > m:
                m = p.mtime()
        if m <= 0.0:
            m = path.mtime()
        return int(m * 1000)

    try:
        pkdc(
            "sid={} force={}, version={} SCHEMA_COMMON.version={}",
            data.pkunchecked_nested_get("models.simulation.simulationId"),
            force,
            data.get("version"),
            SCHEMA_COMMON.version,
        )
        if not force and "version" in data and data.version == SCHEMA_COMMON.version:
            return data, False
        try:
            data.fixup_old_version = data.version
        except AttributeError:
            data.fixup_old_version = _OLDEST_VERSION
        data.version = SCHEMA_COMMON.version
        if "simulationType" not in data:
            if "sourceIntensityReport" in data.models:
                data.simulationType = "srw"
            elif "fieldAnimation" in data.models:
                data.simulationType = "warppba"
            elif "bunchSource" in data.models:
                data.simulationType = "elegant"
            else:
                pkdlog("simulationType: not found; data={}", data)
                raise AssertionError("must have simulationType")
        elif data.simulationType == "warp":
            data.simulationType = "warppba"
        elif data.simulationType == "fete":
            data.simulationType = "warpvnd"
        elif data.simulationType == "ml":
            data.simulationType = "activait"
        if "simulationSerial" not in data.models.simulation:
            data.models.simulation.simulationSerial = 0
        if "lastModified" not in data.models.simulation:
            data.models.simulation.lastModified = _last_modified(data, path)
        from sirepo import sim_data

        sim_data.get_class(data.simulationType).fixup_old_data(data, qcall=qcall)
        data.pkdel("fixup_old_version")
        return data, True
    except Exception as e:
        pkdlog("exception={} data={} stack={}", e, data, pkdexc())
        raise


def get_schema(sim_type):
    """Get the schema for `sim_type`

    If sim_type is None, it will return the schema for the first sim_type
    in `feature_config.cfg().sim_types`

    Args:
        sim_type (str): must be valid
    Returns:
        dict: Shared schema

    """
    t = (
        sirepo.template.assert_sim_type(sim_type)
        if sim_type is not None
        else list(feature_config.cfg().sim_types)[0]
    )
    return _SCHEMA_CACHE[t]


def generate_json(data, pretty=False):
    """Convert data to JSON to be send back to client

    Use only for responses. Use `:func:write_json` to save.
    Args:
        data (dict): what to format
        pretty (bool): pretty print [False]
    Returns:
        str: formatted data
    """
    return util.json_dump(data, pretty=pretty)


def init_module():
    # see _init()
    pass


def iterate_simulation_datafiles(simulation_type, op, search=None, qcall=None):
    res = []
    sim_dir = simulation_dir(simulation_type, qcall=qcall)
    for p in pkio.sorted_glob(sim_dir.join("*", SIMULATION_DATA_FILE)):
        try:
            data = open_json_file(
                simulation_type, path=p, fixup=True, qcall=qcall, save=True
            )
            if search and not _search_data(data, search):
                continue
            op(res, p, data)
        except ValueError as e:
            pkdlog("{}: error: {}", p, e)
    return res


def json_filename(filename, run_dir=None):
    """Append sirepo.const.JSON_SUFFIX if necessary and convert to str

    Args:
        filename (py.path or str): to convert
        run_dir (py.path): which directory to join
    Returns:
        py.path: filename.json
    """
    filename = str(filename)
    if not filename.endswith(sirepo.const.JSON_SUFFIX):
        filename += sirepo.const.JSON_SUFFIX
    if run_dir and not os.path.isabs(filename):
        filename = run_dir.join(filename)
    return pkio.py_path(path=filename)


def json_load(*args, **kwargs):
    return pkcollections.json_load_any(*args, **kwargs)


def lib_dir_from_sim_dir(sim_dir):
    """Path to lib dir from simulation dir

    Args:
        sim_dir (py.path): simulation dir or below

    Return:
        py.path: directory name
    """
    return _sim_from_path(sim_dir)[1].join(_REL_LIB_DIR)


def migrate_guest_to_persistent_user(guest_uid, to_uid):
    """Moves all non-example simulations `guest_uid` into `to_uid`.

    Only moves non-example simulations. Doesn't delete the guest_uid.

    Args:
        guest_uid (str): source user
        to_uid (str): dest user

    """
    with util.THREAD_LOCK:
        for path in glob.glob(
            str(user_path(uid=guest_uid).join("*", "*", SIMULATION_DATA_FILE)),
        ):
            data = read_json(path)
            sim = data["models"]["simulation"]
            if "isExample" in sim and sim["isExample"]:
                continue
            dir_path = os.path.dirname(path)
            new_dir_path = dir_path.replace(guest_uid, to_uid)
            pkdlog("{} -> {}", dir_path, new_dir_path)
            pkio.mkdir_parent(new_dir_path)
            os.rename(dir_path, new_dir_path)


def open_json_file(sim_type, path=None, sid=None, fixup=True, save=False, qcall=None):
    """Read a db file and return result

    Args:
        sim_type (str): simulation type (app)
        path (py.path.local): where to read the file
        sid (str): simulation id
        fixup (bool): run fixup_old_data [True]
        save (bool): save_simulation_json if data changed [False]
    Returns:
        dict: data
    """
    p = path or sim_data_file(sim_type, sid, qcall=qcall)
    if not p.exists():
        if path:
            util.raise_not_found("path={} not found", path)
        raise util.SPathNotFound(sim_type=sim_type, sid=sid, uid=_uid_arg(qcall))
    data = None
    try:
        with p.open() as f:
            data = json_load(f)
        # ensure the simulationId matches the path
        if sid:
            data.models.simulation.simulationId = _sim_from_path(p)[0]
    except Exception as e:
        pkdlog("{}: error: {}", p, pkdexc())
        raise
    if not fixup:
        return data
    d, c = fixup_old_data(data, path=p, qcall=qcall)
    if c and save:
        return save_simulation_json(d, fixup=False, do_validate=False, qcall=qcall)
    return d


def parse_sim_ser(data):
    """Extract simulationSerial from data

    Args:
        data (dict): models or request

    Returns:
        int: simulationSerial
    """
    try:
        return int(data["simulationSerial"])
    except KeyError:
        try:
            return int(data["models"]["simulation"]["simulationSerial"])
        except KeyError:
            return None


def prepare_simulation(data, run_dir):
    """Create and install files, update parameters, and generate command.

    Copies files into the simulation directory (``run_dir``)
    Updates the parameters in ``data`` and save.
    Generate the pkcli command.

    Args:
        data (dict): report and model parameters
        run_dir (py.path.local): dir simulation will be run in
    Returns:
        list, py.path: pkcli command, simulation directory
    """
    from sirepo import sim_data

    sim_type = data.simulationType
    template = sirepo.template.import_module(data)
    s = sim_data.get_class(sim_type)
    s.support_files_to_run_dir(data, run_dir)
    update_rsmanifest(data)
    write_json(run_dir.join(template_common.INPUT_BASE_NAME), data)
    # TODO(robnagler) encapsulate in template
    is_p = s.is_parallel(data)
    c = template.write_parameters(
        data,
        run_dir=run_dir,
        is_parallel=is_p,
    )
    if c:
        return c, run_dir
    cmd = [
        pkinspect.root_package(template),
        pkinspect.module_basename(template),
        "run-background" if is_p else "run",
        str(run_dir),
    ]
    return cmd, run_dir


def process_simulation_list(res, path, data):
    sim = data["models"]["simulation"]
    res.append(
        PKDict(
            simulationId=_sim_from_path(path)[0],
            name=sim["name"],
            folder=sim["folder"],
            isExample=sim["isExample"] if "isExample" in sim else False,
            simulation=sim,
        )
    )


def read_json(filename):
    """Read data from json file

    Args:
        filename (py.path or str): will append sirepo.const.JSON_SUFFIX if necessary

    Returns:
        object: json converted to python
    """
    return json_load(json_filename(filename))


def read_simulation_json(sim_type, sid, qcall):
    """Calls `open_json_file` and fixes up data, possibly saving

    Args:
        sim_type (str): simulation type
        sid (str): simulation id

    Returns:
        data (dict): simulation data
    """
    return open_json_file(
        sim_type=sim_type,
        fixup=True,
        save=True,
        sid=sid,
        qcall=qcall,
    )


def save_new_example(data, qcall=None):
    data.models.simulation.isExample = True
    return save_new_simulation(
        data,
        do_validate=False,
        qcall=qcall,
    )


def save_new_simulation(data, do_validate=True, qcall=None):
    d = simulation_dir(data.simulationType, qcall=qcall)
    sid = _random_id(d, data.simulationType).id
    data.models.simulation.simulationId = sid
    return save_simulation_json(
        data,
        do_validate=do_validate,
        qcall=qcall,
        fixup=True,
        modified=True,
    )


def save_simulation_json(data, fixup, do_validate=True, qcall=None, modified=False):
    """Prepare data and save to json db

    Args:
        data (dict): what to write (contains simulationId)
        fixup (bool): whether to run fixup_old_data
        uid (str): user id [None]
        do_validate (bool): call srschema.validate_name [True]
        modified (bool): call prepare_for_save and update lastModified [False]
    """
    if fixup:
        data = fixup_old_data(data, qcall=qcall)[0]
        # we cannot change the logged in user so we need to
        # not run these fixups here, or we'll get recursion as
        # prepare_for_save may ask for the logged in user
        if modified:
            t = sirepo.template.import_module(data.simulationType)
            if hasattr(t, "prepare_for_save"):
                data = t.prepare_for_save(data, qcall=qcall)
    # old implementation value
    data.pkdel("computeJobHash")
    s = data.models.simulation
    sim_type = data.simulationType
    fn = sim_data_file(sim_type, s.simulationId, qcall=qcall)
    with util.THREAD_LOCK:
        need_validate = True
        try:
            # OPTIMIZATION: If folder/name same, avoid reading entire folder
            on_disk = read_json(fn).models.simulation
            need_validate = not (on_disk.folder == s.folder and on_disk.name == s.name)
        except Exception:
            pass
        if need_validate and do_validate:
            srschema.validate_name(
                data,
                iterate_simulation_datafiles(
                    sim_type,
                    lambda res, _, d: res.append(d),
                    PKDict({"simulation.folder": s.folder}),
                    qcall=qcall,
                ),
                SCHEMA_COMMON.common.constants.maxSimCopies,
            )
            srschema.validate_fields(data, get_schema(data.simulationType))
        s.simulationSerial = _serial_new()
        # Do not write simulationStatus or computeJobCacheKey
        d = copy.deepcopy(data)
        pkcollections.unchecked_del(d.models, "simulationStatus", "computeJobCacheKey")
        if modified:
            d.models.simulation.lastModified = srtime.utc_now_as_milliseconds()
        write_json(fn, d)
    return data


def sid_from_compute_file(path):
    """Get sid from path to report file

    Args:
        path (py.path): must be an existing report file

    Returns:
        str: simulation id
    """
    assert path.check(file=1)
    return _sim_from_path(path)[0]


def sim_data_file(sim_type, sim_id, qcall=None):
    """Simulation data file name

    Args:
        sim_type (str): simulation type
        sim_id (str): simulation id
        uid (str): user id

    Returns:
        py.path.local: simulation path
    """
    return simulation_dir(sim_type, sim_id, qcall=qcall).join(SIMULATION_DATA_FILE)


def sim_db_file_uri(simulation_type, sid, basename):
    return "/".join(
        [
            sirepo.template.assert_sim_type(simulation_type),
            assert_sid(sid),
            _assert_sim_db_file_path(basename),
        ]
    )


def sim_db_file_uri_to_path(path, expect_uid):
    p = path.split("/")
    assert len(p) == 4, f"path={p} has too many parts"
    assert p[0] == expect_uid, f"uid={p[0]} is not expect_uid={expect_uid}"
    sirepo.template.assert_sim_type(p[1]),
    assert_sid(p[2]),
    _assert_sim_db_file_path(p[3]),
    return user_path_root().join(*p)


def sim_from_path(path):
    return _sim_from_path(path)


def simulation_dir(simulation_type, sid=None, qcall=None):
    """Generates simulation directory from sid and simulation_type

    Args:
        simulation_type (str): srw, warppba, ...
        sid (str): simulation id (optional)
        uid (str): user id
    """
    p = user_path(qcall=qcall)
    d = p.join(sirepo.template.assert_sim_type(simulation_type))
    with util.THREAD_LOCK:
        if not d.exists():
            _create_lib_and_examples(p, d.basename, qcall=qcall)
    if not sid:
        return d
    return d.join(assert_sid(sid))


def simulation_lib_dir(simulation_type, qcall=None):
    """String name for user library dir

    Args:
        simulation_type: which app is this for

    Return:
        py.path: directory name
    """
    # POSIT: _create_lib_and_examples
    return simulation_dir(simulation_type, qcall=qcall).join(_LIB_DIR)


def simulation_run_dir(req_or_data, remove_dir=False, qcall=None):
    """Where to run the simulation

    Args:
        req_or_data (dict): may be simulation data or a request
        remove_dir (bool): remove the directory [False]

    Returns:
        py.path: directory to run
    """
    from sirepo import sim_data

    t = req_or_data.simulationType
    s = sim_data.get_class(t)
    d = simulation_dir(
        t,
        s.parse_sid(req_or_data),
        qcall=qcall,
    ).join(s.compute_model(req_or_data))
    if remove_dir:
        pkio.unchecked_remove(d)
    return d


def static_libs():
    return _files_in_schema(SCHEMA_COMMON.common.staticFiles)


@contextlib.contextmanager
def tmp_dir(chdir=False, qcall=None):
    """Generates new, temporary directory

    `uid` or `qcall` must be supplied.

    Args:
        chdir (bool): if true, will save_chdir
        uid (str): user id
        qcall (sirepo.quest.API): request state
    Returns:
        py.path: directory to use for temporary work
    """
    d = None
    try:
        p = user_path(qcall=qcall, check=True)
        d = _cfg.tmp_dir or _random_id(p.join(_TMP_DIR))["path"]
        pkio.unchecked_remove(d)
        pkio.mkdir_parent(d)
        if chdir:
            with pkio.save_chdir(d):
                yield d
        else:
            yield d
    finally:
        if d:
            pkio.unchecked_remove(d)


def uid_from_dir_name(dir_name):
    """Extract user id from user_path

    Args:
        dir_name (py.path): must be top level user dir or sim_dir

    Return:
        str: user id
    """
    r = re.compile(
        r"^{}/({})(?:$|/)".format(
            re.escape(str(user_path_root())),
            _ID_PARTIAL_RE_STR,
        ),
    )
    m = r.search(str(dir_name))
    assert m, "{}: invalid user or sim dir; did not match re={}".format(
        dir_name,
        r.pattern,
    )
    return m.group(1)


def update_rsmanifest(data):
    try:
        data.rsmanifest = read_json(_RSMANIFEST_PATH)
    except Exception as e:
        if pkio.exception_is_not_found(e):
            return
        raise


def user_create():
    """Create a user and initialize the directory

    Returns:
        str: New user id
    """
    return _random_id(user_path_root())["id"]


def user_path(uid=None, qcall=None, check=False):
    """Path for uid or root of all users

    Args:
        uid (str): user id (qcall is preferred)
        qcall (quest.API): logged in user
        check (bool): assert directory exists
    Return:
        py.path: root user's directory
    """
    uid = _uid_arg(uid, qcall)
    d = user_path_root().join(uid)
    if check and not d.check():
        raise util.UserDirNotFound(user_dir=d, uid=uid)
    return d


def user_path_root():
    """Path for uid or root of all users

    Return:
        py.path: root of all users
    """
    return sirepo.srdb.root().join(USER_ROOT_DIR)


def validate_serial(req_data, qcall):
    """Verify serial in data validates

    Args:
        req_data (dict): request with serial and possibly models
    """
    if req_data.get("version") != SCHEMA_COMMON.version:
        pkdlog(
            "req_data={} != server={}", req_data.get("version"), SCHEMA_COMMON.version
        )
        raise util.SRException("serverUpgraded", None)
    with util.THREAD_LOCK:
        sim_type = sirepo.template.assert_sim_type(req_data.simulationType)
        sid = req_data.models.simulation.simulationId
        req_ser = req_data.models.simulation.simulationSerial
        curr = read_simulation_json(sim_type, sid=sid, qcall=qcall)
        curr_ser = curr.models.simulation.simulationSerial
        if not req_ser is None:
            if req_ser == curr_ser:
                return
            status = "newer" if req_ser > curr_ser else "older"
        raise util.Error(
            PKDict(
                sim_type=sim_type,
                error="invalidSerial",
                simulationData=req_data,
            ),
            "{}: incoming serial {} than stored serial={} sid={}, resetting client",
            req_ser,
            status,
            curr_ser,
            sid,
        )


def write_json(filename, data):
    """Write data as json to filename

    pretty is true.

    Args:
        filename (py.path or str): will append sirepo.const.JSON_SUFFIX if necessary
    """
    util.json_dump(data, path=json_filename(filename), pretty=True)


def _assert_sim_db_file_path(basename):
    assert _SIM_DB_FILE_PATH_RE.search(basename), f"basename={basename} is invalid"
    return basename


def _create_lib_and_examples(user_dir, sim_type, qcall=None):
    # POSIT: simulation_lib_dir
    pkio.mkdir_parent(user_dir.join(sim_type).join(_LIB_DIR))
    # POSIT: user_dir structure
    uid = user_dir.basename
    for s in examples(sim_type):
        save_new_example(s, qcall=qcall)


def _files_in_schema(schema):
    """Relative paths of local and external files of the given load and file type listed in the schema
    The order matters for javascript files

    Args:
        schema (PKDict): schema (or portion thereof) to inspect

    Returns:
        str: combined list of local and external file paths, mapped by type
    """
    paths = PKDict(css=[], js=[])
    for source, path in (("externalLibs", "ext"), ("sirepoLibs", "")):
        for file_type, files in schema[source].items():
            for f in files:
                paths[file_type].append(
                    sirepo.resource.static_url(file_type, path, f),
                )
    return paths


def _init():
    global _cfg, JOB_RUN_MODE_MAP

    _cfg = pkconfig.init(
        sbatch_display=(None, str, "how to display sbatch cluster to user"),
        tmp_dir=(None, pkio.py_path, "Used by utilities (not regular config)"),
        logged_in_user=(None, str, "Used in agents"),
    )
    _init_schemas()
    JOB_RUN_MODE_MAP = PKDict(
        sequential="Serial",
        parallel="{} cores (SMP)".format(sirepo.mpi.cfg().cores),
    )
    if _cfg.sbatch_display:
        JOB_RUN_MODE_MAP.sbatch = _cfg.sbatch_display


def _init_schemas():
    global SCHEMA_COMMON
    SCHEMA_COMMON = json_load(
        sirepo.resource.static("json", f"schema-common{sirepo.const.JSON_SUFFIX}")
    )
    a = SCHEMA_COMMON.appInfo
    for t in sirepo.feature_config.cfg().sim_types:
        s = read_json(sirepo.resource.static("json", f"{t}-schema.json"))
        _merge_dicts(s.get("appInfo", PKDict()), a)
        s.update(SCHEMA_COMMON)
        s.feature_config = feature_config.for_sim_type(t)
        s.simulationType = t

        # TODO(mvk): improve merging common and local schema
        _merge_dicts(s.common.dynamicFiles, s.dynamicFiles)
        s.dynamicModules = _files_in_schema(s.dynamicFiles)
        for i in [
            "appDefaults",
            "appModes",
            "constants",
            "cookies",
            "enum",
            "notifications",
            "localRoutes",
            "model",
            "strings",
            "view",
        ]:
            if i not in s:
                s[i] = PKDict()
            _merge_dicts(s.common[i], s[i])
        _merge_subclasses(s, "model", extend_arrays=False)
        _merge_subclasses(s, "view", extend_arrays=True)
        srschema.validate(s)
        _SCHEMA_CACHE[t] = s
    SCHEMA_COMMON.appInfo = a
    for s in _SCHEMA_CACHE.values():
        s.appInfo = a
    # In development, any schema update creates a new version
    if pkconfig.channel_in("dev"):
        SCHEMA_COMMON.version = str(sirepo.srtime.utc_now_as_float())
    else:
        SCHEMA_COMMON.version = max(
            [m.__version__ for m in sirepo.resource.root_modules()]
        )


def _merge_dicts(base, derived, depth=-1, extend_arrays=True):
    """Copy the items in the base dictionary into the derived dictionary, to the specified depth

    Args:
        base (dict): source
        derived (dict): receiver
        depth (int): how deep to recurse:
            >= 0:  <depth> levels
            < 0:   all the way
        extend_arrays (bool): if True, merging will extend arrays that exist in both
        dicts. Otherwise, the arrays are replaced
    """
    if depth == 0:
        return
    for key in base:
        # Items with the same name are not replaced
        if key not in derived:
            derived[key] = base[key]
        else:
            try:
                if extend_arrays:
                    _extend_no_dupes(derived[key], base[key])
            except AttributeError:
                # The value was not an array
                pass
        try:
            _merge_dicts(base[key], derived[key], depth - 1 if depth > 0 else depth)
        except TypeError:
            # The value in question is not itself a dict, move on
            pass


def _merge_subclasses(schema, item, extend_arrays=True):
    for m in schema[item]:
        item_schema = schema[item]
        model = item_schema[m]
        subclasses = []
        _unnest_subclasses(schema, item, m, subclasses)
        for s in subclasses:
            _merge_dicts(item_schema[s], model, extend_arrays=extend_arrays)
        # _super is a special case
        if subclasses:
            _extend_no_dupes(model[_SCHEMA_SUPERCLASS_FIELD], subclasses)


def _extend_no_dupes(arr1, arr2):
    arr1.extend(x for x in arr2 if x not in arr1)


def _unnest_subclasses(schema, item, key, subclass_keys):
    item_schema = schema[item]
    try:
        if _SCHEMA_SUPERCLASS_FIELD not in item_schema[key]:
            return
    except TypeError:
        # Ignore non-indexable types
        return
    sub_model = item_schema[key]
    sub_item = sub_model[_SCHEMA_SUPERCLASS_FIELD][1]
    sub_key = sub_model[_SCHEMA_SUPERCLASS_FIELD][2]
    assert sub_item in schema, util.err(sub_item, "No such field in schema")
    assert sub_item == item, util.err(
        sub_item, "Superclass must be in same section of schema {}", item
    )
    assert sub_key in item_schema, util.err(sub_key, "No such superclass")
    subclass_keys.append(sub_key)
    _unnest_subclasses(schema, item, sub_key, subclass_keys)


def _random_id(parent_dir, simulation_type=None):
    """Create a random id in parent_dir

    Args:
        parent_dir (py.path): where id should be unique
    Returns:
        dict: id (str) and path (py.path)
    """
    pkio.mkdir_parent(parent_dir)
    r = random.SystemRandom()
    # Generate cryptographically secure random string
    for _ in range(5):
        i = "".join(r.choice(_ID_CHARS) for x in range(_ID_LEN))
        if simulation_type:
            if find_global_simulation(simulation_type, i):
                continue
        d = parent_dir.join(i)
        try:
            os.mkdir(str(d))
            return PKDict(id=i, path=d)
        except OSError as e:
            if e.errno == errno.EEXIST:
                pass
            raise
    raise RuntimeError("{}: failed to create unique directory".format(parent_dir))


def _search_data(data, search):
    for field, expect in search.items():
        path = field.split(".")
        if len(path) == 1:
            # TODO(robnagler) is this a bug? Why would you supply a search path
            # value that didn't want to be searched.
            continue
        path.insert(0, "models")
        v = data
        for key in path:
            if key in v:
                v = v[key]
        if v != expect:
            return False
    return True


def _serial_new():
    """Generate a serial number

    Serial numbers are 16 digits (time represented in microseconds
    since epoch) which are always less than Javascript's
    Number.MAX_SAFE_INTEGER (9007199254740991=2*53-1).

    Timestamps are not guaranteed to be sequential. If the
    system clock is adjusted, we'll throw an exception here.
    """
    global _serial_prev
    res = int(time.time() * 1000000)
    with util.THREAD_LOCK:
        # Good enough assertion. Any collisions will also be detected
        # by parameter hash so order isn't only validation
        assert res > _serial_prev, "{}: serial did not increase: prev={}".format(
            res, _serial_prev
        )
        _serial_prev = res
    return res


def _sim_from_path(path):
    prev = None
    p = path
    # SECURITY: go up three levels at most (<type>/<id>/<report>/<output>)
    for _ in range(3):
        if p == prev:
            break
        i = p.basename
        if _ID_RE.search(i):
            return i, p
        prev = p
        p = p.dirpath()
    raise AssertionError("path={} is not valid simulation".format(path))


def _timestamp(time=None):
    if not time:
        time = datetime.datetime.utcnow()
    elif not isinstance(time, datetime.datetime):
        time = datetime.datetime.fromtimestamp(time)
    return time.strftime("%Y%m%d.%H%M%S")


def _uid_arg(uid=None, qcall=None):
    if uid:
        return uid
    if qcall:
        # Avoid recursion to user_path with check_path=False
        return qcall.auth.logged_in_user(check_path=False)
    uid = _cfg.logged_in_user
    assert uid, "uid not supplied and no logged_in_user config"
    return uid


_init()