namespace bts.BlueOrb { using BCG; using System.IO; using System.Diagnostics; using System; [DataStreamUserRespresentation("PythonDataframeStream")] public class PythonDataframeStream : ViewStream { private ViewStream _parent_str; private readonly string _in_csv_file_name; private readonly string _out_csv_filename; private readonly string _transform_function; public PythonDataframeStream(ViewStream parent_str, string transform_function) { var guid = System.Guid.NewGuid(); _in_csv_file_name = $"pdfs_in_{guid}.csv"; _out_csv_filename = $"pdfs_out_{guid}.csv"; _parent_str = new CSVSaverStream(parent_str, _in_csv_file_name); _transform_function = transform_function; } protected override void myOnFirstSubscriber() { _parent_str.onChange += _parent_str_onchange; } protected override void myOnLastSubscriber() { _parent_str.onChange -= _parent_str_onchange; } private void _parent_str_onchange(object dp_obj) { Log.assert(dp_obj != null); // the file _in_csv_file_name is now populated. // call py_func_wrapper_main and pass this file name as well // as the output file name // py_func_wrapper_main reads _in_csv_file_name into the df and calls // transform function, which returns the tranformed df. py_func_wrapper_main then // writes it into the _out_csv_filename file. // now this function should read _out_csv_filename into a dp // and fire it _compute_and_write(); //ViewStream rs = new SchemaGuesserStream(new CSVReaderStream(_out_csv_filename)); ViewStream rs = new CSVReaderStream(_out_csv_filename); IBlueRowCollection nrows = null; rs.onChange += (dp_obj) => { nrows = (dp_obj as DataPacket).newRows; }; var dp = BlueDataUtility.build_dp(nrows, myLastValue as DataPacket); myFireEvent(dp); } private void _compute_and_write() { File.WriteAllText("user_code.py", _transform_function); // wrapper function string py_func_wrapper_main = @" import sys from user_code import transform import pandas as pd if __name__ == '__main__': input_file_name = sys.argv[1] output_file_name = sys.argv[2] # 1. load input_file_name into df # 2. do some test calc # 3. write the output of 2 into output_file_name df = pd.read_csv(input_file_name) df_out = transform(df) df_out.to_csv(output_file_name, index=False)"; File.WriteAllText("runner.py", py_func_wrapper_main); ProcessStartInfo psi = new ProcessStartInfo { FileName = "python", Arguments = $"runner.py {_in_csv_file_name} {_out_csv_filename}", RedirectStandardOutput = false, UseShellExecute = false }; var proc = Process.Start(psi); proc.WaitForExit(); } #region Unused test code private void _simple_test() { // actual function that needs to be called //////////////////////// string py_real_function = @" def test(x, out_path): rv = x * 10 with open(out_path, 'w') as f: f.write(str(rv)) return rv"; // the function is in a string object. File.WriteAllText("user_code.py", py_real_function); // runner app (top level main function) ////////////////////////////// string py_func_wrapper_main = @" import sys from user_code import test if __name__ == '__main__': arg = int(sys.argv[1]) out_path = sys.argv[2] rv = test(arg, out_path) rv = rv + 12 with open(out_path, 'w') as f:to \ f.write(str(rv))"; File.WriteAllText("runner.py", py_func_wrapper_main); string outputPath = "py_out.txt"; ProcessStartInfo psi = new ProcessStartInfo { FileName = "python", Arguments = $"runner.py 4 {outputPath}", RedirectStandardOutput = false, UseShellExecute = false }; var proc = Process.Start(psi); proc.WaitForExit(); // Now read the output written by the function string result = File.ReadAllText(outputPath); Console.WriteLine(result); // Should print "6" } #endregion } }