Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions dcp/api/compute_for.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ def compute_for(*args, **kwargs):
# 3. For each input element, dereference js_ref if from dcp-client, add a guard if pythonmonkey will mutate it, else as it as-is.
if job_input_idx != None:
if hasattr(args[job_input_idx], 'js_ref') and dry.class_manager.reg.find_from_js_instance(args[job_input_idx].js_ref):
args[job_input_idx] = args[job_input_idx]
args[job_input_idx] = args[job_input_idx].js_ref
else:
try:
tmp = args[job_input_idx][0]
Expand All @@ -65,7 +65,7 @@ def compute_for(*args, **kwargs):

if job_args_idx != None:
if hasattr(args[job_args_idx], 'js_ref') and dry.class_manager.reg.find_from_js_instance(args[job_args_idx].js_ref):
args[job_args_idx] = args[job_args_idx]
args[job_args_idx] = args[job_args_idx].js_ref
else:
try:
tmp = args[job_args_idx][0]
Expand Down
11 changes: 4 additions & 7 deletions dcp/api/job.py
Original file line number Diff line number Diff line change
Expand Up @@ -78,8 +78,8 @@ def _before_exec(self, *args, **kwargs):
serialized_input_data = []
if len(self.serializers):
validate_serializers(self.serializers)
if hasattr(self.js_ref.jobInputData, 'js_ref') and dry.class_manager.reg.find_from_js_instance(self.js_ref.jobInputData.js_ref):
serialized_input_data = self.js_ref.jobInputData.js_ref
if hasattr(self.jobInputData, 'js_ref') and dry.class_manager.reg.find_from_js_instance(self.jobInputData.js_ref):
serialized_input_data = self.jobInputData.js_ref
elif isinstance(self.js_ref.jobInputData, list) or utils.instanceof(self.js_ref.jobInputData, pm.globalThis.Array):
for input_slice in self.js_ref.jobInputData:
# TODO - find better solution
Expand All @@ -95,11 +95,8 @@ def _before_exec(self, *args, **kwargs):
serialized_input_data.append(serialized_slice)
else:
serialized_input_data = self.js_ref.jobInputData
if hasattr(self.js_ref.jobArguments, 'js_ref') and dry.class_manager.reg.find_from_js_instance(self.js_ref.jobArguments.js_ref):
serialized_arguments = self.js_ref.jobArguments.js_ref
# if utils.instanceof(self.js_ref.jobArguments, pm.eval("globalThis.dcp.compute.RemoteDataSet")):
# convertToURL = pm.eval('(urlString) => new URL(urlString)')
# self.js_ref.jobArguments.forEach(lambda argument: serialized_arguments.append(convertToURL(argument)))
Comment thread
wiwichips marked this conversation as resolved.
if hasattr(self.jobArguments, 'js_ref') and dry.class_manager.reg.find_from_js_instance(self.jobArguments.js_ref):
serialized_arguments = self.jobArguments.js_ref
else:
for argument in self.js_ref.jobArguments:
# TODO - find better solution
Expand Down
50 changes: 50 additions & 0 deletions examples/remote-data-job-deploy.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
from http.server import BaseHTTPRequestHandler, HTTPServer
import threading
import json

import dcp; dcp.init()

class SimpleHandler(BaseHTTPRequestHandler):
def do_GET(self):
path = int(self.path.strip('/'))
self.send_response(200)
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(path).encode())

#start http server on seperate thread
server_address = ("localhost", 12345)
httpd = HTTPServer(server_address, SimpleHandler)
print("Server running at http://localhost:12345")
server_thread = threading.Thread(target=httpd.serve_forever)
server_thread.daemon = True
server_thread.start()

def workfn(x):
import dcp
dcp.progress()
return x * x


# create job with RemoteDataPattern
my_rdp = dcp.compute.RemoteDataPattern('http://localhost:12345/{slice}',5)
Comment thread
JosephAcernese marked this conversation as resolved.
my_j = dcp.compute_for(my_rdp, workfn)


# add event listeners
my_j.on('readystatechange', print)
my_j.on('result', print)
my_j.on('error', print)

@my_j.on('accepted')
def accepted_handler(ev):
print(f"jobid = {my_j.id}")

my_j.public.name = 'simple bifrost2 remote data pattern example'

my_j.exec()
res = my_j.wait()

print(">>>>>>>>>>>>>>>>>>>>>>>>>> RESULTS ARE IN")
print(res)
Loading