Analyzing and summarizing batch process results is a common task in automation, especially when you need to evaluate performance or quality across multiple runs. In this challenge, you will work with a hardcoded pandas DataFrame representing results from several batches. Your goal is to group the data by batch ID, calculate the mean and standard deviation for each batch, and present the findings in a summary DataFrame. This mirrors real-world scenarios where automation engineers need to quickly assess variability and trends across repeated processes.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import pandas as pd
import numpy as np

class TestTask(unittest.TestCase):
    def setUp(self):
        self.df = pd.DataFrame({
            "batch_id": [1, 1, 2, 2, 2, 3, 3, 3, 3],
            "result": [10.0, 12.0, 9.5, 9.7, 9.8, 11.2, 11.3, 11.4, 11.1]
        })
        self.df_single = pd.DataFrame({
            "batch_id": [1, 2, 3],
            "result": [5, 10, 15]
        })

    def test_grouping_by_batch_id(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'summarize_batches', None)
        _dynamic_test(
            self,
            callable(func),
            "Function summarize_batches is defined.",
            "Function summarize_batches is not defined."
        )
        summary = func(self.df)
        _dynamic_test(
            self,
            isinstance(summary, pd.DataFrame),
            "Returned value is a DataFrame.",
            f"Expected DataFrame, got {type(summary)}."
        )
        _dynamic_test(
            self,
            set(summary.index) == {1, 2, 3},
            "DataFrame is grouped by batch_id.",
            f"Expected index {{1,2,3}}, got {set(summary.index)}."
        )

    def test_mean_calculation(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'summarize_batches', None)
        summary = func(self.df)
        expected_means = self.df.groupby("batch_id")["result"].mean()
        for batch_id in expected_means.index:
            actual = summary.loc[batch_id, "mean_result"] if batch_id in summary.index else None
            expected = expected_means.loc[batch_id]
            _dynamic_test(
                self,
                np.isclose(actual, expected),
                f"Mean for batch {batch_id} is correct.",
                f"Expected mean {expected} for batch {batch_id}, got {actual}."
            )

    def test_std_calculation(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'summarize_batches', None)
        summary = func(self.df)
        expected_stds = self.df.groupby("batch_id")["result"].std()
        for batch_id in expected_stds.index:
            actual = summary.loc[batch_id, "std_result"] if batch_id in summary.index else None
            expected = expected_stds.loc[batch_id]
            _dynamic_test(
                self,
                np.isclose(actual, expected, equal_nan=True),
                f"Standard deviation for batch {batch_id} is correct.",
                f"Expected std {expected} for batch {batch_id}, got {actual}."
            )

    def test_column_names_and_index(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'summarize_batches', None)
        summary = func(self.df)
        _dynamic_test(
            self,
            list(summary.columns) == ["mean_result", "std_result"],
            "Returned DataFrame has correct column names.",
            f"Expected columns ['mean_result', 'std_result'], got {list(summary.columns)}."
        )
        _dynamic_test(
            self,
            summary.index.name in ["batch_id", None],
            "DataFrame index is batch_id or unnamed (default).",
            f"Expected index name 'batch_id' or None, got {summary.index.name}."
        )

    def test_single_row_batch(self):
        import user_code
        importlib.reload(user_code)
        func = getattr(user_code, 'summarize_batches', None)
        summary = func(self.df_single)
        _dynamic_test(
            self,
            all(np.isnan(summary["std_result"])),
            "Standard deviation is NaN for single-value batches.",
            f"Expected NaN std_result for single row batches, got {summary['std_result'].values}."
        )
        expected_means = self.df_single.groupby("batch_id")["result"].mean()
        for batch_id in expected_means.index:
            actual = summary.loc[batch_id, "mean_result"] if batch_id in summary.index else None
            expected = expected_means.loc[batch_id]
            _dynamic_test(
                self,
                np.isclose(actual, expected),
                f"Mean for batch {batch_id} with single value is correct.",
                f"Expected mean {expected} for batch {batch_id}, got {actual}."
            )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    # Collect all assignment nodes to modify
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    # If nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

if __name__ == "__main__":
    unittest.main()


test_main.py

Master Python as a powerful tool for automation engineering. This course guides you through practical and advanced Python techniques to automate repetitive tasks, analyze data, and interact with systems, all tailored for automation engineers. Each section blends engaging theory with hands-on, real-world challenges.

Learn how Python can automate repetitive and time-consuming tasks commonly faced by automation engineers, from log parsing to report generation.

Dive into data manipulation and analysis using Python and pandas, tailored for automation engineers handling structured data.

Master Python techniques for interacting with systems, automating tests, and analyzing results—core skills for automation engineers.

Challenge: Aggregate Batch Results

解答