如何分组交通灯连续相同状态序列统计相位时长？已尝试Counter

Got it, let's break this down. The problem with using collections.Counter here is that it only counts total occurrences of each traffic light phase, not consecutive runs of the same state—which is exactly what you need to analyze timing patterns.

Here are two straightforward solutions to group consecutive identical phases, along with code to analyze their timing:

1. Use itertools.groupby (Python's built-in tool for consecutive grouping) #

itertools.groupby is perfect for this scenario because it groups consecutive elements with the same key. Since you're storing one phase per second in your deque, the length of each group directly translates to the duration (in seconds) that phase was active.

Here's how to integrate it with your existing code:

from itertools import groupby
from collections import deque
from enum import Enum
import time
import threading

# Your existing TrafficLight enum
class TrafficLightPhase(Enum):
    RED = "RED"
    YELLOW = "YELLOW"
    GREEN = "GREEN"

# Replace this with your real function to fetch the current traffic light phase
def get_current_phase_phases():
    # Simulating a traffic light cycle for demo purposes
    phase_cycle = [TrafficLightPhase.RED]*5 + [TrafficLightPhase.GREEN]*10 + [TrafficLightPhase.YELLOW]*2
    while True:
        for phase in phase_cycle:
            yield phase

# Global state for polling control and phase storage
running = True
last_phases = deque()

# Your existing polling function
def read_phases():
    phase_source = get_current_phase_phases()
    while running:
        current_phase = next(phase_source)
        last_phases.append(current_phase)
        time.sleep(1)

# Function to group consecutive phases and calculate their durations
def group_consecutive_phases(phase_queue):
    consecutive_runs = []
    # Group consecutive identical phases
    for phase, consecutive_group in groupby(phase_queue):
        # Each element in the group represents 1 second of the same phase
        duration = len(list(consecutive_group))
        consecutive_runs.append((phase, duration))
    return consecutive_runs

# Example workflow: Run polling, then analyze results
if __name__ == "__main__":
    # Start polling in a separate thread to avoid blocking
    poll_thread = threading.Thread(target=read_phases)
    poll_thread.start()
    
    # Let the polling run for 30 seconds
    time.sleep(30)
    running = False
    poll_thread.join()
    
    # Get the grouped consecutive phase runs
    phase_runs = group_consecutive_phases(last_phases)
    print("Consecutive Phase Runs:")
    for phase, duration in phase_runs:
        print(f"{phase.name}: {duration} seconds")

2. Manual Grouping (if you prefer not to use itertools) #

If you want to avoid relying on itertools, you can manually iterate through the deque and track consecutive phases:

def group_consecutive_phases_manual(phase_queue):
    if not phase_queue:
        return []
    
    consecutive_runs = []
    current_phase = phase_queue[0]
    current_duration = 1
    
    # Iterate through the rest of the queue
    for phase in list(phase_queue)[1:]:
        if phase == current_phase:
            current_duration += 1
        else:
            # Save the completed run of the previous phase
            consecutive_runs.append((current_phase, current_duration))
            current_phase = phase
            current_duration = 1
    
    # Don't forget to add the last run of phases
    consecutive_runs.append((current_phase, current_duration))
    return consecutive_runs

Analyzing Timing Patterns #

Once you have the consecutive runs, you can calculate useful stats like average duration per phase, total time spent in each phase, etc. Here's a quick way to do that:

from collections import defaultdict

def analyze_timing_patterns(consecutive_runs):
    phase_stats = defaultdict(lambda: {"total_time": 0, "run_count": 0})
    
    for phase, duration in consecutive_runs:
        phase_stats[phase]["total_time"] += duration
        phase_stats[phase]["run_count"] += 1
    
    # Calculate averages and format the results
    timing_patterns = {}
    for phase, stats in phase_stats.items():
        avg_duration = stats["total_time"] / stats["run_count"]
        timing_patterns[phase] = {
            "average_duration": round(avg_duration, 2),
            "total_time": stats["total_time"],
            "number_of_runs": stats["run_count"]
        }
    return timing_patterns

# Example usage
timing_stats = analyze_timing_patterns(phase_runs)
print("\nTiming Patterns:")
for phase, stats in timing_stats.items():
    print(f"{phase.name}:")
    print(f"  Average Duration: {stats['average_duration']} seconds")
    print(f"  Total Time Active: {stats['total_time']} seconds")
    print(f"  Number of Runs: {stats['number_of_runs']}")

Why Counter Didn't Work #

Just to clarify: Counter counts how many times each phase appears in total (e.g., "RED appeared 15 times"), but it doesn't track if those 15 times were in a single 15-second run or 3 separate 5-second runs. That's why we need to group consecutive phases first before doing any meaningful timing analysis.

内容的提问来源于stack exchange，提问作者logoff