replace auto-gen of XLS doc files and Modbus with python

scripts/build_interface.py

@@ -1,29 +1,112 @@
 from pathlib import Path
 import os
+import subprocess
+import shutil
 
 Import("env")
 
 
-def buildWeb():
-    os.chdir("interface")
-    print("Building web interface...")
+def get_pnpm_executable():
+    """Get the appropriate pnpm executable for the current platform."""
+    # Try different pnpm executable names
+    pnpm_names = ['pnpm', 'pnpm.cmd', 'pnpm.exe']
+    
+    for name in pnpm_names:
+        if shutil.which(name):
+            return name
+    
+    # Fallback to pnpm if not found
+    return 'pnpm'
+
+
+def run_command_in_directory(command, directory):
+    """Run a command in a specific directory."""
     try:
-        env.Execute("pnpm install")
-        env.Execute("pnpm typesafe-i18n")
-        with open("./src/i18n/i18n-util.ts") as r:
-            text = r.read().replace("Locales = 'pl'", "Locales = 'en'")
-        with open("./src/i18n/i18n-util.ts", "w") as w:
-            w.write(text)
-        print("Setting WebUI locale to 'en'")
-        env.Execute("pnpm build")
-        env.Execute("pnpm webUI")
-    finally:
-        os.chdir("..")
+        result = subprocess.run(
+            command,
+            shell=True,
+            cwd=directory,
+            check=True,
+            capture_output=True,
+            text=True
+        )
+        
+        if result.stdout:
+            print(result.stdout)
+        if result.stderr:
+            print(result.stderr)
+            
+        return True
+        
+    except subprocess.CalledProcessError as e:
+        print(f"Command failed: {command}")
+        print(f"Error: {e}")
+        if e.stdout:
+            print(f"Output: {e.stdout}")
+        if e.stderr:
+            print(f"Error output: {e.stderr}")
+        return False
+    except Exception as e:
+        print(f"Unexpected error running command '{command}': {e}")
+        return False
 
 
-# Don't buuld webUI if called from GitHub Actions
+def buildWeb():
+    interface_dir = Path("interface")
+    pnpm_exe = get_pnpm_executable()
+    
+    print("Building web interface...")
+    
+    # Check if interface directory exists
+    if not interface_dir.exists():
+        print(f"Error: Interface directory '{interface_dir}' not found!")
+        return False
+    
+    # Check if pnpm is available
+    if not shutil.which(pnpm_exe):
+        print(f"Error: '{pnpm_exe}' not found in PATH!")
+        return False
+    
+    try:
+        # Run pnpm commands in the interface directory
+        commands = [
+            f"{pnpm_exe} install",
+            f"{pnpm_exe} typesafe-i18n",
+            f"{pnpm_exe} build",
+            f"{pnpm_exe} webUI"
+        ]
+        
+        for command in commands:
+            print(f"Running: {command}")
+            if not run_command_in_directory(command, interface_dir):
+                return False
+        
+        # Modify i18n-util.ts file
+        i18n_file = interface_dir / "src" / "i18n" / "i18n-util.ts"
+        if i18n_file.exists():
+            with open(i18n_file, 'r') as r:
+                text = r.read().replace("Locales = 'pl'", "Locales = 'en'")
+            with open(i18n_file, 'w') as w:
+                w.write(text)
+            print("Setting WebUI locale to 'en'")
+        else:
+            print(f"Warning: {i18n_file} not found, skipping locale modification")
+        
+        print("Web interface build completed successfully!")
+        return True
+        
+    except Exception as e:
+        print(f"Error building web interface: {e}")
+        return False
+
+
+# Don't build webUI if called from GitHub Actions
 if "NO_BUILD_WEBUI" in os.environ:
     print("!! Skipping the build of the web interface !!")
 else:
     if not (env.IsCleanTarget()):
-        buildWeb()
+        success = buildWeb()
+        if not success:
+            print("Web interface build failed!")
+            # Optionally exit with error code
+            # sys.exit(1)

scripts/build_modbus_entity_parameters_post.py

@@ -0,0 +1,52 @@
+import subprocess
+import os
+import sys
+import shutil
+from pathlib import Path
+
+def get_python_executable():
+    """Get the appropriate Python executable for the current platform."""
+    # Try different Python executable names
+    python_names = ['python3', 'python', 'py']
+    
+    for name in python_names:
+        if shutil.which(name):
+            return name
+    
+    # Fallback to sys.executable if available
+    return sys.executable
+
+
+def csv_to_header(csv_file_path, header_file_path, script_path):
+
+    # Ensure the output directory exists
+    Path(header_file_path).parent.mkdir(parents=True, exist_ok=True)
+    
+    # delete the output file if it exists
+    if os.path.exists(header_file_path):
+        os.remove(header_file_path)
+    
+    # Read CSV file and pipe to Python script to generate header
+    python_exe = get_python_executable()
+    
+    with open(csv_file_path, 'r') as csv_file:
+        with open(header_file_path, 'w') as header_file:
+            subprocess.run(
+                [python_exe, script_path],
+                stdin=csv_file,
+                stdout=header_file,
+                check=True
+            )
+    
+    print(f"Generated header file: {header_file_path} ({os.path.getsize(header_file_path)} bytes)")
+
+
+def main():
+    csv_file = os.path.join("docs", "dump_entities.csv")
+    header_file = os.path.join("src", "core", "modbus_entity_parameters.hpp")
+    script_file = os.path.join("scripts", "update_modbus_registers.py")
+    
+    csv_to_header(csv_file, header_file, script_file)
+
+if __name__ == "__main__":
+    main()

scripts/build_modbus_entity_parameters_pre.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+import os
+Import("env")
+
+def create_dummy_modbus_header():
+    """Create a dummy modbus_entity_parameters.hpp so the first pass compiles."""
+    header_content = '''#include "modbus.h"
+#include "emsdevice.h"
+
+/*
+ * This file is auto-generated. Do not modify.
+*/
+
+// clang-format off
+
+namespace emsesp {
+
+using dt = EMSdevice::DeviceType;
+
+#define REGISTER_MAPPING(device_type, device_value_tag_type, long_name, modbus_register_offset, modbus_register_count) \\
+    { device_type, device_value_tag_type, long_name[0], modbus_register_offset, modbus_register_count }
+
+// IMPORTANT: This list MUST be ordered by keys "device_type", "device_value_tag_type" and "modbus_register_offset" in this order.
+const std::initializer_list<Modbus::EntityModbusInfo> Modbus::modbus_register_mappings = {};
+
+} // namespace emsesp
+
+// clang-format on
+'''
+    
+    header_path = Path("src") / "core" / "modbus_entity_parameters.hpp"
+    header_path.parent.mkdir(parents=True, exist_ok=True)
+    
+    with open(header_path, 'w') as f:
+        f.write(header_content)
+    
+    print(f"Created dummy header file: {header_path} ({os.path.getsize(header_path)} bytes)")
+
+if not (env.IsCleanTarget()):
+    create_dummy_modbus_header()

scripts/build_modbus_generate_doc_post.py

@@ -0,0 +1,64 @@
+import subprocess
+import os
+import sys
+import shutil
+from pathlib import Path
+
+# Import the streaming function from the separate module
+from run_executable import run_with_streaming_input
+
+def get_python_executable():
+    """Get the appropriate Python executable for the current platform."""
+    # Try different Python executable names
+    python_names = ['python3', 'python', 'py']
+    
+    for name in python_names:
+        if shutil.which(name):
+            return name
+    
+    # Fallback to sys.executable if available
+    return sys.executable
+
+
+def csv_to_md(csv_file_path, output_file_path, script_path):
+
+    # Ensure the output directory exists
+    Path(output_file_path).parent.mkdir(parents=True, exist_ok=True)
+    
+    # delete the output file if it exists
+    if os.path.exists(output_file_path):
+        os.remove(output_file_path)
+    
+    # Read CSV file and pipe to Python script to generate header
+    python_exe = get_python_executable()
+    
+    with open(csv_file_path, 'r') as csv_file:
+        with open(output_file_path, 'w') as output_file:
+            subprocess.run(
+                [python_exe, script_path],
+                stdin=csv_file,
+                stdout=output_file,
+                check=True
+            )
+    
+    print(f"Generated MD file: {output_file_path} ({os.path.getsize(output_file_path)} bytes)")
+
+
+def main(program_path="./emsesp"):
+    csv_file = os.path.join("docs", "dump_entities.csv")
+    output_file = os.path.join("docs", "Modbus-Entity-Registers.md")
+    script_file = os.path.join("scripts", "generate-modbus-register-doc.py")
+    
+    # generate the MD file
+    csv_to_md(csv_file, output_file, script_file)
+    
+    # run the test command and generate the dump_telegrams.csv file
+    test_command = "test telegram_dump"
+    telegram_output_file = os.path.join("docs", "dump_telegrams.csv")
+    print(f"Running test command: telegram_dump > {telegram_output_file}")
+    run_with_streaming_input(program_path, test_command, telegram_output_file)
+
+if __name__ == "__main__":
+    # Get program path from command line argument or use default
+    program_path = sys.argv[1] if len(sys.argv) > 1 else "./emsesp"
+    main(program_path)

scripts/build_run_test.py

@@ -0,0 +1,46 @@
+import os
+import shutil
+import subprocess
+import sys
+Import("env")
+
+# Import the streaming function from the separate module
+from run_executable import run_with_streaming_input
+
+def get_python_executable():
+    """Get the appropriate Python executable for the current platform."""
+    # Try different Python executable names
+    python_names = ['python3', 'python', 'py']
+    
+    for name in python_names:
+        if shutil.which(name):
+            return name
+    
+    # Fallback to sys.executable if available
+    return sys.executable
+
+
+def build_run_test(source, target, env):
+    
+     # Get the executable path
+    program_path = source[0].get_abspath()
+
+    # Get output file and test command from environment variable or use defaults
+    output_file = os.path.join("docs", env.GetProjectOption("custom_output_file", "dump_default_output.txt"))
+    test_command = env.GetProjectOption("custom_test_command", "test entity_dump")
+
+    # run the test command and save the output to the output file
+    run_with_streaming_input(program_path, test_command, output_file)
+    
+    # if we have a post command defined run it
+    post_script = env.GetProjectOption("custom_post_script", None)    
+    if post_script:
+        print(f"Running post script: {post_script}")
+        python_exe = get_python_executable()
+        subprocess.run([python_exe, post_script, program_path], check=True)
+ 
+env.AddCustomTarget(
+    "build",
+    "$BUILD_DIR/${PROGNAME}$PROGSUFFIX",
+    build_run_test
+)

scripts/force_clean.py

@@ -0,0 +1,14 @@
+
+Import("env")
+import os
+import shutil
+
+def force_clean(source, target, env):
+    """Remove build directory before building"""
+    build_dir = env.subst("$BUILD_DIR")
+    if os.path.exists(build_dir):
+        print(f"Force cleaning: {build_dir}")
+        shutil.rmtree(build_dir)
+
+# Register the callback to run before building
+env.AddPreAction("$BUILD_DIR/${PROGNAME}$PROGSUFFIX", force_clean)

scripts/run_executable.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+
+"""
+Utility functions for running executables with streaming input and CSV output extraction.
+"""
+
+import subprocess
+import sys
+import os
+from pathlib import Path
+
+
+def run_with_streaming_input(program_path, test_command, output_file=None):
+    """
+    Run the executable and stream text input to it.
+    
+    Args:
+        program_path (str): Path to the executable to run
+        test_command (str): Command to send to the executable
+        output_file (str, optional): Path to save CSV output. If None, no file is saved.
+    
+    Returns:
+        int: Return code of the executed process
+    """
+    try:
+        # Start the process with pipes for stdin, stdout, and stderr
+        process = subprocess.Popen(
+            [str(program_path)],
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=1  # Line buffered
+        )
+        
+        # add "test " to test_command if it doesn't already start with "test "
+        if not test_command.startswith("test "):
+            test_command = "test " + test_command
+
+        # Stream input line by line
+        for line in test_command.strip().split('\n'):
+            process.stdin.write(line + '\n')
+            process.stdin.flush()
+
+        # Close stdin to signal end of input
+        process.stdin.close()
+
+        # Read and collect output between CSV START and CSV END, then export to file
+        in_cvs_section = False
+        csv_output = []
+        
+        for line in process.stdout:
+            if "---- CSV START ----" in line:
+                in_cvs_section = True
+                continue
+            elif "---- CSV END ----" in line:
+                in_cvs_section = False
+                continue
+            elif in_cvs_section:
+                csv_output.append(line)
+                # print(line, end='')
+        
+        # Export CSV output to file if output_file is specified
+        if output_file:
+            # Ensure the output directory exists
+            Path(output_file).parent.mkdir(parents=True, exist_ok=True)
+            
+            # delete the output file if it exists
+            if os.path.exists(output_file):
+                os.remove(output_file)
+            
+            # Export CSV output to file
+            with open(output_file, 'w') as f:
+                f.writelines(csv_output)
+            print(f"CSV file created: {output_file} ({os.path.getsize(output_file)} bytes)")
+
+        # Wait for process to complete
+        return_code = process.wait()
+
+        # Print any errors
+        stderr_output = process.stderr.read()
+        if stderr_output:
+            print("\nErrors:", file=sys.stderr)
+            print(stderr_output, file=sys.stderr)
+
+        return return_code
+
+    except Exception as e:
+        print(f"Error running executable: {e}", file=sys.stderr)
+        return 1
+
+
+def run_executable_with_command(program_path, command, output_file=None):
+    """
+    Simplified interface to run an executable with a command and optionally save output.
+    
+    Args:
+        program_path (str): Path to the executable to run
+        command (str): Command to send to the executable
+        output_file (str, optional): Path to save CSV output. If None, no file is saved.
+    
+    Returns:
+        int: Return code of the executed process
+    """
+    return run_with_streaming_input(program_path, command, output_file)
+
+
+def main():
+    """Command-line interface for running executables with streaming input."""
+    if len(sys.argv) < 3:
+        print("Usage: python3 run_executable.py <program_path> <command> [output_file]")
+        print("Example: python3 run_executable.py ./emsesp entity_dump ./output.csv")
+        sys.exit(1)
+    
+    program_path = sys.argv[1]
+    command = sys.argv[2]
+    output_file = sys.argv[3] if len(sys.argv) > 3 else None
+    
+    print(f"Running: {program_path}")
+    print(f"Command: {command}")
+    if output_file:
+        print(f"Output file: {output_file}")
+    
+    return_code = run_with_streaming_input(program_path, command, output_file)
+    
+    if return_code == 0:
+        print("Execution completed successfully!")
+    else:
+        print(f"Execution failed with return code: {return_code}")
+    
+    sys.exit(return_code)
+
+
+if __name__ == "__main__":
+    main()

scripts/update_all.sh

@@ -26,4 +26,3 @@ pnpm webUI
 cd ..
 npx cspell "**"
 
-sh ./scripts/generate_csv_and_headers.sh