ll := import("@platforma-sdk/workflow-tengo:ll")
assets := import("@platforma-sdk/workflow-tengo:assets")
exec := import("@platforma-sdk/workflow-tengo:exec")
validation := import("@platforma-sdk/workflow-tengo:validation")
objects := import("@platforma-sdk/workflow-tengo:objects")
util := import("@platforma-sdk/workflow-tengo:pframes.util")
constants := import("@platforma-sdk/workflow-tengo:pframes.constants")
execConstants := import("@platforma-sdk/workflow-tengo:exec.constants")
xsvImportPt := import("@platforma-sdk/workflow-tengo:pframes.xsv-import-pt")

transposeXsvSw := assets.importSoftware("@platforma-open/milaboratories.software-small-binaries:table-converter")
pfconvSw := assets.importSoftware("@milaboratories/software-pframes-conv:main")


importDirTpl := assets.importTemplate("@platforma-sdk/workflow-tengo:pframes.import-dir")

renderOps := func(defaults, ...ops) {
	o := { }

	for k, v in defaults {
		o[k] = v
	}

	if len(ops) > 0 {
		for k, v in ops[0] {
			o[k] = v
		}
	}
	return o
}

getAxesSpec := func(spec) {
	axes := []
	for ax in spec.axes {
		axes = append(axes, ax.spec)
	}
	return axes
}

getColumnSpec := func(axesSpec, col) {
	spec := {
		kind: constants.KIND_P_COLUMN,
		axesSpec: axesSpec
	}

	for k, v in col.spec {
		spec[k] = v
	}

	return objects.deleteUndefined(spec)
}

































importFile := func(xsvFile, xsvType, spec, ...ops) {
	util.validateXsvType(xsvType)

	validation.assertType(spec, util.PFCONV_IMPORT_CFG_SCHEMA)


	ll.assert(
		spec.partitionKeyLength == undefined || spec.partitionKeyLength < len(spec.axes),
		"partitionKeyLength (%d) must be strictly less than the number of axes (%d)",
		spec.partitionKeyLength,
		len(spec.axes)
	)

	storageFormat := spec.storageFormat
	if is_undefined(storageFormat) {
		storageFormat = "Parquet"
	}

	xsvFileName := "file." + xsvType

	ops := renderOps({ dataOnly: false, splitDataAndSpec: false, queue: execConstants.LIGHT_QUEUE, inputCache: undefined }, ops...)

	if ops.transpose != undefined {
		ll.assert(xsvType != "parquet",
			"transpose is not supported with xsvType 'parquet', please use 'csv' or 'tsv' instead")

		tops := ops.transpose

		transposedFileName := "fileTransposed." + xsvType


		transposeXsv := exec.builder().
			software(transposeXsvSw).
			printErrStreamToStdout()

		if ops.queue != undefined {
			transposeXsv.setQueue(ops.queue)
		}
		if ops.cpu != undefined {
			transposeXsv.cpu(ops.cpu)
		}
		if ops.mem != undefined {
			transposeXsv.mem(ops.mem)
		}
		if ops.inputCache != undefined && ops.inputCache > 0 {
			transposeXsv.cacheInputs(ops.inputCache)
		}

		if (tops.separator != undefined) {
			transposeXsv = transposeXsv.arg("-input-separator").arg(tops.separator)
		}
		if (tops.pAxisIdx != undefined) {
			transposeXsv = transposeXsv.arg("-sample-column-i").arg("" + tops.pAxisIdx) // always convert to string
		}
		if (tops.pAxisName != undefined) {
			transposeXsv = transposeXsv.arg("-sample-column-name").arg(tops.pAxisName)
		}
		if (tops.pAxisSearch != undefined) {
			transposeXsv = transposeXsv.arg("-sample-column-search").arg(tops.pAxisSearch)
		}
		if (tops.pAxisNameOverride != undefined) {
			transposeXsv = transposeXsv.arg("-sample-label").arg(tops.pAxisNameOverride)
		}
		if (tops.sAxisSearch != undefined) {
			transposeXsv = transposeXsv.arg("-metric-columns-search").arg(tops.sAxisSearch)
		}
		if (tops.sAxisName != undefined) {
			transposeXsv = transposeXsv.arg("-metric-label").arg(tops.sAxisName)
		}
		if (tops.valueName != undefined) {
			transposeXsv = transposeXsv.arg("-value-label").arg(tops.valueName)
		}
		if (tops.separatorOverride != undefined) {
			transposeXsv = transposeXsv.arg("-separator").arg(tops.separatorOverride)
		}

		transposeXsv = transposeXsv.
			arg(xsvFileName).
			arg(transposedFileName).
			addFile(xsvFileName, xsvFile).
			saveFile(transposedFileName).
			run()

		xsvFile = transposeXsv.getFile(transposedFileName)
	}

	if storageFormat == "Parquet" {
		return xsvImportPt.importFileParquet(xsvFile, xsvType, spec, ops)
	}
	ll.assert(xsvType != "parquet",
		"storageFormat '%v' is not supported with xsvType 'parquet', please use storageFormat 'Parquet'", storageFormat)


	pureSpec := util.purifySpec(spec)


	pfconv := exec.builder().
		software(pfconvSw).
		printErrStreamToStdout().
		arg("importCsv").
		arg(xsvFileName).
		arg("-p").arg("spec.json").
		arg("-o").arg("out").
		addFile(xsvFileName, xsvFile).
		writeFile("spec.json", pureSpec).
		processWorkdir("pf", importDirTpl, pureSpec)

	if ops.queue != undefined {
		pfconv.setQueue(ops.queue)
	}
	if ops.cpu != undefined {
		pfconv.cpu(ops.cpu)
	}
	if ops.mem != undefined {
		pfconv.mem(ops.mem)
	}
	if ops.inputCache != undefined && ops.inputCache > 0 {
		pfconv.cacheInputs(ops.inputCache)
	}

	pfconv = pfconv.run()


	pf := pfconv.getProcessorResult("pf")
	if ops.dataOnly {

		return pf
	}

	axesSpec := getAxesSpec(spec)

	result := {}
	for col in spec.columns {
		id := util.xsvColumnId(col)

		if ops.splitDataAndSpec {
			result[id] = {
				data: pf.getFutureInputField(id),
				spec: getColumnSpec(axesSpec, col)
			}
		} else {
			result[id + ".data"] = pf.getFutureInputField(id)
			result[id + ".spec"] = getColumnSpec(axesSpec, col)
		}
	}

	return result
}

export ll.toStrict({
	importFile: importFile
})