It seems that the new implementation of the cluster works

Author: mesonepigreco

Modules/Cluster.py

@@ -334,9 +334,49 @@ def __setattr__(self, name, value):
 
 
 
+    def copy_file(self, source, destination, server_source = False, server_dest = True, raise_error=False, **kwargs):
+        """
+        COPY A FILE
+        ===========
+
+        This function copies a file or directory from the source to the destination.
+        The destination is on the cluster, the source is in the local machine.
+
+        It uses scp to perform the copy.
+        Alternative implementations can be performed by overloading this function.
+
+        args and kwargs are passed to the ExecuteCMD function.
+
+        The result is the output of the ExecuteCMD function.
+
+        Parameters
+        ----------
+            source : string
+                The source file to be copied
+            destination : string
+                The destination file
+            server_source : bool
+                If true, the source is on the server
+            server_dest : bool
+                If true, the destination is on the server
+            raise_error : bool
+                If True, raises an error upon failure
+        """
+        server_path = "%s:" % self.hostname
+        source_path = f"{source}"
+        dest_path = f"{destination}"
+
+        if server_source:
+            source_path = server_path + source_path 
+        if server_dest:
+            dest_path = server_path + dest_path 
+        
+        cmd = self.scpcmd + f" {source_path} {dest_path}"
+        result = self.ExecuteCMD(cmd, raise_error = raise_error, **kwargs)
+        return result
 
 
-    def ExecuteCMD(self, cmd, raise_error = False, return_output = False, on_cluster = False):
+    def ExecuteCMD(self, cmd, raise_error = False, return_output = False, on_cluster = False, use_active_shell = False):
         """
         EXECUTE THE CMD ON THE CLUSTER
         ==============================
@@ -355,6 +395,10 @@ def ExecuteCMD(self, cmd, raise_error = False, return_output = False, on_cluster
                 returned as second value.
             on_cluster : bool
                 If true, the command is executed directly on the cluster through ssh
+            use_active_shell : bool
+                If true, the command is executed in a new shell on the cluster
+                This is usefull if the command is a script that must be executed
+                in a new shell, or if the command requires .bashrc to be sourced.
 
         Returns
         -------
@@ -367,6 +411,14 @@ def ExecuteCMD(self, cmd, raise_error = False, return_output = False, on_cluster
 
         if on_cluster:
             cmd = self.sshcmd + " {} '{}'".format(self.hostname, cmd)
+            if use_active_shell:
+                cmd = "{ssh} {host} -t '{shell} --login -c \"echo {string}\"'".format(ssh = self.sshcmd, 
+                         host = self.hostname, 
+                         string = parse_symbols(string), 
+                         shell = self.terminal)
+
+
+
 
 
         success = False
@@ -630,7 +682,7 @@ def prepare_input_file(self, structures, calc, labels):
 
 Error message:
 '''.format(label)
-                MSG += str(e)
+                MSG += str(repr(e))
                 print(MSG)
 
 
@@ -710,8 +762,7 @@ def copy_files(self, list_of_input, list_of_output, to_server):
 
 
             # Clean eventually input/output file of this very same calculation
-            cmd = self.sshcmd + " %s '%s'" % (self.hostname, rm_cmd)
-            self.ExecuteCMD(cmd, False)
+            self.ExecuteCMD(rm_cmd, False, on_cluster = True)
 #            cp_res = os.system(cmd + " > /dev/null")
 #            if cp_res != 0:
 #                print "Error while executing:", cmd
@@ -720,8 +771,9 @@ def copy_files(self, list_of_input, list_of_output, to_server):
 #
 
             # Copy the file into the cluster
-            cmd = self.scpcmd + " %s %s:%s/" % (tar_file, self.hostname, self.workdir)
-            cp_res = self.ExecuteCMD(cmd, False)
+            cp_res = self.copy_file(tar_file, self.workdir, raise_error=False)
+            #cmd = self.scpcmd + " %s %s:%s/" % (tar_file, self.hostname, self.workdir)
+            #cp_res = self.ExecuteCMD(cmd, False)
             if not cp_res:
                 print ("Error while executing:", cmd)
                 print ("Return code:", cp_res)
@@ -734,8 +786,9 @@ def copy_files(self, list_of_input, list_of_output, to_server):
 
             # Unpack the input files and remove the archive
             decompress = 'cd {}; tar xf {};'.format(self.workdir, tar_name)
-            cmd = self.sshcmd + " %s '%s'" % (self.hostname, decompress)
-            cp_res = self.ExecuteCMD(cmd, False)
+            #cmd = self.sshcmd + " %s '%s'" % (self.hostname, decompress)
+            #cp_res = self.ExecuteCMD(cmd, False)
+            cp_res = self.ExecuteCMD(decompress, False, on_cluster = True)
             if not cp_res:
                 print ("Error while executing:", cmd)
                 print ("Return code:", cp_res)
@@ -751,16 +804,18 @@ def copy_files(self, list_of_input, list_of_output, to_server):
 
             compress_cmd = 'cd {}; {}'.format(self.workdir, tar_command)
 
-            cmd = self.sshcmd + " %s '%s'" % (self.hostname, compress_cmd)
-            cp_res = self.ExecuteCMD(cmd, False)
+            # cmd = self.sshcmd + " %s '%s'" % (self.hostname, compress_cmd)
+            # cp_res = self.ExecuteCMD(cmd, False)
+            cp_res = self.ExecuteCMD(compress_cmd, False, on_cluster = True)
             if not cp_res:
                 print ("Error while compressing the outputs:", cmd, list_of_output, "\nReturn code:", cp_res)
                 #return cp_res
 
 
-            # Copy the tar and unpack
-            cmd = self.scpcmd + "%s:%s %s/" % (self.hostname, os.path.join(self.workdir, tar_name), self.local_workdir)
-            cp_res = self.ExecuteCMD(cmd, False)
+            # Copy the tar from the server to the local and unpack
+            # cmd = self.scpcmd + "%s:%s %s/" % (self.hostname, os.path.join(self.workdir, tar_name), self.local_workdir)
+            # cp_res = self.ExecuteCMD(cmd, False)
+            cp_res = self.copy_file(os.path.join(self.workdir, tar_name), self.local_workdir, raise_error=False, server_source=True, server_dest=False)
             if not cp_res:
                 print ("Error while executing:", cmd)
                 print ("Return code:", cp_res)
@@ -1114,10 +1169,9 @@ def run_atoms(self, ase_calc, ase_atoms, label="ESP",
         submission += mpicmd + " " + binary + "\n"
 
         # First of all clean eventually input/output file of this very same calculation
-        cmd = self.sshcmd + " %s 'rm -f %s/%s%s %s/%s%s'" % (self.hostname,
-                                                             self.workdir, label, in_extension,
-                                                             self.workdir, label, out_extension)
-        self.ExecuteCMD(cmd, False)
+        cmd = "rm -f %s/%s%s %s/%s%s" % (self.workdir, label, in_extension,
+                                         self.workdir, label, out_extension)
+        self.ExecuteCMD(cmd, False, on_cluster = True)
 #        cp_res = os.system(cmd)
 #        if cp_res != 0:
 #            print "Error while executing:", cmd
@@ -1128,16 +1182,18 @@ def run_atoms(self, ase_calc, ase_atoms, label="ESP",
         f = open("%s/%s.sh" % (self.local_workdir, label), "w")
         f.write(submission)
         f.close()
-        cmd = self.scpcmd + " %s/%s.sh %s:%s" % (self.local_workdir, label, self.hostname, self.workdir)
-        self.ExecuteCMD(cmd, False)
+        #cmd = self.scpcmd + " %s/%s.sh %s:%s" % (self.local_workdir, label, self.hostname, self.workdir)
+        self.copy_file("%s/%s.sh" % (self.local_workdir, label), self.workdir, server_source=False, server_dest=True)
+        #self.ExecuteCMD(cmd, False)
 #        cp_res = os.system(cmd)
 #        if cp_res != 0:
 #            print "Error while executing:", cmd
 #            print "Return code:", cp_res
 #            sys.stderr.write(cmd + ": exit with code " + str(cp_res))
 #
-        cmd = self.scpcmd + " %s/%s%s %s:%s" % (self.local_workdir, label, in_extension, self.hostname, self.workdir)
-        cp_res = self.ExecuteCMD(cmd, False)
+        cp_res = self.copy_file("%s/%s%s" % (self.local_workdir, label, in_extension), self.workdir, server_source=False, server_dest=True, raise_error=False)
+        #cmd = self.scpcmd + " %s/%s%s %s:%s" % (self.local_workdir, label, in_extension, self.hostname, self.workdir)
+        #cp_res = self.ExecuteCMD(cmd, False)
         #cp_res = os.system(cmd)
         if not cp_res:
             #print "Error while executing:", cmd
@@ -1146,18 +1202,20 @@ def run_atoms(self, ase_calc, ase_atoms, label="ESP",
             return
 
         # Run the simulation
-        cmd = self.sshcmd + " %s '%s %s/%s.sh'" % (self.hostname, self.submit_command, self.workdir, label)
-        self.ExecuteCMD(cmd, False)
+        cmd = "%s %s/%s.sh" % (self.submit_command, self.workdir, label)
+        #cmd = self.sshcmd + " %s '%s %s/%s.sh'" % (self.hostname, self.submit_command, self.workdir, label)
+        self.ExecuteCMD(cmd, False, on_cluster = True)
 #        cp_res = os.system(cmd)
 #        if cp_res != 0:
 #            print "Error while executing:", cmd
 #            print "Return code:", cp_res
 #            sys.stderr.write(cmd + ": exit with code " + str(cp_res))
 
         # Get the response
-        cmd = self.scpcmd + " %s:%s/%s%s %s/" % (self.hostname, self.workdir, label, out_extension,
-                                                 self.local_workdir)
-        cp_res = self.ExecuteCMD(cmd, False)
+        #cmd = self.scpcmd + " %s:%s/%s%s %s/" % (self.hostname, self.workdir, label, out_extension,
+        #self.local_workdir)
+        #cp_res = self.ExecuteCMD(cmd, False)
+        cp_res = self.copy_file("%s/%s%s" % (self.workdir, label, out_extension), self.local_workdir, server_source=True, server_dest=False)
         #cp_res = os.system(cmd)
         if not cp_res:
             print ("Error while executing:", cmd)
@@ -1401,10 +1459,11 @@ def setup_workdir(self, verbose = True):
         """
         workdir = self.parse_string(self.workdir)
 
-        sshcmd = self.sshcmd + " %s 'mkdir -p %s'" % (self.hostname,
-                                                      workdir)
+        cmd = "mkdir -p %s" % workdir
+        # sshcmd = self.sshcmd + " %s 'mkdir -p %s'" % (self.hostname,
+        #                                               workdir)
 
-        self.ExecuteCMD(sshcmd, raise_error= True)
+        self.ExecuteCMD(cmd, raise_error= True, on_cluster=True)
 #
 #        retval = os.system(sshcmd)
 #        if retval != 0:
@@ -1441,18 +1500,17 @@ def parse_string(self, string):
 
         # Open a pipe with the server
         # Use single ' to avoid string parsing by the local terminal
-        cmd = "%s %s 'echo \"%s\"'" % (self.sshcmd, self.hostname, string)
+        #cmd = "%s %s 'echo \"%s\"'" % (self.sshcmd, self.hostname, string)
+        cmd = f"echo \"{string}\""
 
-        if self.use_active_shell_for_parsing:
-            cmd = "{ssh} {host} -t '{shell} --login -c \"echo {string}\"'".format(ssh = self.sshcmd, 
-                         host = self.hostname, 
-                         string = parse_symbols(string), 
-                         shell = self.terminal)
-        #print cmd
+        status, output = self.ExecuteCMD(cmd, return_output = True, raise_error= True, use_active_shell = self.use_active_shell_for_parsing, on_cluster = True)
+
+                #print cmd
 
         #print(cmd)
 
-        status, output = self.ExecuteCMD(cmd, return_output = True, raise_error= True)
+        #status, output = self.ExecuteCMD(cmd, return_output = True, raise_error= True)
+        
 #
 #        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
 #        output, err = p.communicate()