Fhuad Balogun
03/16/2023, 3:13 PMTypeTransformerFailedError: Type of Val 'typing.Dict' is not an instance of typing.Dict[str, str]
Evan Sadler
03/16/2023, 3:21 PMFhuad Balogun
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,
)
Evan Sadler
03/16/2023, 4:17 PMsimulation_cpu_task
to be able to judge morepyflyte 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])
Fhuad Balogun
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.