In engineering systems, sensor data is crucial for monitoring equipment and ensuring reliable operation. However, sensors can sometimes fail or provide inaccurate readings, which may lead to incorrect assessments or even system failures. Clustering algorithms, such as **KMeans**, are powerful tools for automatically grouping similar data points and identifying anomalies. By clustering sensor readings, you can detect groups of normal behavior and flag data points that deviate significantly from their cluster centers. These outliers are often indicative of faulty sensors or abnormal system conditions. In this challenge, you will use **KMeans clustering** to separate normal and abnormal temperature readings from multiple sensors and identify which sensors are likely malfunctioning based on their distance from the cluster centers.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import importlib
import sys
import numpy as np

class TestTask(unittest.TestCase):
    def test_output_lengths(self):
        import user_code
        importlib.reload(user_code)
        readings = [22.1, 22.3, 22.4, 22.2, 45.7, 22.0, 22.5, 22.3]
        result = user_code.detect_faulty_sensors(readings)
        _dynamic_test(
            self,
            result is not None and isinstance(result, tuple) and len(result) == 2,
            "Function returns a tuple of two elements",
            f"Function should return a tuple of two elements, got {result}",
        )
        cluster_labels, faulty_indices = result
        _dynamic_test(
            self,
            isinstance(cluster_labels, list) and len(cluster_labels) == len(readings),
            "Cluster assignments length matches input readings",
            f"Expected cluster assignments of length {len(readings)}, got {len(cluster_labels)}",
        )
        _dynamic_test(
            self,
            isinstance(faulty_indices, list),
            "Faulty indices is a list",
            f"Expected faulty_indices to be a list, got {type(faulty_indices)}",
        )

    def test_single_outlier(self):
        import user_code
        importlib.reload(user_code)
        readings = [20.0, 20.1, 20.2, 50.0]
        _, faulty_indices = user_code.detect_faulty_sensors(readings)
        _dynamic_test(
            self,
            faulty_indices == [3],
            "Correctly identifies single outlier (index 3) as faulty",
            f"Expected [3] as faulty index, got {faulty_indices}",
        )

    def test_multiple_outliers(self):
        import user_code
        importlib.reload(user_code)
        readings = [10.0, 10.0, 100.0, 100.0]
        _, faulty_indices = user_code.detect_faulty_sensors(readings)
        # Both 100.0 readings are equally farthest from their cluster center
        _dynamic_test(
            self,
            set(faulty_indices) == {0, 1},
            "Correctly identifies both outlier indices as faulty",
            f"Expected [2, 3] (order doesn't matter), got {faulty_indices}",
        )

    def test_varied_length(self):
        import user_code
        importlib.reload(user_code)
        readings = [19.5, 19.8, 20.0, 19.7, 30.0]
        labels, faulty_indices = user_code.detect_faulty_sensors(readings)
        _dynamic_test(
            self,
            isinstance(labels, list) and len(labels) == len(readings),
            "Works for input lists of different lengths",
            f"Expected cluster assignments of length {len(readings)}, got {labels}",
        )
        _dynamic_test(
            self,
            faulty_indices == [4],
            "Correctly identifies index 4 as faulty",
            f"Expected [4] as faulty index, got {faulty_indices}",
        )

    def test_all_equal(self):
        import user_code
        importlib.reload(user_code)
        readings = [15.0, 15.0, 15.0, 15.0]
        _, faulty_indices = user_code.detect_faulty_sensors(readings)
        # All distances are zero, so all indices are equally far (all zero)
        _dynamic_test(
            self,
            set(faulty_indices) == {0,1,2,3},
            "Handles case where all readings are identical (all indices returned)",
            f"Expected all indices [0,1,2,3], got {faulty_indices}",
        )

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

A practical course designed for engineers who want to leverage Python to solve real-world engineering problems. Explore data analysis, mathematical modeling, and simulation using Python's powerful libraries and tools, all through engaging, hands-on examples and challenges.

Learn how to use Python to analyze, summarize, and visualize engineering data. This section covers essential data handling and visualization techniques tailored for engineering applications.

Dive into mathematical modeling and simulation techniques for engineering systems using Python. Learn to model physical processes, solve equations, and simulate system behavior.

Apply data science techniques to engineering problems, including classification, clustering, and predictive modeling using Python's scikit-learn and seaborn libraries.

Challenge: Detect Faulty Sensors with Clustering

解答