dataManager_michel.py

#!/usr/bin/env python

import os
import subprocess
import socket
import psutil
import json
import math
import numpy as np
import yt
from spectral_cube import SpectralCube
from astropy.io import fits
from astropy.stats import sigma_clip
from datetime import datetime

from DataBlock import DataBlock

DataRoot = "dataroot"
FITSFilePrefix = DataRoot + '/FITS/'
PNGFilePrefix = DataRoot + '/PNG/'
OBJFilePrefix = DataRoot + '/OBJ/'
ApiDocPrefix = DataRoot + '/apidoc/'

#
#
# Tell DataBlock where the files will be located
#
# The FITS files
DataBlock.setFITSFilePrefix(FITSFilePrefix)

# The PNG files
DataBlock.setPNGFilePrefix(PNGFilePrefix)

# The PNG files
DataBlock.setOBJFilePrefix(OBJFilePrefix)

#
# An implementation of the class used in the serverWsgi-ish server.
#
class DataManagerImpl :
    __default_palette_name, __default_transformation_name, __default_video_mode_name = DataBlock.getDefaults()

    #===========================================================================
    # Private methods
    #

    # check if the FITS data identified by the relFITSFilePath are present in
    # memory. If yes update the corresponding DataBlock's timestamp.

    def __checkPresence(self, relFITSFilePath):
        self.__logger.debug("__checkPresence : entering")
        result = {"status": True, "message" : '', "result": None}
        if relFITSFilePath in self.__dataBlocks:
            self.__dataBlocks[relFITSFilePath].setLastAccess()
        else:
            result = {"status": False, "message": f'FITS file "{relFITSFilePath}" is not present in memory. Call "setData" first'}
        self.__logger.debug("__checkPresence : exiting")
        return result

    def __getEntries_0(self, relKey):
        self.__logger.debug("__getEntries_0: entering")
        try :
            absFITSFilePrefix = FITSFilePrefix + relKey
            entries = (os.listdir(absFITSFilePrefix))
            self.__logger.debug("type of entries %s" % (type(entries)))
            sortedEntries = entries.sort()
            children = []
            for entry in entries:
                p = absFITSFilePrefix + '/' + entry
                condition = os.path.isfile(p) and p.endswith(".fits")
                condition  = condition or (os.path.islink(p) and os.path.isdir(os.readlink(p)))
                condition = condition or (os.path.isdir(p))
                if not condition :
                    continue
                elif entry in ["log", "NOFITS", "IGNORE"]:
                    continue
                else:
                    d = dict()
                    d["key"]    = relKey + '/' + entry
                    d["folder"] = not os.path.isfile(p) 
                    d["lazy"]   = d["folder"]
                    if entry.endswith(".fits") :
                        size = DataBlock.convert_size(os.path.getsize(p))
                        d["title"] = "<a href = 'visit/?relFITSFilePath=%s/%s' target = '_blank'>%s %s</a>" % (relKey, entry, entry, size)
                    else:
                        d["title"] = entry
                    children.append(d)
            result = {"status": True, "message": "", "result": json.dumps(children)}
        except Exception as e :
            result = {"status": False, "message": "Problem while looking for entries under '%s'. Error message was '%s'" % ( relKey, e)}
            self.__logger.debug("%r" % result)
        self.__logger.debug("__getEntries_0: exiting")
        return result

    #
    # Return the keys of the dataBlocks dictionary as if it was ordered by timestamp ( asc or desc)
    #
    def __getKeysOrderedByTimestamp (self, reverse=True):
        return [x[0] for x in sorted({x: self.__dataBlocks[x]["timestamp"] for x in self.__dataBlocks.keys()}.items(), key=lambda kv: kv[1], reverse=reverse)]

    #===========================================================================
    # Public methods.

    #===========================================================================
    # CTOR
    #
    def __init__(self, logger):

        self.__dataBlocks = dict()
        self.__logger = logger
        self.__logger.debug(f"An instance of DataManagerImpl is created")

    #===========================================================================
    # THE data selector.
    # Its execution will trigger the creation of a DataBlock populated by
    # the content of a FITS file if it's not already loaded.
    # In any case return the header of the FITS data as a dictionary.
    #
    def setData(self, relFITSFilePath):
        self.__logger.debug("setData : entering");
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            self.__dataBlocks[relFITSFilePath].setLastAccess()
            result = self.__dataBlocks[relFITSFilePath].getHeader()
        else :
            self.purgeDataBlocks()
            db = DataBlock(self.__logger)
            result = db.setData(relFITSFilePath)
            if (result["status"]):
                self.__dataBlocks[relFITSFilePath]=db
        self.__logger.debug(f"About to return {result}")
        self.__logger.debug("setData : Exiting")
        return result

    #===========================================================================
    # General getters
    #
    #


    #===========================================================================
    # Getters on a DataBlock identified by its relFITSFilePath ( i.e. the dictionary keys in __dataBlocks )
    #
    def getDimensions(self, relFITSFilePath):
        self.__logger.debug( "getDimensions : entering");
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].getDimensions()
        self.__logger.debug( "getDimensions : exiting");
        return result

    def getSlice(self, relFITSFilePath, iFREQ, step=1 ):
        self.__logger.debug("getSlice : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].getSlice(iFREQ, step)
        self.__logger.debug("getSlice : exiting")
        return result

    def getSpectrum(self, relFITSFilePath, iRA=None, iDEC=None, iFREQ0=None, iFREQ1=None):
        self.__logger.debug( "getSpectrum : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].getSpectrum(iRA, iDEC, iFREQ0, iFREQ1)
        self.__logger.debug( "getSpectrum : exiting")
        return result

    def getAverageSpectrum(self, relFITSFilePath, iDEC0=None, iDEC1=None, iRA0=None, iRA1=None, retFITS=False):
        self.__logger.debug("getAverageSpectrum : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].getAverageSpectrum(iDEC0, iDEC1, iRA0, iRA1, retFITS)
        self.__logger.debug("getAverageSpectrum : exiting")
        return result


    def getHeader(self, relFITSFilePath):
        self.__logger.debug("getHeader : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].getHeader()
        self.__logger.debug("getHeader : exiting")
        return result

    def RADECRangeInDegrees( self, relFITSFilePath):
        self.__logger.debug("RADECRangeInDegrees : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].RADECRangeInDegrees()
        self.__logger.debug("RADECRangeInDegrees : exiting")
        return result

    def degToHMSDMS(self, relFITSFilePath, RAinDD, DECinDD):
        self.__logger.debug("degToHMSDMS : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].degToHMSDMS(RAinDD, DECinDD)
        self.__logger.debug("degToHMSDMS : exiting")
        return result

    def rangeToHMS(self,  relFITSFilePath, iRA0, iRA1, iRAstep):
        self.__logger.debug("degToHMSDMS : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result=self.__dataBlocks[relFITSFilePath].rangeToHMS(iRA0, iRA1, iRAstep)
        self.__logger.debug("degToHMSDMS : exiting")
        return result

    def rangeToDMS(self,  relFITSFilePath, iDEC0, iDEC1, iDECstep):
        self.__logger.debug("rangeToDMS : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].rangeToDMS(iDEC0, iDEC1, iDECstep)
        self.__logger.debug("rangeToDMS : exiting")
        return result

    def getPixelValueAtiFreqiRAiDEC(self, relFITSFilePath, iFreq, iRA, iDEC):
        self.__logger.debug("getPixelValueAtiFreqiRAiDEC : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result=self.__dataBlocks[relFITSFilePath].getPixelValueAtiFreqiRAiDEC(iFreq, iRA, iDEC)
        self.__logger.debug("getPixelValueAtiFreqiRAiDEC : exiting")
        return result

    def getSumOverSliceRectArea(self, relFITSFilePath, iFREQ, iRA0=None, iRA1=None, iDEC0=None, iDEC1=None):
        self.__logger.debug("getSumOnSliceRectArea : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result=self.__dataBlocks[relFITSFilePath].getSumOverSliceRectArea(iFREQ, iRA0, iRA1, iDEC0, iDEC1)
        self.__logger.debug("getSumOnSliceRectArea : exiting")
        return result

    def getAverage(self, relFITSFilePath, iFREQ0, iFREQ1, iDEC0, iDEC1, iRA0, iRA1, retFITS):
        self.__logger.debug("getAverage : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].getAverage(iFREQ0, iFREQ1, iDEC0, iDEC1, iRA0, iRA1, retFITS)
        self.__logger.debug("getAverage : exiting")
        return result

    def getOneSliceAsPNG (self, iFREQ, relFITSFilePath, **kwargs):
        self.__logger.debug("getOneSliceAsPNG : entering.")
        self.__logger.debug("iFREQ = %r, relFITSFilePath = %r" % (iFREQ, relFITSFilePath))
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].getOneSliceAsPNG(iFREQ, **kwargs)
        self.__logger.debug("getOneSliceAsPNG : exiting.")
        return result

    def getSummedSliceRangeAsPNG( self, relFITSFilePath, iFREQ0, iFREQ1, **kwargs):
        self.__logger.debug("getSummedSliceRangeAsPNG : entering.")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"]:
            result = self.__dataBlocks[relFITSFilePath].getSummedSliceRangeAsPNG(iFREQ0, iFREQ1, **kwargs)
        self.__logger.debug("getSummedSliceRangeAsPNG : exiting.")
        return result

    def getContours(self, relFITSFilePath, iFREQ, iDEC0, iDEC1, iRA0, iRA1, **kwargs):
        self.__logger.debug("getContours : entering.")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"] :
            result = self.__dataBlocks[relFITSFilePath].getContours(iFREQ, iDEC0, iDEC1, iRA0, iRA1, **kwargs)
        self.__logger.debug("getContours : exiting.")
        return result

    def measureContour(self, relFITSFilePath, iFREQ, contour, level):
        self.__logger.debug("measureContour - dispatcher : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"] :
            result = self.__dataBlocks[relFITSFilePath].measureContour(iFREQ, contour, level)
        self.__logger.debug("measureContour - dispatcher : exiting")
        return result

    def measureBox(self, relFITSFilePath, iFREQ, iRA0, iRA1, iDEC0, iDEC1):
        self.__logger.debug("measureBox - dispatcher : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result["status"] :
            result = self.__dataBlocks[relFITSFilePath].measureBox(iFREQ, iRA0, iRA1, iDEC0, iDEC1)
        self.__logger.debug("measureBox - dispatcher : exiting")
        return result


    #
    # create fits file containing a spectrum at iRA, iDEC
    # The use case is interoperability via SAMP
    #
    def createFits(self, relFITSFilePath, iRA, iDEC):
        self.__logger.debug("createFITS : entering")
        result = self.__checkPresence(relFITSFilePath)
        if result :
            result = self.__dataBlocks[relFITSFilePath].createFits(iRA, iDEC)
        self.__logger.debug("createFITS : exiting")
        return result

    #
    # Returns all the entries directly located under a given subdirectory.
    # The subdirectory must be passed as a full path implicitely
    # rooted at FITSFilePrefix.
    # i.e. :
    # "/" will be interpreted as FITSFilePrefix+"/"
    # "/2018" will be interpreted as FITSFilePrefix+"/2018"
    # etc...
    # The result is a JSON array of dictionaries , on the basis of one dictionary per inode:
    # [{'title': filename of the entry, 'key': the FITSFilePrefix rooted path of the entry, 'folder' : true (resp. false) if the inode is (resp. is not) a subdir, 'lazy': true}]
    def getEntries(self, relKey):
        self.__logger.debug("getEntries: entering")
        return self.__getEntries_0(relKey)
        self.__logger.debug("getEntries: exiting")

    #
    # Returns the capabilities of the server relative to the way the image pixels are calculated.
    #
    #  Intensity Transformations names are defined in a list itts
    #  Colors names are defined in a list luts
    #  Video modes are defined in a list vmode
    #
    #  For each of this rendering filters default names are defined.
    #
    def renderingCapabilities(self) :
        self.__logger.debug("renderingCapabilities: entering")
        result = {"status": True, "message": "", "result": DataBlock.getRenderingCapabilities()}
        self.__logger.debug("renderingCapabilities: exiting")
        return result

    #
    # Return the list of defined DataBlock s sorted ( asc or desc ) by timestamp
    #
    def getDataBlockInfos(self, reverse=True):
        self.__logger.debug("getDataBlockInfos : entering")
        result = dict()
        #
        # report time
        #
        result["when"] = datetime.now().strftime(DataBlock.getDateTimeFormat())

        # Machine name
        #
        hostname = socket.gethostname()
        result["hostname"] = hostname

        #
        # Machine informations
        #
        numsockets =  int(subprocess.check_output('cat /proc/cpuinfo | grep "physical id" | sort -u | wc -l', shell=True))
        cpucount = psutil.cpu_count()
        cpuInfos = {"numsockets": numsockets, "cpucount": cpucount}
        result["cpu"] = cpuInfos

        #
        # Physical memory informations
        #
        physMemoryInBytes = os.sysconf('SC_PAGE_SIZE') * os.sysconf('SC_PHYS_PAGES')
        usedMemoryInBytes = sum([self.__dataBlocks[dbId].sizeInBytes for dbId in self.__dataBlocks])
        usedMemoryPercentage = math.floor((usedMemoryInBytes / physMemoryInBytes) * 10000) / 100.
        physMemory = DataBlock.convert_size(physMemoryInBytes)
        usedMemory = DataBlock.convert_size(usedMemoryInBytes)
        memoryInfos = {"physMemory": physMemory, "usedMemory": usedMemory, "usedMemoryPercentage": usedMemoryPercentage}

        result["memory"] = memoryInfos

        # Maximum idleness duration
        result["maxidle"] = DataBlock.getMaxIdle()

        #
        # DataBlocks informations
        #
        infoNames = DataBlock.getDataBlockInfoNames()
        x = [infoNames]
        now = datetime.now()
        for k in self.__dataBlocks:
            x.append(self.__dataBlocks[k].getDataBlockInfos(now, infoNames))
        self.__logger.debug("getDataBlockInfos : exiting")

        result["dataBlocks"] = x

        self.__logger.debug("getDataBlockInfos : exiting")
        return {"status": True, "message": "", "result": result}

    #
    # Delete the DataBlocks having their "idle" property
    # greater than DataBlock.getMaxIdle()
    #
    def purgeDataBlocks(self):
        self.__logger.debug("purgeDataBlocks : entering")
        tobePurged = [db for db in self.__dataBlocks if (datetime.now() - self.__dataBlocks[db].lastAccess).total_seconds() > DataBlock.getMaxIdle()]
        for db in tobePurged:
            del self.__dataBlocks[db]
        self.__logger.debug("purgeDataBlocks : exiting")
        return {"status": True, "message": f"{len(tobePurged)} data set(s) erased.", "result": tobePurged}

    #
    #
    #
    def getYtObj(self,relFITSFilePath,product,coord):
        self.__logger.debug("getYtObj : entering")
        self.__logger.debug("relFITSFilePath = %r, product= %r, (%i,%i,%i,%i,%i,%i)" % (relFITSFilePath,product,coord['iRA0'],coord['iRA1'],coord['iDEC0'],coord['iDEC1'],coord['iFREQ0'],coord['iFREQ1']))
        result = self.__checkPresence(relFITSFilePath)
        if result :
            result = self.__dataBlocks[relFITSFilePath].getYtObj(relFITSFilePath,product,coord)
        self.__logger.debug("getYtObj : exiting")
        return result

#
#
#    End of the DataManagerImpl class
#
#