Source code for lfd.analysis.utils

"""A collection of various miscelaneous functionality that helps visualize the
profiles and other utilities.
"""
import os
import os.path as ospath
import glob
import time
import warnings

import numpy as np
import matplotlib.pyplot as plt

from lfd.analysis.profiles.convolution import convolve
from lfd.analysis.profiles.objectprofiles import *
from lfd.analysis.profiles.seeing import *
from lfd.analysis.profiles.consts import *


__all__ = ["PERSISTENT_CACHE", "EPHEMERAL_CACHE",
           "search_cached", "cache_data", "get_data"]


def get_cache_dirs(path=None, name="data"):
    """Returns the profiles module path. If name is given, joins the module
    path and the name.

    Parameters
    ----------
    name : `str`, optional
        If provided, name will be joined to the module path.

    Returns
    -------
    cacheList : `list`
        List of paths to all cache directories.
    """
    path = ospath.dirname(__file__) if path is None else path
    cacheList = []
    exclude_prefixes = ('__', '.')
    for dirpath, dirnames, filenames in os.walk(path):
        dirnames[:] = [dirname for dirname in dirnames
                       if not dirname.startswith(exclude_prefixes)]
        for dirname in dirnames:
            if dirname == name:
                cacheList.append(ospath.join(dirpath, dirname))

    return cacheList

PERSISTENT_CACHE = get_cache_dirs(name="data")
"""Cache for data that ships with lfd. Contains Rabina profiles as well as data
that recreates plots from published papers. This cache is not meant to be
written to be the user, for that see EPHEMERAL_CACHE."""

EPHEMERAL_CACHE = ospath.expanduser("~/.lfd")
"""Ephemeral cache is used for data created by the user. If it doesn't exist
one will be created at import."""
if not ospath.exists(EPHEMERAL_CACHE):
    os.makedirs(EPHEMERAL_CACHE)


[docs]def search_cached(name, ephemeral=True, persistent=True): """Returns a list of files matching the given name. By default both caches are searched. If no matches are found an error is raised. Parameters ---------- names: `str`` Name of the desired data file. ephemeral: `bool`, optional Search ephemeral cache. True by default. persistent: `bool`, optional Search persistent cache. True by default. Returns ------- filePaths : `list` List of paths to all cached files that matched the given name pattern. Raises ------ FileNotFoundError: No cached files match given name. """ ephemFiles, persistFiles = [], [] if ephemeral: ephemFiles = glob.glob(ospath.join(EPHEMERAL_CACHE, '*'+name+'*')) if persistent: persistFiles = [] for pcache in PERSISTENT_CACHE: persistFiles.extend(glob.glob(ospath.join(pcache, '*'+name+'*'))) filePaths = ephemFiles + persistFiles if len(filePaths) != 0: return filePaths else: raise FileNotFoundError(f"File {name} does not exist in any cache.")
def get_rabina_profile(angle, useCV2=False): """Returns image of projected Rabina profile as a function of angle between the observers line of sight and the velocity vector of the meteor. Parameters ---------- angle: `float` angle, in radians, between LOS and vector of the meteor. useCV2: `bool`, optional If True will use cv2 to open the Rabina profile, otherwise matplotlib. Default is False. Returns ------- profile : `np.array` Projected image of a Rabina prfile as numpy array. Notes ----- Pre-rendered Rabina profiles exist for angles in 0.1 increments from 0 to 1.5 inclusive. This is approximately every 6 degrees from 0 to 90 degrees. See `gen_rabina_profile.py` in persistent cache on how to pre-render Rabina profiles for other angles. Profiles are searched for in both ephemeral and persistent cache locations. If multiple matching files are found, those from peristent cache only are returned. Raises ------ ValueError: when the given angle was not in the range 0-1.5 (step 0.1) inclusive. FileNotFoundError: when the searched for Rabina profile was not found. This can happen only when a 3rd party has manually cleared the persistent cache. """ rounded = int(angle*10) if rounded not in range(0, 16, 1): raise ValueError("No premade Rabina profiles exist outside of the 0 to 1.5 radians!") rabinaPaths = search_cached(f"lfd_rabina{rounded}_r.png") if len(rabinaPaths) > 1: rabinaPaths = search_cached(f"lfd_rabina{rounded}_r.png", ephemeral=False) if useCV2: import cv2 return cv2.imread(rabinaPaths[0], cv2.IMREAD_GRAYSCALE) return plt.imread(rabinaPaths[0])
[docs]def cache_data(data, name): """Caches given array-like data under the given name. Utilizes numpy.save function to store the array like data. Use get_data to retrieve the saved array. Not thread safe. Parameters ---------- data : `np.array` Array like object to save to cache. The EPHEMERAL_CACHE variable stores the location of the cache, by default at ~/.lfd/. name : `str`, `list` or `tuple` Name under which the data will be cached. If the name conflicts the current date will pre pre-pended to the name. """ if isinstance(name, str): filename = (name, ) data = (data, ) for fname, data in zip(filename, data): try: paths = search_cached(fname) except FileNotFoundError: # no files with that name exist in either cache, all is well filepath = os.path.join(EPHEMERAL_CACHE, fname) else: # file with that name exists, rename and warn newname = time.strftime("%Y%m%d-%H%M%S") + fname warnings.warn(f"Filename {fname} exists, saving as {newname}") filepath = os.path.join(EPHEMERAL_CACHE, fname) np.save(filepath, data, allow_pickle=False) print(f" File cached as {os.path.basename(filepath)}.")
[docs]def get_data(fname): """If fname is an existing file tries to load the data in it. If fname is a file name searches the cache for matches. If the match is unique, loads it. Parameters ---------- fname : `str` File path or file name of the file to load. Does not support loading. Returns ------- data : `np.array` A numpy array with the requested data. Raises ------ FileNotFoundError: When fname could not be found. """ if os.path.isfile(fname): return np.load(fname, allow_pickle=False) filePaths = search_cached(fname) if len(filePaths) == 1: return np.load(filePaths[0], allow_pickle=False) elif len(filePaths) > 1: errmsg = "Multiple possible matches found. Which file were you looking for: \n" for fp in filePaths: errmsg += " * {os.path.basename(fp)} \n" raise FileNotFoundError(errmsg)