106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
from temporalio import workflow, client
|
|
from temporalio.worker import Worker
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
from laborious.workflows.predictions_batch import PredictionsBatch
|
|
from laborious.activities.activities import Activities
|
|
import os
|
|
import logging
|
|
from sientia_do.notifications.handlers import NotificationHandler
|
|
|
|
|
|
async def main():
|
|
host = os.getenv('TEMPORAL_HOST', 'localhost:7233')
|
|
logger = logging.getLogger(__name__)
|
|
stream_handler = logging.StreamHandler()
|
|
stream_handler.setLevel(
|
|
os.getenv('LOG_LEVEL', 'INFO').upper()
|
|
)
|
|
stream_handler.setFormatter(
|
|
logging.Formatter(
|
|
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
)
|
|
|
|
logger.addHandler(stream_handler)
|
|
|
|
notification_handler = NotificationHandler(
|
|
servers=os.getenv('NOTIFICATION_SERVERS', 'http://localhost:29092'),
|
|
logger=logger,
|
|
project_name=os.getenv('PROJECT_NAME', 'laborious'),
|
|
pipeline_name='-',
|
|
trigger_name='-',
|
|
model_name='-',
|
|
model='-'
|
|
)
|
|
|
|
postgres_config = {
|
|
'host': os.getenv('POSTGRES_HOST', 'localhost'),
|
|
'port': int(os.getenv('POSTGRES_PORT', '5432')),
|
|
'user': os.getenv('POSTGRES_USER', 'sientia'),
|
|
'password': os.getenv('POSTGRES_PASSWORD', 'sientia'),
|
|
'dbname': os.getenv('POSTGRES_DBNAME', 'sientia'),
|
|
'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '5')),
|
|
'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '20'))
|
|
}
|
|
|
|
mlflow_config = {
|
|
'host': os.getenv('MLFLOW_HOST', 'localhost'),
|
|
'port': int(os.getenv('MLFLOW_PORT', '5000')),
|
|
'username': os.getenv('MLFLOW_USERNAME', 'aignosi'),
|
|
'password': os.getenv('MLFLOW_PASSWORD', 'aignosi')
|
|
}
|
|
|
|
opc_config = {
|
|
'name': os.getenv('OPC_NAME', 'opc'),
|
|
'url': os.getenv('OPC_URL', 'opc.tcp://localhost:4840'),
|
|
'server_uri': os.getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
|
|
'cert_path': os.getenv('OPC_CERT_PATH', None),
|
|
'private_key_path': os.getenv('OPC_PRIVATE_KEY_PATH', None),
|
|
'server_cert_path': os.getenv('OPC_SERVER_CERT_PATH', None)
|
|
}
|
|
|
|
activities = Activities(
|
|
postgres_config=postgres_config,
|
|
mlflow_config=mlflow_config,
|
|
opc_config=opc_config,
|
|
logger=logger,
|
|
notification_handler=notification_handler
|
|
)
|
|
|
|
temporal_client = await client.Client.connect(target_host=host)
|
|
workers = [
|
|
Worker(
|
|
temporal_client,
|
|
task_queue='predictions',
|
|
workflows=[PredictionsBatch],
|
|
activities=[
|
|
# Base
|
|
activities.prepare_activity,
|
|
# MLFlow
|
|
activities.request_predict,
|
|
activities.request_transform,
|
|
# Gates
|
|
activities.input_gate,
|
|
activities.mlflow_response_gate,
|
|
activities.mlflow_content_gate,
|
|
activities.format_prediction,
|
|
activities.format_default_prediction,
|
|
activities.get_last_timestamp,
|
|
# OPC
|
|
activities.write_opc_data,
|
|
# Postgres
|
|
activities.load_custom_query,
|
|
activities.repeat_last_prediction,
|
|
activities.export_data_to_postgres
|
|
]
|
|
)
|
|
]
|
|
|
|
for w in workers:
|
|
await w.run()
|
|
|
|
if __name__ == '__main__':
|
|
import asyncio
|
|
asyncio.run(main())
|