изменения в comsdk

comsdk/communication.py

 import os
 import os.path
 import shutil
-import paramiko
 import subprocess
 import shlex
 import json
@@ -10,6 +9,8 @@ from stat import S_ISDIR
 from abc import ABCMeta, abstractmethod
 import logging
 
+import paramiko
+
 import comsdk.comaux as aux
 
 
@@ -30,12 +31,21 @@ class Host(object):
         self.commands[cmd_name] = cmd
 
     def get_program_launch_path(self, prog_name):
+        if prog_name not in self.programs:
+            raise ValueError(f'Program "{prog_name}" is not recognized. '
+                             'Please add this program to "custom_programs" '
+                             'in the corresponding host in the config file '
+                             'if you want to use it.')
         path_to_prog = self.programs[prog_name]
         if path_to_prog is not None:
-            return self.programs[prog_name] + '/' + prog_name
+            print(self.programs[prog_name], prog_name)
+            return self.join_path(self.programs[prog_name], prog_name)
         else:
             return prog_name
 
+    def join_path(self, *path_list):
+        return os.path.join(*path_list)
+
 
 class RemoteHost(Host):
     '''
@@ -69,6 +79,10 @@ class RemoteHost(Host):
         self.set_job_id = aux.load_function_from_module(state['job_setter'])
         self.check_task_finished = aux.load_function_from_module(state['job_finished_checker'])
 
+    def join_path(self, *path_list):
+        # For RemoteHost, we assume that it is posix-based
+        return '/'.join(path_list)
+
 
 # Decorator
 def enable_sftp(func):
@@ -152,12 +166,20 @@ class LocalCommunication(BaseCommunication):
         return LocalCommunication(local_host)
 
     def execute(self, command, working_dir=None):
-        command_line = command if working_dir is None else 'cd {}; {}'.format(working_dir, command)
-        #print('\t' + command_line)
-        # use PIPEs to avoid breaking the child process when the parent process finishes
-        # (works on Linux, solution for Windows is to add creationflags=0x00000010 instead of stdout, stderr, stdin)
+        if working_dir is None:
+            command_line = command 
+        else:
+            if os.name == 'posix':
+                command_line = 'cd {}; {}'.format(working_dir, command)
+            elif os.name == 'nt':
+                command_line = ''
+                if working_dir[0] != 'C':
+                    command_line += f'{working_dir[0]}: && '
+                command_line += 'cd {} && {}'.format(working_dir, command)
         #self._print_exec_msg(command_line, is_remote=False)
-        subprocess.call([command_line], shell=True)
+        #res = subprocess.call([command_line], shell=True)
+    #    print(command_line)
+        res = subprocess.run(command_line, shell=True)
         return [], []
 
     def copy(self, from_, to_, mode='from_local', show_msg=False):

comsdk/edge.py

@@ -53,7 +53,8 @@ class Edge:
         'postprocess',
         'order',
         'comment',
-        'mandatory_keys'
+        'mandatory_keys',
+        'use_proxy_data'
     ]
 
     def __init__(self, predicate, morphism, 
@@ -70,6 +71,7 @@ class Edge:
         self.order = int(0 if order is None else order)
         self.comment = comment
         self.mandatory_keys = mandatory_keys
+        self.use_proxy_data=False
 
     def predicate(self, data, dynamic_keys_mapping={}):
         proxy_data = self._io_mapping.build_proxy_data(data, dynamic_keys_mapping)
@@ -79,10 +81,16 @@ class Edge:
         #print(self.pred_name, self.morph_name, self.order)
         proxy_data = self._io_mapping.build_proxy_data(data, dynamic_keys_mapping)
         # print(proxy_data)
-        self.preprocess(data)
+        if (self.use_proxy_data):
+            self.preprocess(proxy_data)
+        else:
+            self.preprocess(data)
         self._throw_if_not_set(proxy_data, self.mandatory_keys)
         self.morph_f.func(proxy_data)
-        self.postprocess(data)
+        if (self.use_proxy_data):
+            self.postprocess(proxy_data)
+        else:
+            self.postprocess(data)
 
     def _throw_if_not_set(self, data, mandatory_keys: Sequence[str]):
         for k in mandatory_keys:
@@ -124,8 +132,8 @@ class ExecutableProgramEdge(Edge):
                  trailing_args_keys=(),  # "local keys" where trailing args are stored
                  remote=False,
                  stdout_processor=None,
-                 chaining_command_at_start='',
-                 chaining_command_at_end='',
+                 chaining_command_at_start=lambda d: '',
+                 chaining_command_at_end=lambda d: '',
                  ):
         #predicate = predicate if predicate is not None else dummy_predicate
         self._output_dict = output_dict
@@ -145,8 +153,8 @@ class ExecutableProgramEdge(Edge):
         args_str = build_args_line(data, self._keyword_names, self._flag_names, self._trailing_args_keys)
         working_dir = data[self._working_dir_key]
         stdout_lines, stderr_lines = self._comm.execute_program(self._program_name, args_str, working_dir,
-                                                                self.chaining_command_at_start,
-                                                                self.chaining_command_at_end)
+                                                                self.chaining_command_at_start(data),
+                                                                self.chaining_command_at_end(data))
         output_data = self._output_dict
         if self._stdout_processor:
             stdout_data = self._stdout_processor(data, stdout_lines)
@@ -339,7 +347,10 @@ def make_dump(dump_name_format, format_keys=(), omit=None, method='pickle'):
         if omit is None:
             dumped_d = d
         else:
-            dumped_d = {key: val for key, val in d.items() if not key in omit}
+            if (isinstance(d, aux.ProxyDict)):
+                dumped_d = {key: val for key, val in d._data.items() if not key in omit}
+            else:
+                dumped_d = {key: val for key, val in d.items() if not key in omit}
         if method == 'pickle':
             with open(dump_path, 'wb') as f:
                 pickle.dump(dumped_d, f)

comsdk/research.py

 import pickle
 from datetime import date
-from typing import Sequence, Mapping
-from typing_extensions import TypedDict
+from typing import Sequence, Mapping, TypedDict
 
 from comsdk.comaux import *
 from comsdk.communication import BaseCommunication, LocalCommunication, SshCommunication, Host