# -*- coding: UTF-8 -*-
""" @file erecord_vpz/models_mixins/workspace.py
..
Copyright (C) 2014-2016 INRA http://www.inra.fr \n\n
This file is part of erecord - Record platform web development \n\n
License : see LICENSE file. Authors : see AUTHORS file.
@include LICENSE @include AUTHORS
"""
""" @package erecord_vpz.models_mixins.workspace
mixins for erecord_vpz model, about workspace management
see erecord_vpz
"""
import os
import shutil
from erecord_cmn.utils.dir_and_file import delete_path_if_present
from erecord_cmn.utils.dir_and_file import clean_dir
from erecord_cmn.utils.dir_and_file import create_dir_if_absent
from erecord_cmn.utils.dir_and_file import is_a_directory
from erecord_cmn.utils.dir_and_file import recursive_overwrite
from erecord_cmn.utils.dir_and_file import get_available_pathname
from erecord_cmn.utils.vle import get_rep_pkgs_path
from erecord_cmn.utils.vle import get_pkg_output_path
from erecord_cmn.utils.vle import get_data_name
from erecord_cmn.utils.vle import get_rep_pkg_data_path
from erecord_cmn.utils.vle import get_rep_pkg_path
from erecord_cmn.utils.logger import get_logger
LOGGER = get_logger(__name__) # should be passed as a parameter of methods ?
# Some mixins for the erecord_vpz models, about workspace management
[docs]class WorkspaceMixin(object):
[docs] @classmethod
def get_undefined_value(cls):
return "undefined"
[docs] @classmethod
def is_undefined_value(cls, v):
return (v == cls.get_undefined_value())
[docs] def get_experiment_file_path(self):
return os.path.join(self.homepath, "user_experiment.xls")
[docs]class VlehomeWorkspaceMixin(WorkspaceMixin):
[docs] def init_vlehome(self, LOGGER, as_vlehome=False):
vlehome = os.path.join(self.homepath, "vle_home")
if as_vlehome :
(deletion_done, creation_done) = clean_dir(dirpath=vlehome)
if deletion_done :
LOGGER.warning(u"%s has been deleted before to be recreated",
vlehome)
return vlehome
[docs] def get_vlehome(self):
return self.vlehome
[docs] def clean_vlehome(self):
clean_dir(dirpath=self.vlehome)
[docs] def clear_vlehome(self):
delete_path_if_present(path=self.vlehome)
[docs] def build_pkg_and_its_dependencies(self, vpzname, vlepackage_list,
vle_version):
"""Builds into vlehome a vle package and its dependency ones.
The input vle packages are given as a list of their \
(package name, package path), where the first one is the main package \
(containing the vpz file named vpzname).
The main vle package is built as a copy, and the dependency packages \
as links.
About the main package : its whole 'exp' subdirectory is copied, even \
if only its vpz file vpzname is needed.
About the main package : 'output' subdirectory is created if not there.
Go to 'output' subdirectory.
"""
pkgs_path = get_rep_pkgs_path(rep_path=self.vlehome,
vle_version=vle_version)
try:
(pkgname,vlepkgpath) = vlepackage_list[0] # main package
src = vlepkgpath
dest = os.path.join(pkgs_path,pkgname)
shutil.copytree(src, dest)
output_path = get_pkg_output_path(pkg_path=dest)
create_dir_if_absent(dirpath=output_path)
#for (pkgname,vlepkgpath) in vlepackage_list[1:-1] : # dependencies
#modified because notok with debian (the last pkg was forgotten)
for i,(pkgname,vlepkgpath) in enumerate(vlepackage_list) :
if i > 0 : # dependencies
src = vlepkgpath
dest = os.path.join(pkgs_path,pkgname)
if pkgname == "erecord" : # particular case
shutil.copytree(src, dest)
output_path = get_pkg_output_path(pkg_path=dest)
create_dir_if_absent(dirpath=output_path)
else :
os.symlink(src, dest)
os.chdir(output_path)
except OSError as e:
msg = "unable to build vle packages into vle_home workspace "+ \
self.vlehome
raise Exception(msg)
[docs]class ReporthomeWorkspaceMixin(WorkspaceMixin):
[docs] def init_reporthome(self, LOGGER, as_reporthome=False):
reporthome = os.path.join(self.homepath, "report_home")
if as_reporthome :
(deletion_done, creation_done) = clean_dir(dirpath=reporthome)
if deletion_done :
LOGGER.warning(u"%s has been deleted before to be recreated",
reporthome)
return reporthome
[docs] def get_reporthome(self):
return self.reporthome
[docs] def get_report_folder_path(self, name) :
"""Defines and return a folder path name as a reporthome subdirectory"""
return os.path.join(self.reporthome, name+'_folder')
[docs] def clean_reporthome(self):
clean_dir(dirpath=self.reporthome)
[docs] def clear_reporthome(self):
delete_path_if_present(path=self.reporthome)
[docs]class DatahomeWorkspaceMixin(WorkspaceMixin):
""" datahome workspace
The datahome workspace is dedicated to modify the 'data' directory of
the main vle package according to a data folder sent into the request.
Implied folders and files :
datahome = homepath / data_home
homepath / vle_home
pkg_data_path = homepath / vle_home / ... / pkgname / data
pkg_data_svg_path = homepath / vle_home / ... / pkgname / data_original
datafolder_zip_file = homepath / data_home / datafolder.zip
datafolder_src = homepath / data_home / data
pkg_data_path, pkg_data_svg_path : the 'data' directory of the main vle
package is pkg_data_path. In some case it will be saved before
modification, into pkg_data_svg_path.
datafolder_zip_file, datafolder_src : the datafolder_zip_file zip file
contains the data folder sent into a request. It will be unzipped to get
datafolder_src. The pkg_data_path content will be replaced or overwritten
by the datafolder_src content.
"""
[docs] def init_datahome(self, LOGGER, as_datahome=False):
datahome = os.path.join(self.homepath, "data_home")
if as_datahome :
(deletion_done, creation_done) = clean_dir(dirpath=datahome)
if deletion_done :
LOGGER.warning(u"%s has been deleted before to be recreated",
datahome)
return datahome
[docs] def get_datafolder_zip_file_path(self):
return os.path.join(self.datahome, "datafolder.zip")
[docs] def get_datafolder_src_path(self):
return os.path.join(self.datahome, get_data_name()) # datahome / data
[docs] def get_and_verify_datafolder_src_path(self):
datafolder_src_path = self.get_datafolder_src_path()
data_name = os.path.basename(datafolder_src_path)
cr_ok = is_a_directory(datafolder_src_path)
return (cr_ok, datafolder_src_path, data_name)
[docs] def get_pkg_data_svg_path(self, vle_version, pkgname):
pkg_path = get_rep_pkg_path(rep_path=self.vlehome,
vle_version=vle_version, pkgname=pkgname)
pkg_data_svg_path = os.path.join(pkg_path, "data_original")
return pkg_data_svg_path
[docs] def get_datahome(self):
return self.datahome
[docs] def clean_datahome(self):
"""creates datahome after having deleted it before if necessary"""
(deletion_done, creation_done) = clean_dir(dirpath=self.datahome)
if deletion_done :
LOGGER.warning(u"%s has been deleted before to be recreated",
self.datahome)
[docs] def clear_datahome(self):
delete_path_if_present(path=self.datahome)
[docs] def replace_pkg_data(self, data_src, vle_version, pkgname):
"""replaces 'data' of the 'pkgname' vle package by data_src.
Replaces the content of the 'data' directory of the 'pkgname' vle
package by the content of the data_src folder, after having saved
the original 'data' directory as pkg_data_svg_path
(see get_pkg_data_svg_path).
The vle package is located under vlehome.
"""
pkg_data_path = get_rep_pkg_data_path(rep_path=self.vlehome,
vle_version=vle_version,
pkgname=pkgname)
pkg_data_svg_path = self.get_pkg_data_svg_path(vle_version=vle_version,
pkgname=pkgname)
try:
shutil.move(pkg_data_path, pkg_data_svg_path)
shutil.move(data_src, pkg_data_path)
except:
#msg = "unable to replace the 'data' folder of the " +pkgname+ " vle package (into vle_home workspace " +self.vlehome+ " ) by the " +data_src+ "folder"
raise
[docs] def overwrite_pkg_data(self, data_src, vle_version, pkgname):
"""Completes 'data' of the 'pkgname' vle package with data_src.
Completes (by overwriting) the content of the 'data' directory of
the 'pkgname' vle package with the content of the data_src folder.
The vle package is located under vlehome.
"""
pkg_data_path = get_rep_pkg_data_path(rep_path=self.vlehome,
vle_version=vle_version,
pkgname=pkgname)
try:
recursive_overwrite(data_src, pkg_data_path)
except:
raise
[docs]class RunhomeWorkspaceMixin(WorkspaceMixin):
[docs] def init_runhome(self, LOGGER, as_runhome=False):
runhome = os.path.join(self.homepath, "run_home")
if as_runhome :
(deletion_done, creation_done) = clean_dir(dirpath=runhome)
if deletion_done :
LOGGER.warning(u"%s has been deleted before to be recreated",
runhome)
return runhome
[docs] def get_runhome(self):
return self.runhome
[docs] def clean_runhome(self):
clean_dir(dirpath=self.runhome)
[docs] def clear_runhome(self):
delete_path_if_present(path=self.runhome)
[docs] def define_and_build_runhome_subdirectory(self, rootname="s_"):
runhome_subdirectory = get_available_pathname(rootpath=self.runhome,
rootname=rootname)
create_dir_if_absent(runhome_subdirectory)
return runhome_subdirectory