Skip to content
Snippets Groups Projects
Select Git revision
  • af99ed699fcf2ca030dd2c6af4e1a0bd621a2422
  • master default
  • bing_issues#190_tf2
  • bing_tf2_convert
  • bing_issue#189_train_modular
  • simon_#172_integrate_weatherbench
  • develop
  • bing_issue#188_restructure_ambs
  • yan_issue#100_extract_prcp_data
  • bing_issue#170_data_preprocess_training_tf1
  • Gong2022_temperature_forecasts
  • bing_issue#186_clean_GMD1_tag
  • yan_issue#179_integrate_GZAWS_data_onfly
  • bing_issue#178_runscript_bug_postprocess
  • michael_issue#187_bugfix_setup_runscript_template
  • bing_issue#180_bugs_postprpocess_meta_postprocess
  • yan_issue#177_repo_for_CLGAN_gmd
  • bing_issue#176_integrate_weather_bench
  • michael_issue#181_eval_era5_forecasts
  • michael_issue#182_eval_subdomain
  • michael_issue#119_warmup_Horovod
  • bing_issue#160_test_zam347
  • ambs_v1
  • ambs_gmd_nowcasting_v1.0
  • GMD1
  • modular_booster_20210203
  • new_structure_20201004_v1.0
  • old_structure_20200930
28 results

kth_dataset.py

Blame
  • JsonFileStorageAdapter.py 4.18 KiB
    import os
    import json
    import uuid
    
    from .LocationStorage import AbstractLocationDataStorageAdapter, LocationData, LocationDataType
    
    from typing import List
    
    DEFAULT_JSON_FILEPATH: str = "./app/data"
    
    class StoredData:
        actualData: LocationData
        users: List[str]
    
    
    # This stores LocationData via the StoredData Object as json files
    # These Jsonfiles then contain the actualData, as well as the users with permissions for this LocationData
    # all users have full permission to to anything with this dataobject, uncluding removing their own access (this might trigger a confirmation via the frontend, but this is not enforced via the api)
    # IMPORTANT: The adapter does not check for authentication or authorization, it should only be invoked if the permissions have been checked
    class JsonFileStorageAdapter(AbstractLocationDataStorageAdapter):
        data_dir: str
    
        def __init__(self, data_directory: str = DEFAULT_JSON_FILEPATH):
            AbstractLocationDataStorageAdapter.__init__(self)
            self.data_dir = data_directory
            if not (os.path.exists(self.data_dir) and os.path.isdir(self.data_dir)):
                raise Exception('Data Directory \"' + self.data_dir + '\" does not exist.')
    
        def getList(self, type: LocationDataType):
            localpath = os.path.join(self.data_dir, type.value)
            if not (os.path.isdir(localpath)):
                # This type has apparently not yet been used at all, create its directory and return an empty json file
                os.mkdir(localpath)
                return {}
            else:
                allFiles = [f for f in os.listdir(localpath) if os.path.isfile(os.path.join(localpath, f))]
                # now each file has to be checked for its filename (= id) and the LocationData name (which is inside the json)
                retList = []
                for f in allFiles:
                    with open(os.path.join(localpath, f)) as file:
                        data = json.load(file)
                        retList.append({data['actualData']['name'] : f})
                return retList
    
        def addNew(self, type: LocationDataType, data: LocationData, usr: str):
            localpath = os.path.join(self.data_dir, type.value)
            if not (os.path.isdir(localpath)):
                # This type has apparently not yet been used at all, therefore we need to create its directory
                os.mkdir(localpath)
            # create a unique id, by randomly generating one, and re-choosing if it is already taken
            id = str(uuid.uuid4())
            while (os.path.exists(os.path.join(localpath, id))):
                id = str(uuid.uuid4())
            toStore = StoredData()
            toStore.users = [usr]
            toStore.actualData = data
            with open(os.path.join(localpath, id), 'w') as json_file:
                json.dump(toStore.__dict__, json_file)
            return {id : data}
    
        def getDetails(self, type: LocationDataType, id: str):
            localpath = os.path.join(self.data_dir, type.value)
            fullpath = os.path.join(localpath, id)
            if not os.path.isfile(fullpath):
                raise FileNotFoundError('The requested Object does not exist.')
            with open(fullpath) as file:
                data = json.load(file)
            return data['actualData']
    
        def updateDetails(self, type:LocationDataType, id:str, data: LocationData, usr: str):
            localpath = os.path.join(self.data_dir, type.value)
            fullpath = os.path.join(localpath, id)
            if not os.path.isfile(fullpath):
                raise FileNotFoundError('The requested Object does not exist.')
            
            toStore = StoredData()
    
            # get permissions from old file
            with open(fullpath) as file:
                data = json.load(file)
                toStore.users = data['users']
    
            toStore.actualData = data
            with open(fullpath, 'w') as file:
                json.dump(data.__dict__, file)
            return {id : data}
    
        def getOwner(self, type: LocationDataType, id: str):
            raise NotImplementedError()
    
        def checkPerm(self, type: LocationDataType, id: str, usr: str):
            raise NotImplementedError()
    
        def addPerm(self, type: LocationDataType, id: str, authUsr: str, newUser: str):
            raise NotImplementedError()
    
        def rmPerm(self, type: LocationDataType, id: str, usr: str, rmUser: str):
            raise NotImplementedError()