Add get_model_spec module for API access to model spec files

Author: taldcroft

docs/api.rst

@@ -27,3 +27,9 @@ Mask classes
 
 .. automodule:: xija.component.mask
    :members:
+
+Get model spec
+--------------
+.. automodule:: xija.get_model_spec
+   :members:
+

xija/get_model_spec.py

@@ -0,0 +1,164 @@
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
+"""
+Get Chandra model specifications
+"""
+import os
+import re
+from pathlib import Path
+from typing import List, Optional
+
+from git import Repo
+import requests
+from Ska.File import get_globfiles
+
+__all__ = ['get_xija_model_file', 'get_xija_model_names', 'get_repo_version',
+           'check_github_version']
+
+REPO_PATH = Path(os.environ['SKA'], 'data', 'chandra_models')
+MODELS_PATH = REPO_PATH / 'chandra_models' / 'xija'
+CHANDRA_MODELS_URL = 'https://api.github.com/repos/sot/chandra_models/releases'
+
+
+def get_xija_model_file(model_name, models_path=MODELS_PATH) -> str:
+    """
+    Get file name of Xija model specification for the specified ``model_name``.
+
+    Supported model names include (but are not limited to): ``'aca'``,
+    ``'acisfp'``, ``'dea'``, ``'dpa'``, ``'psmc'``, ``'minusyz'``, and
+    ``'pftank2t'``.
+
+    Use ``get_xija_model_names()`` for the full list.
+
+    Examples
+    --------
+    >>> import xija
+    >>> from xija.get_model_spec import get_xija_model_file
+    >>> model_spec = get_xija_model_file('acisfp')
+    >>> model = xija.XijaModel('acisfp', model_spec=model_spec, start='2012:001', stop='2012:010')
+    >>> model.make()
+    >>> model.calc()
+
+    Parameters
+    ----------
+    model_name : str
+        Name of model
+    models_path : str, Path
+        Path to directory containing xija model spec files (default is
+        $SKA/data/chandra_models)
+
+    Returns
+    -------
+    str
+        File name of the corresponding Xija model specification
+    """
+    models_path = Path(models_path)
+
+    if not models_path.exists():
+        raise FileNotFoundError(f'xija models directory {models_path} does not exist')
+
+    file_glob = str(models_path / '*' / f'{model_name.lower()}_spec.json')
+    try:
+        # get_globfiles() default requires exactly one file match and returns a list
+        file_name = get_globfiles(file_glob)[0]
+    except ValueError:
+        names = get_xija_model_names()
+        raise ValueError(f'no models matched {model_name}. Available models are: '
+                         f'{", ".join(names)}')
+
+    return file_name
+
+
+def get_xija_model_names(models_path=MODELS_PATH) -> List[str]:
+    """Return list of available xija model names.
+
+    Parameters
+    ----------
+    models_path : str, Path
+        Path to directory containing xija model spec files (default is
+        $SKA/data/chandra_models)
+
+    Examples
+    --------
+    >>> from xija.get_model_spec import get_xija_model_names
+    >>> names = get_xija_model_names()
+    ['aca',
+     'acisfp',
+     'dea',
+     'dpa',
+     '4rt700t',
+     'minusyz',
+     'pm1thv2t',
+     'pm2thv1t',
+     'pm2thv2t',
+     'pftank2t',
+     'pline03t_model',
+     'pline04t_model',
+     'psmc',
+     'tcylaft6']
+
+    Returns
+    -------
+    list
+        List of available xija model names
+    """
+    models_path = Path(models_path)
+
+    fns = get_globfiles(str(models_path / '*' / '*_spec.json'), minfiles=0, maxfiles=None)
+    names = [re.sub(r'_spec\.json', '', Path(fn).name) for fn in sorted(fns)]
+
+    return names
+
+
+def get_repo_version() -> str:
+    """Return version (most recent tag) of models repository.
+
+    Returns
+    -------
+    str
+        Version (most recent tag) of models repository
+    """
+    repo = Repo(REPO_PATH)
+
+    if repo.is_dirty():
+        raise ValueError('repo is dirty')
+
+    tags = sorted(repo.tags, key=lambda tag: tag.commit.committed_datetime)
+    tag_repo = tags[-1]
+    if tag_repo.commit != repo.head.commit:
+        raise ValueError(f'repo tip is not at tag {tag_repo}')
+
+    return tag_repo.name
+
+
+def check_github_version(tag_name, url=CHANDRA_MODELS_URL, timeout=5) -> Optional[bool]:
+    """Check that latest chandra_models GitHub repo release matches ``tag_name``.
+
+    This queries GitHub for the latest release of chandra_models.
+
+    Parameters
+    ----------
+    tag_name : str
+        Tag name e.g. '3.32'
+    url : str
+        URL for chandra_models releases on GitHub API
+    timeout : int, float
+        Request timeout (sec, default=5)
+
+    Returns
+    -------
+    bool, None
+        True if chandra_models release on GitHub matches tag_name.
+        None if the request timed out, indicating indeterminate answer.
+    """
+    try:
+        req = requests.get(url, timeout=timeout)
+    except requests.ConnectTimeout:
+        return None
+
+    if req.status_code != requests.codes.ok:
+        req.raise_for_status()
+
+    tags_gh = sorted(req.json(), key=lambda tag: tag['published_at'])
+    tag_gh_name = tags_gh[-1]['tag_name']
+
+    return tag_gh_name == tag_name

xija/tests/test_get_model_spec.py

@@ -0,0 +1,54 @@
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
+
+import os
+import json
+from pathlib import Path
+import re
+import pytest
+import requests
+
+from ..get_model_spec import (get_xija_model_file, get_xija_model_names,
+                              get_repo_version, check_github_version)
+
+
+def test_get_model_file_aca():
+    fn = get_xija_model_file('aca')
+    assert fn.startswith(os.environ['SKA'])
+    assert Path(fn).name == 'aca_spec.json'
+    spec = json.load(open(fn))
+    assert spec['name'] == 'aacccdpt'
+
+
+def test_get_model_file_fail():
+    with pytest.raises(ValueError, match='no models matched xxxyyyzzz'):
+        get_xija_model_file('xxxyyyzzz')
+
+    with pytest.raises(FileNotFoundError, match='xija models directory'):
+        get_xija_model_file('aca', models_path='__NOT_A_DIRECTORY__')
+
+
+def test_get_xija_model_names():
+    names = get_xija_model_names()
+    assert all(name in names for name in ('aca', 'acisfp', 'dea', 'dpa', 'pftank2t'))
+
+
+def test_get_repo_version():
+    version = get_repo_version()
+    assert isinstance(version, str)
+    assert re.match(r'^[0-9.]+$', version)
+
+
+def test_check_github_version():
+    version = get_repo_version()
+    status = check_github_version(version)
+    assert status is True
+
+    status = check_github_version('asdf')
+    assert status is False
+
+    # Force timeout
+    status = check_github_version(version, timeout=0.00001)
+    assert status is None
+
+    with pytest.raises(requests.ConnectionError):
+        check_github_version(version, 'https://______bad_url______')