Spaces:
Sleeping
Sleeping
File size: 3,966 Bytes
69601d4 1b179fe 69601d4 d7fd1be 69601d4 1b179fe 69601d4 d7fd1be 69601d4 d7fd1be 69601d4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | from ..components.metadataGenerator import MetadataGenerator
from ..workflows.reportingWorkflow import reportingToolWorkflow
from ..components.speechToText import SpeechToText
from ..utils.exceptions import CustomException
from ..utils.functions import readYaml
from ..components import replManager
from supabase import create_client
from urllib.request import urlopen
from ..utils.logger import logger
import orjson
import time
import os
class CompletePipeline:
def __init__(self):
logger.info("Initializing CompletePipeline components.")
self.speechToTextModule = SpeechToText()
self.metadataGenerator = MetadataGenerator()
self.yamlPath = os.path.join(os.getcwd(), "params.yaml")
self.supabaseClient = create_client(
supabase_url = os.environ["SUPABASE_URL"],
supabase_key = os.environ["SUPABASE_KEY"]
)
def generateMetadata(self, projectId: str) -> dict:
try:
dataFiles = [x.get("name") for x in self.supabaseClient.storage.from_("AnalyticsHub").list(path = projectId) if x.get("name").endswith(".parquet")]
results = ""
for fileName in dataFiles:
dataframeName = fileName.replace(".parquet", "")
codeString = readYaml(self.yamlPath)["attributeInfoCode"].format(dataframeName = dataframeName, projectId = projectId)
results += replManager.run(codeString)
metadataChain = self.metadataGenerator.getMetadataChain()
metadata = metadataChain.invoke({"metadata": results})
metadataParts = metadata.split("```")
metadata = metadataParts[-2]
metadata = orjson.loads("\n".join(metadata.split("\n")[1:]).encode())
return metadata
except Exception as e:
logger.error(CustomException(e))
raise CustomException(e)
def generateChart(self, inputQuery: str, metadata: dict, projectId: str) -> dict:
try:
response = reportingToolWorkflow.invoke({
"inputQuery": inputQuery,
"metadata": metadata,
"projectId": projectId
})
return response
except Exception as e:
logger.error(CustomException(e))
raise CustomException(e)
def generateChartFromPanel(self, projectId: str, chartType: str, xAxis: str, yAxis: str, aggregationMetric: str, dataSource: str) -> dict:
try:
blendConfigUrl = os.environ["FILE_URL"].format(projectId = projectId, fileName = "blendConfig.json").replace(".parquet", "") + f"?cb={int(time.time())}"
blendConfig = orjson.loads(urlopen(blendConfigUrl).read())
blendedTables = list(blendConfig.keys())
if dataSource in blendedTables:
tablesUsed = blendConfig[dataSource].get("tables")
joinTypes = blendConfig[dataSource].get("joinTypes")
blendOn = blendConfig[dataSource].get("blendOn")
response = replManager.run(f"getDataForChart(projectId='{projectId}', chartType='{chartType}', xAxis='{xAxis}', yAxis='{yAxis}', aggregationMetric='{aggregationMetric}', tablesUsed={tablesUsed}, joinTypes={joinTypes}, blendOn={blendOn})")
else:
response = replManager.run(f"getDataForChart(projectId='{projectId}', chartType='{chartType}', xAxis='{xAxis}', yAxis='{yAxis}', aggregationMetric='{aggregationMetric}', tablesUsed='{dataSource}')")
response = orjson.loads(response.encode())
return response
except Exception as e:
logger.error(CustomException(e))
raise CustomException(e)
def speechToText(self, b64String: str) -> str:
try:
return self.speechToTextModule.getTranscript(b64String = b64String)
except Exception as e:
logger.error(CustomException(e))
raise CustomException(e)
|