diff --git a/extensions/business/fastapi/_ai4everyone/ai4e_deploy.py b/extensions/business/ai4e/ai4e_deploy.py similarity index 99% rename from extensions/business/fastapi/_ai4everyone/ai4e_deploy.py rename to extensions/business/ai4e/ai4e_deploy.py index 1dfbb51c..2ade6bd1 100644 --- a/extensions/business/fastapi/_ai4everyone/ai4e_deploy.py +++ b/extensions/business/ai4e/ai4e_deploy.py @@ -1,6 +1,7 @@ from naeural_core.business.default.web_app.fast_api_web_app import FastApiWebAppPlugin as BasePlugin + _CONFIG = { **BasePlugin.CONFIG, @@ -9,7 +10,7 @@ "PROCESS_DELAY": 0, # 'PORT': 5001, - 'ASSETS': 'extensions/business/fastapi/_ai4everyone', + 'ASSETS': 'extensions/business/ai4e', 'JINJA_ARGS': { 'html_files': [ { diff --git a/extensions/business/utils/ai4e_utils.py b/extensions/business/ai4e/ai4e_utils.py similarity index 88% rename from extensions/business/utils/ai4e_utils.py rename to extensions/business/ai4e/ai4e_utils.py index 664e4a40..f33a8b30 100644 --- a/extensions/business/utils/ai4e_utils.py +++ b/extensions/business/ai4e/ai4e_utils.py @@ -1,4 +1,4 @@ -from ratio1 import Session, Pipeline, Instance +# from ratio1 import Session, Pipeline, Instance from datetime import datetime import numpy as np @@ -63,9 +63,9 @@ def get_job_config( class Job: - def __init__(self, owner, session: Session, job_id: str, node_id: str, pipeline: str, signature: str, instance: str): + def __init__(self, owner, job_id: str, node_id: str, pipeline: str, signature: str, instance: str): self.owner = owner - self.session = session + # self.session = session self.job_id = job_id # TODO: instead of single values for instance_id and the rest, use a dict to # have a set of properties for each subtask(each signature) @@ -249,7 +249,6 @@ def maybe_update_data(self, data: dict, pipeline: str, signature: str): self.train_status['total_grid_iterations'] = train_data.get('NR_ALL_GRID_ITER', self.train_status['total_grid_iterations']) if 'TRAIN_FINAL' in data.keys(): self.train_final = data.get('TRAIN_FINAL', self.train_final) - self.deploy_job({}) # endif training finished # endif status for training @@ -333,24 +332,24 @@ def get_train_status(self): # 'full_last_payload': self.train_status_full_payload } - def __get_pipeline_and_instance(self, node=None, pipeline=None, signature=None, instance_id=None): - node = node or self.node_id - active_pipelines = self.session.get_active_pipelines(node=node) - if active_pipelines is None: - return None, None, f"Node_ID {node} not found" - pipeline = pipeline or self.pipeline - curr_pipeline = self.session.attach_to_pipeline(node=node, name=pipeline) - if curr_pipeline is None: - return None, None, f"Pipeline {pipeline} not found on {node}" - signature = signature or self.signature - instance_id = instance_id or self.instance_id - curr_instance = curr_pipeline.attach_to_plugin_instance( - signature=signature, - instance_id=instance_id - ) - if curr_instance is None: - return curr_pipeline, None, f"Instance {instance_id} of {signature} not found on {pipeline}" - return curr_pipeline, curr_instance, "" + # def __get_pipeline_and_instance(self, node=None, pipeline=None, signature=None, instance_id=None): + # node = node or self.node_id + # active_pipelines = self.session.get_active_pipelines(node=node) + # if active_pipelines is None: + # return None, None, f"Node_ID {node} not found" + # pipeline = pipeline or self.pipeline + # curr_pipeline = self.session.attach_to_pipeline(node=node, name=pipeline) + # if curr_pipeline is None: + # return None, None, f"Pipeline {pipeline} not found on {node}" + # signature = signature or self.signature + # instance_id = instance_id or self.instance_id + # curr_instance = curr_pipeline.attach_to_plugin_instance( + # signature=signature, + # instance_id=instance_id + # ) + # if curr_instance is None: + # return curr_pipeline, None, f"Instance {instance_id} of {signature} not found on {pipeline}" + # return curr_pipeline, curr_instance, "" def send_instance_command(self, node=None, pipeline=None, signature=None, instance_id=None, **kwargs): node = node or self.node_id @@ -443,7 +442,7 @@ def publish_labels(self): return self.send_instance_command(publish=True) def start_train(self, body: dict): - self.session.P(f'Starting training for {self.job_id} on {self.node_id}...') + self.owner.P(f'Starting training for {self.job_id} on {self.node_id}...') success, err_msg = self.send_instance_update( config={ 'START_TRAINING': True, @@ -495,38 +494,58 @@ def deploy_configs(self, lst_allowed, deploy_ngrok_edge_label): 'MODEL_INSTANCE_ID': self.train_meta['MODEL_INSTANCE_ID'] }, 'DESCRIPTION': self.description, - 'OBJECTIVE_NAME': self.objective_name + 'OBJECTIVE_NAME': self.objective_name, + "INSTANCE_ID": self.job_id, } - pipeline = self.session.create_or_attach_to_pipeline( - node=chosen_node, - name=f'deploy_{self.job_id}', - data_source='ON_DEMAND_INPUT', - # plugins=[det_plugin_config] - ) - instance = pipeline.create_or_attach_to_plugin_instance( - signature='ai4e_custom_inference_agent', - instance_id=self.job_id, - config=instance_config + inference_pipeline_config = { + "TYPE": "ON_DEMAND_INPUT", + "NAME": f"deploy_{self.job_id}", + "PLUGINS": [ + { + "INSTANCES": [ + instance_config + ], + "SIGNATURE": "ai4e_custom_inference_agent", + } + ] + } + self.owner.cmadpi_start_pipeline( + config=inference_pipeline_config, + node_address=chosen_node, ) - pipeline.deploy() # END DETECTION PIPELINE # START FASTAPI PIPELINE + fastapi_instance_id = "AI4EveryoneDeploys" fastapi_instance_config = { "RESPONSE_FORMAT": self.owner.cfg_response_format, "NGROK_EDGE_LABEL": deploy_ngrok_edge_label, + "INSTANCE_ID": fastapi_instance_id, } - pipeline = self.session.create_or_attach_to_pipeline( - node=chosen_node, - name=f'AI4EveryoneDeploys', - data_source='VOID', - ) - instance = pipeline.create_or_attach_to_plugin_instance( - signature="AI4E_DEPLOY", - instance_id='AI4EveryoneDeploys', - config=fastapi_instance_config + fastapi_pipeline_name = f"AI4EveryoneDeploys" + fastapi_signature = "AI4E_DEPLOY" + fastapi_pipeline_config = { + "NAME": fastapi_pipeline_name, + "TYPE": "VOID", # Change to custom NetworkListener + "PLUGINS": [ + { + "INSTANCES": [ + { + fastapi_instance_config + } + ], + "SIGNATURE": fastapi_signature, + } + ] + } + self.owner.cmadpi_start_pipeline( + config=fastapi_pipeline_config, + node_address=chosen_node, ) - pipeline.deploy() - instance.send_instance_command( + self.send_instance_command( + node=chosen_node, + pipeline=fastapi_pipeline_name, + signature=fastapi_signature, + instance_id=fastapi_instance_id, command='REGISTER', command_params={ 'CONFIG': instance_config @@ -539,8 +558,9 @@ def deploy_job(self, body): if self.deployed: return True, "Job already deployed" self.started_deploying = True - lst_allowed = self.session.get_allowed_nodes() - self.session.P(f"Allowed nodes: {lst_allowed}") + # lst_allowed = self.session.get_allowed_nodes() + lst_allowed = self.owner.netmon.accessible_nodes() + self.owner.P(f"Allowed nodes: {lst_allowed}") if len(lst_allowed) == 0: self.started_deploying = False return False, "No node available at the moment." diff --git a/extensions/business/fastapi/_ai4everyone/ai4everyone.py b/extensions/business/ai4e/ai4everyone.py similarity index 82% rename from extensions/business/fastapi/_ai4everyone/ai4everyone.py rename to extensions/business/ai4e/ai4everyone.py index c2068145..2e55c247 100644 --- a/extensions/business/fastapi/_ai4everyone/ai4everyone.py +++ b/extensions/business/ai4e/ai4everyone.py @@ -1,14 +1,15 @@ -from ratio1 import Payload, Session +# from ratio1 import Payload, Session from naeural_core.business.default.web_app.fast_api_web_app import FastApiWebAppPlugin as BasePlugin -from extensions.business.utils.ai4e_utils import AI4E_CONSTANTS, Job, get_job_config, job_data_to_id +from naeural_core.business.mixins_libs.network_processor_mixin import _NetworkProcessorMixin +from extensions.business.ai4e.ai4e_utils import AI4E_CONSTANTS, Job, get_job_config, job_data_to_id __VER__ = '0.1.0.0' _CONFIG = { **BasePlugin.CONFIG, - 'SAVE_PERIOD': 60, + 'SAVE_PERIOD': 300, 'REQUEST_TIMEOUT': 10, "PROCESS_DELAY": 0, "DEPLOY_NGROK_EDGE_LABEL": None, @@ -16,7 +17,7 @@ "DEBUG_WEB_APP": True, # 'PORT': 5000, - 'ASSETS': 'extensions/business/fastapi/_ai4everyone', + 'ASSETS': 'extensions/business/ai4e', 'JINJA_ARGS': { 'html_files': [ { @@ -32,81 +33,120 @@ } -class AI4EveryonePlugin(BasePlugin): +class AI4EveryonePlugin( + BasePlugin, + _NetworkProcessorMixin +): CONFIG = _CONFIG def __init__(self, **kwargs): - self.__init_done = False + # self.__init_done = False super(AI4EveryonePlugin, self).__init__(**kwargs) self.jobs_data = {} self.requests_responses = {} self.last_persistence_save = self.time() self.request_cache = {} self.force_persistence = False - # !!!This approach, although works, will not be allowed in the future because it's not safe - self.session = Session( - name=f'{self.str_unique_identification}', - config=self.global_shmem['config_communication']['PARAMS'], - log=self.log, - bc_engine=self.global_shmem[self.ct.BLOCKCHAIN_MANAGER], - on_payload=self.on_payload, - ) return def on_init(self): super(AI4EveryonePlugin, self).on_init() + self.network_processor_init() self.jobs_data = self.load_persistence_data() - self.__init_done = True + # self.__init_done = True return - """SESSION SECTION""" + """PAYLOAD HANDLING SECTION""" if True: - def on_payload(self, sess: Session, node_id: str, pipeline: str, signature: str, instance: str, payload: Payload): - if signature.lower() not in AI4E_CONSTANTS.RELEVANT_PLUGIN_SIGNATURES: - return - if not self.__init_done: - sess.P(f"[DEBUG_AI4E]Session not initialized yet, ignoring payload.") - return - is_status = payload.data.get('IS_STATUS', False) - is_final_dataset_status = payload.data.get('IS_FINAL_DATASET_STATUS', False) + def payload_handler_helper(self, data): + is_status = data.get('IS_STATUS', False) + is_final_dataset_status = data.get('IS_FINAL_DATASET_STATUS', False) if is_status or is_final_dataset_status: try: - self.maybe_update_job_data(node_id, pipeline, signature, instance, payload) + self.maybe_update_job_data(data) except Exception as e: if self.cfg_debug_web_app: self.P(f"[DEBUG_AI4E]Error while updating job data: {e}") # endtry else: try: - self.register_request_response(node_id, pipeline, signature, instance, payload) + self.register_request_response(data) except Exception as e: if self.cfg_debug_web_app: self.P(f"[DEBUG_AI4E]Error while registering request response: {e}") # endtry + # endif is_status or is_final_dataset_status return - def maybe_update_job_data(self, node_id: str, pipeline: str, signature: str, instance: str, payload: Payload): + @_NetworkProcessorMixin.payload_handler(signature="AI4E_CROP_DATA") + def on_payload_crop_data(self, data): + return self.payload_handler_helper(data) + + @_NetworkProcessorMixin.payload_handler(signature="AI4E_LABEL_DATA") + def on_payload_label_data(self, data): + return self.payload_handler_helper(data) + + @_NetworkProcessorMixin.payload_handler(signature="SECOND_STAGE_TRAINING_PROCESS") + def on_payload_second_stage_training(self, data): + return self.payload_handler_helper(data) + + @_NetworkProcessorMixin.payload_handler(signature="GENERAL_TRAINING_PROCESS") + def on_payload_general_training(self, data): + return self.payload_handler_helper(data) + + @_NetworkProcessorMixin.payload_handler(signature="MINIO_UPLOAD_DATASET") + def on_payload_minio_upload_dataset(self, data): + return self.payload_handler_helper(data) + + # def on_payload(self, sess: Session, node_id: str, pipeline: str, signature: str, instance: str, payload: Payload): + # if signature.lower() not in AI4E_CONSTANTS.RELEVANT_PLUGIN_SIGNATURES: + # return + # if not self.__init_done: + # sess.P(f"[DEBUG_AI4E]Session not initialized yet, ignoring payload.") + # return + # is_status = payload.data.get('IS_STATUS', False) + # is_final_dataset_status = payload.data.get('IS_FINAL_DATASET_STATUS', False) + # if is_status or is_final_dataset_status: + # try: + # self.maybe_update_job_data(node_id, pipeline, signature, instance, payload) + # except Exception as e: + # if self.cfg_debug_web_app: + # self.P(f"[DEBUG_AI4E]Error while updating job data: {e}") + # # endtry + # else: + # try: + # self.register_request_response(node_id, pipeline, signature, instance, payload) + # except Exception as e: + # if self.cfg_debug_web_app: + # self.P(f"[DEBUG_AI4E]Error while registering request response: {e}") + # # endtry + # return + + def maybe_update_job_data(self, data): + payload_path = data.get(self.ct.PAYLOAD_DATA.EE_PAYLOAD_PATH) + if payload_path is None: + return + node_id, pipeline, signature, instance = payload_path job_id = job_data_to_id(node_id, pipeline, signature, instance) if job_id not in self.jobs_data: self.jobs_data[job_id] = Job( - owner=self, - session=self.session, job_id=job_id, + owner=self, job_id=job_id, node_id=node_id, pipeline=pipeline, signature=signature, instance=instance ) job = self.jobs_data[job_id] job.maybe_update_data( - data=payload.data, + data=data, pipeline=pipeline, signature=signature ) return - def register_request_response(self, node_id: str, pipeline: str, signature: str, instance: str, payload: Payload): - request_id = payload.data.get('REQUEST_ID') + def register_request_response(self, data: dict): + request_id = data.get('REQUEST_ID') if request_id is None: return - self.requests_responses[request_id] = payload + self.requests_responses[request_id] = data return def send_request(self, job: Job, **kwargs): @@ -121,8 +161,8 @@ def solve_postponed_process_request( self, request_id: str, job: Job, request_ts: float, **request_kwargs ): if request_id in self.requests_responses: - response = self.requests_responses.pop(request_id) - return response.data + data = self.requests_responses.pop(request_id) + return data if self.time() - request_ts > self.cfg_request_timeout: return {"error": "Request timed out"} return self.create_postponed_request( @@ -168,8 +208,7 @@ def solve_postponed_process_sample_request( **request_kwargs ): if request_id in self.requests_responses: - response = self.requests_responses.pop(request_id) - response_data = response.data + response_data = self.requests_responses.pop(request_id) sample_filename = response_data.get('SAMPLE_FILENAME') if sample_filename is None: return {"error": "Sample not found"} @@ -267,7 +306,7 @@ def start_job(self, nodeAddress: str, job_config: dict): data_source="VOID", plugins=[job_config] ).deploy() - """END SESSION SECTION""" + """END PAYLOAD HANDLING SECTION""" """ENDPOINTS SECTION""" if True: @@ -337,6 +376,7 @@ def publish_job( return None @BasePlugin.endpoint(method="post") + # TODO: receive list of votes def vote(self, job_id: str, filename: str, label: str): if job_id in self.jobs_data: success, result = self.jobs_data[job_id].send_vote(filename=filename, label=label) @@ -506,8 +546,7 @@ def load_persistence_data(self): node_id, pipeline = data.get('node_id'), data.get('pipeline') signature, instance = data.get('signature'), data.get('instance_id') res[key] = Job( - owner=self, - session=self.session, job_id=key, + owner=self, job_id=key, node_id=node_id, pipeline=pipeline, signature=signature, instance=instance ) @@ -518,6 +557,7 @@ def load_persistence_data(self): def process(self): super(AI4EveryonePlugin, self).process() + self.network_processor_loop() self.maybe_persistence_save() return diff --git a/extensions/business/fastapi/_ai4everyone/assets/bundle.js b/extensions/business/ai4e/assets/bundle.js similarity index 100% rename from extensions/business/fastapi/_ai4everyone/assets/bundle.js rename to extensions/business/ai4e/assets/bundle.js diff --git a/extensions/business/fastapi/_ai4everyone/assets/deploy_index.html b/extensions/business/ai4e/assets/deploy_index.html similarity index 100% rename from extensions/business/fastapi/_ai4everyone/assets/deploy_index.html rename to extensions/business/ai4e/assets/deploy_index.html diff --git a/extensions/business/fastapi/_ai4everyone/assets/index.html b/extensions/business/ai4e/assets/index.html similarity index 100% rename from extensions/business/fastapi/_ai4everyone/assets/index.html rename to extensions/business/ai4e/assets/index.html diff --git a/extensions/business/dauth/dauth_manager.py b/extensions/business/dauth/dauth_manager.py index 4d612256..a4e47e31 100644 --- a/extensions/business/dauth/dauth_manager.py +++ b/extensions/business/dauth/dauth_manager.py @@ -55,7 +55,6 @@ "EE_NGROK_EDGE_LABEL_EPOCH_MANAGER", "EE_NGROK_EDGE_LABEL_RELEASE_APP", "EE_NGROK_EDGE_LABEL_DAUTH_MANAGER", - "EE_NGROK_EDGE_LABEL_CSTORE_MANAGER", "EE_NGROK_EDGE_LABEL_DEEPLOY_MANAGER", "EE_NGROK_EDGE_LABEL_TUNNELS_MANAGER", @@ -64,7 +63,6 @@ "EE_CLOUDFLARE_TOKEN_EPOCH_MANAGER", "EE_CLOUDFLARE_TOKEN_RELEASE_APP", "EE_CLOUDFLARE_TOKEN_DAUTH_MANAGER", - "EE_CLOUDFLARE_TOKEN_CSTORE_MANAGER", "EE_CLOUDFLARE_TOKEN_DEEPLOY_MANAGER", "EE_CLOUDFLARE_TOKEN_TUNNELS_MANAGER", "EE_CLOUDFLARE_TOKEN_LIVENESS_API", diff --git a/ver.py b/ver.py index ce747d36..735af3c1 100644 --- a/ver.py +++ b/ver.py @@ -1,2 +1,2 @@ -__VER__ = '2.9.510' +__VER__ = '2.9.520'