mysterious-hair-40059
03/16/2023, 3:13 PMTypeTransformerFailedError: Type of Val 'typing.Dict' is not an instance of typing.Dict[str, str]
abundant-hamburger-66584
03/16/2023, 3:21 PMmysterious-hair-40059
03/16/2023, 3:57 PMclass FlyteRemote:
def __init__(self):
...
def _register_execute_workflow(
self,
wf: flytekit.core.workflow.WorkflowBase,
inputs: Dict,
) -> flytekit.remote.executions.FlyteWorkflowExecution:
self.remote.register_workflow(wf, self.serialization_settings)
fetched_wf = self.remote.fetch_workflow(
name=wf.name,
version=self.version,
)
execution_response = self.remote.execute(
entity=fetched_wf,
inputs=inputs,
version=self.version,
image_config=self.image_config,
)
return execution_response
def run_regular_simulation_wf(
self,
inputs: Dict,
) -> flytekit.remote.executions.FlyteWorkflowExecution:
wf_response = self._register_execute_workflow(wf=regular_simulation_cpu_wf, inputs=inputs)
@workflow
def regular_simulation_cpu_wf(
param1: Challenge,
param2: List[int],
param3: str,
param4: str,
param5: module.Model,
param6: module.Explorer,
param7: module.Landscape,
param8: int,
param9: str,
):
return simulation_cpu_task(
param1=param1,
param2=param2,
param3=param3,
param4=param4,
param5=param5,
param6=param6,
param7=param7,
param8=param8,
param9=param9,
)
abundant-hamburger-66584
03/16/2023, 4:17 PMsimulation_cpu_task
to be able to judge moreabundant-hamburger-66584
03/16/2023, 4:22 PMpyflyte run
from the command line might help isolate if the issues faster. Because you can nest workflows, you can create one for testing that wraps regular_simulation_cpu_wf
and pass in defaults.
Alternatively you can call regular_simulation_cpu_wf
from inside a notebook too like normal python.
@workflow
def test_regular_simulation_cpu_wf():
regular_simulation_cpu_wf([pass in values])
mysterious-hair-40059
03/16/2023, 4:42 PMregular_simulation_cpu_wf
is a one-task workflow, so the parameters of that workflow is the same as the parameters of the task. Below is the snippet of the simulation_cpu_task
@task(requests=Resources(cpu="1", mem="1Gi"))
def simulation_cpu_task(
param1: Challenge,
param2: List[int],
param3: str,
param4: str,
param5: module.Model,
param6: module.Explorer,
param7: module.Landscape,
param8: int,
param9: str,
) -> List[Campaign]:
"""CPU simulation task"""
simulation = module.Simulation(
param1=param1,
param2=param2,
param3=param3,
param4=param4,
param5=param5,
param6=param6,
param7=param7,
param8=param8,
param9=param9,
)
experiments = simulation.run_sim()
return experiments
Thank you very much. I'm currently testing out your suggestions to see if they work. I will revert.