style: format with ruff
This commit is contained in:
+114
-90
@@ -58,10 +58,16 @@ class ImageStatsReceiver:
|
||||
x_coords = np.arange(len(x_projection))
|
||||
y_coords = np.arange(len(y_projection))
|
||||
|
||||
x_centroid = np.sum(x_coords * x_projection) / np.sum(x_projection) if np.sum(x_projection) > 0 else len(
|
||||
x_projection) / 2
|
||||
y_centroid = np.sum(y_coords * y_projection) / np.sum(y_projection) if np.sum(y_projection) > 0 else len(
|
||||
y_projection) / 2
|
||||
x_centroid = (
|
||||
np.sum(x_coords * x_projection) / np.sum(x_projection)
|
||||
if np.sum(x_projection) > 0
|
||||
else len(x_projection) / 2
|
||||
)
|
||||
y_centroid = (
|
||||
np.sum(y_coords * y_projection) / np.sum(y_projection)
|
||||
if np.sum(y_projection) > 0
|
||||
else len(y_projection) / 2
|
||||
)
|
||||
|
||||
# Calculate FWHM (Full Width at Half Maximum) approximation
|
||||
def calculate_fwhm(profile):
|
||||
@@ -76,20 +82,20 @@ class ImageStatsReceiver:
|
||||
y_fwhm = calculate_fwhm(y_projection)
|
||||
|
||||
return {
|
||||
'x_projection': x_projection.tolist(),
|
||||
'y_projection': y_projection.tolist(),
|
||||
'x_projection_max': x_projection_max.tolist(),
|
||||
'y_projection_max': y_projection_max.tolist(),
|
||||
'x_peak_position': int(x_peak_idx),
|
||||
'y_peak_position': int(y_peak_idx),
|
||||
'x_peak_value': float(x_peak_value),
|
||||
'y_peak_value': float(y_peak_value),
|
||||
'x_centroid': float(x_centroid),
|
||||
'y_centroid': float(y_centroid),
|
||||
'x_fwhm': float(x_fwhm),
|
||||
'y_fwhm': float(y_fwhm),
|
||||
'x_coords': list(range(len(x_projection))),
|
||||
'y_coords': list(range(len(y_projection)))
|
||||
"x_projection": x_projection.tolist(),
|
||||
"y_projection": y_projection.tolist(),
|
||||
"x_projection_max": x_projection_max.tolist(),
|
||||
"y_projection_max": y_projection_max.tolist(),
|
||||
"x_peak_position": int(x_peak_idx),
|
||||
"y_peak_position": int(y_peak_idx),
|
||||
"x_peak_value": float(x_peak_value),
|
||||
"y_peak_value": float(y_peak_value),
|
||||
"x_centroid": float(x_centroid),
|
||||
"y_centroid": float(y_centroid),
|
||||
"x_fwhm": float(x_fwhm),
|
||||
"y_fwhm": float(y_fwhm),
|
||||
"x_coords": list(range(len(x_projection))),
|
||||
"y_coords": list(range(len(y_projection))),
|
||||
}
|
||||
|
||||
def calculate_radial_integration(self, image: np.ndarray, num_bins: int = 50) -> Dict[str, Any]:
|
||||
@@ -104,7 +110,7 @@ class ImageStatsReceiver:
|
||||
center_y, center_x = np.array(gray_image.shape) // 2
|
||||
|
||||
# Create coordinate grids
|
||||
y_coords, x_coords = np.ogrid[:gray_image.shape[0], :gray_image.shape[1]]
|
||||
y_coords, x_coords = np.ogrid[: gray_image.shape[0], : gray_image.shape[1]]
|
||||
|
||||
# Calculate distance from center for each pixel
|
||||
distances = np.sqrt((x_coords - center_x) ** 2 + (y_coords - center_y) ** 2)
|
||||
@@ -137,13 +143,13 @@ class ImageStatsReceiver:
|
||||
pixel_counts.append(0)
|
||||
|
||||
return {
|
||||
'r_centers': r_centers.tolist(),
|
||||
'radial_profile': radial_profile,
|
||||
'radial_std': radial_std,
|
||||
'pixel_counts': pixel_counts,
|
||||
'max_radius': float(max_radius),
|
||||
'center': [int(center_x), int(center_y)],
|
||||
'num_bins': num_bins
|
||||
"r_centers": r_centers.tolist(),
|
||||
"radial_profile": radial_profile,
|
||||
"radial_std": radial_std,
|
||||
"pixel_counts": pixel_counts,
|
||||
"max_radius": float(max_radius),
|
||||
"center": [int(center_x), int(center_y)],
|
||||
"num_bins": num_bins,
|
||||
}
|
||||
|
||||
def calculate_image_stats(self, image: np.ndarray) -> Dict[str, Any]:
|
||||
@@ -151,30 +157,30 @@ class ImageStatsReceiver:
|
||||
image_float = image.astype(np.float64)
|
||||
|
||||
stats = {
|
||||
'timestamp': time.time(),
|
||||
'shape': image.shape,
|
||||
'dtype': str(image.dtype),
|
||||
'mean': float(np.mean(image_float)),
|
||||
'std': float(np.std(image_float)),
|
||||
'median': float(np.median(image_float)),
|
||||
'min': float(np.min(image_float)),
|
||||
'max': float(np.max(image_float)),
|
||||
'message_count': self.message_count
|
||||
"timestamp": time.time(),
|
||||
"shape": image.shape,
|
||||
"dtype": str(image.dtype),
|
||||
"mean": float(np.mean(image_float)),
|
||||
"std": float(np.std(image_float)),
|
||||
"median": float(np.median(image_float)),
|
||||
"min": float(np.min(image_float)),
|
||||
"max": float(np.max(image_float)),
|
||||
"message_count": self.message_count,
|
||||
}
|
||||
|
||||
# Calculate per-channel stats if color image
|
||||
if len(image.shape) == 3 and image.shape[2] > 1:
|
||||
for channel in range(image.shape[2]):
|
||||
channel_data = image_float[:, :, channel]
|
||||
stats[f'mean_ch{channel}'] = float(np.mean(channel_data))
|
||||
stats[f'std_ch{channel}'] = float(np.std(channel_data))
|
||||
stats[f'median_ch{channel}'] = float(np.median(channel_data))
|
||||
stats[f"mean_ch{channel}"] = float(np.mean(channel_data))
|
||||
stats[f"std_ch{channel}"] = float(np.std(channel_data))
|
||||
stats[f"median_ch{channel}"] = float(np.median(channel_data))
|
||||
|
||||
radial_data = self.calculate_radial_integration(image)
|
||||
stats['radial'] = radial_data
|
||||
stats["radial"] = radial_data
|
||||
|
||||
#projection_data = self.calculate_projections(image)
|
||||
#stats['projections'] = projection_data
|
||||
# projection_data = self.calculate_projections(image)
|
||||
# stats['projections'] = projection_data
|
||||
|
||||
return stats
|
||||
|
||||
@@ -200,7 +206,7 @@ class ImageStatsReceiver:
|
||||
rgb_image = rgb_image[:, ::-1, :].copy()
|
||||
|
||||
stats = self.calculate_image_stats(rgb_image)
|
||||
stats['image_type'] = 'rgb_from_bayer'
|
||||
stats["image_type"] = "rgb_from_bayer"
|
||||
|
||||
with self.stats_lock:
|
||||
self.latest_stats = stats
|
||||
@@ -213,7 +219,7 @@ class ImageStatsReceiver:
|
||||
rgb_image = rgb_image.reshape(header_shape)
|
||||
|
||||
stats = self.calculate_image_stats(rgb_image)
|
||||
stats['image_type'] = 'rgb_direct'
|
||||
stats["image_type"] = "rgb_direct"
|
||||
|
||||
with self.stats_lock:
|
||||
self.latest_stats = stats
|
||||
@@ -245,45 +251,53 @@ class ImageStatsReceiver:
|
||||
|
||||
def print_formatted_stats(self, stats: Dict[str, Any]):
|
||||
"""Print formatted statistics in a compact terminal format"""
|
||||
timestamp = time.strftime('%H:%M:%S', time.localtime(stats['timestamp']))
|
||||
timestamp = time.strftime("%H:%M:%S", time.localtime(stats["timestamp"]))
|
||||
|
||||
# Compact one-line format
|
||||
print(f"[{timestamp}] #{stats['message_count']:4d} | "
|
||||
f"Shape: {stats['shape']} | "
|
||||
f"Mean: {stats['mean']:6.1f} | "
|
||||
f"Std: {stats['std']:6.1f} | "
|
||||
f"Median: {stats['median']:6.1f} | "
|
||||
f"Range: [{stats['min']:3.0f}-{stats['max']:3.0f}]",
|
||||
flush=True)
|
||||
print(
|
||||
f"[{timestamp}] #{stats['message_count']:4d} | "
|
||||
f"Shape: {stats['shape']} | "
|
||||
f"Mean: {stats['mean']:6.1f} | "
|
||||
f"Std: {stats['std']:6.1f} | "
|
||||
f"Median: {stats['median']:6.1f} | "
|
||||
f"Range: [{stats['min']:3.0f}-{stats['max']:3.0f}]",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
if 'radial' in stats:
|
||||
radial = stats['radial']
|
||||
center_intensity = radial['radial_profile'][0] if radial['radial_profile'] else 0
|
||||
edge_intensity = radial['radial_profile'][-1] if radial['radial_profile'] else 0
|
||||
peak_radius_idx = np.argmax(radial['radial_profile']) if radial['radial_profile'] else 0
|
||||
peak_radius = radial['r_centers'][peak_radius_idx] if radial['r_centers'] else 0
|
||||
if "radial" in stats:
|
||||
radial = stats["radial"]
|
||||
center_intensity = radial["radial_profile"][0] if radial["radial_profile"] else 0
|
||||
edge_intensity = radial["radial_profile"][-1] if radial["radial_profile"] else 0
|
||||
peak_radius_idx = np.argmax(radial["radial_profile"]) if radial["radial_profile"] else 0
|
||||
peak_radius = radial["r_centers"][peak_radius_idx] if radial["r_centers"] else 0
|
||||
|
||||
print(f"{'':21} Radial: Center={center_intensity:.1f} | "
|
||||
f"Edge={edge_intensity:.1f} | "
|
||||
f"Peak@r={peak_radius:.1f} | "
|
||||
f"Center=({radial['center'][0]},{radial['center'][1]})",
|
||||
flush=True)
|
||||
print(
|
||||
f"{'':21} Radial: Center={center_intensity:.1f} | "
|
||||
f"Edge={edge_intensity:.1f} | "
|
||||
f"Peak@r={peak_radius:.1f} | "
|
||||
f"Center=({radial['center'][0]},{radial['center'][1]})",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
# Print projection summary
|
||||
if 'projections' in stats:
|
||||
proj = stats['projections']
|
||||
print(f"{'':21} X-Profile: Peak@{proj['x_peak_position']}({proj['x_peak_value']:.1f}) | "
|
||||
f"Centroid={proj['x_centroid']:.1f} | FWHM={proj['x_fwhm']:.1f}",
|
||||
flush=True)
|
||||
print(f"{'':21} Y-Profile: Peak@{proj['y_peak_position']}({proj['y_peak_value']:.1f}) | "
|
||||
f"Centroid={proj['y_centroid']:.1f} | FWHM={proj['y_fwhm']:.1f}",
|
||||
flush=True)
|
||||
if "projections" in stats:
|
||||
proj = stats["projections"]
|
||||
print(
|
||||
f"{'':21} X-Profile: Peak@{proj['x_peak_position']}({proj['x_peak_value']:.1f}) | "
|
||||
f"Centroid={proj['x_centroid']:.1f} | FWHM={proj['x_fwhm']:.1f}",
|
||||
flush=True,
|
||||
)
|
||||
print(
|
||||
f"{'':21} Y-Profile: Peak@{proj['y_peak_position']}({proj['y_peak_value']:.1f}) | "
|
||||
f"Centroid={proj['y_centroid']:.1f} | FWHM={proj['y_fwhm']:.1f}",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
# Optional: Print per-channel stats if available
|
||||
if any(k.startswith('mean_ch') for k in stats.keys()):
|
||||
if any(k.startswith("mean_ch") for k in stats.keys()):
|
||||
channels = []
|
||||
i = 0
|
||||
while f'mean_ch{i}' in stats:
|
||||
while f"mean_ch{i}" in stats:
|
||||
channels.append(f"Ch{i}({stats[f'mean_ch{i}']:.1f})")
|
||||
i += 1
|
||||
if channels:
|
||||
@@ -292,16 +306,23 @@ class ImageStatsReceiver:
|
||||
def get_radial_profile_summary(self):
|
||||
"""Get a summary of the current radial profile"""
|
||||
with self.stats_lock:
|
||||
if self.latest_stats and 'radial' in self.latest_stats:
|
||||
radial = self.latest_stats['radial']
|
||||
if self.latest_stats and "radial" in self.latest_stats:
|
||||
radial = self.latest_stats["radial"]
|
||||
return {
|
||||
'r_centers': radial['r_centers'],
|
||||
'radial_profile': radial['radial_profile'],
|
||||
'center_intensity': radial['radial_profile'][0] if radial['radial_profile'] else 0,
|
||||
'edge_intensity': radial['radial_profile'][-1] if radial['radial_profile'] else 0,
|
||||
'max_intensity_radius': radial['r_centers'][np.argmax(radial['radial_profile'])] if radial[
|
||||
'radial_profile'] else 0,
|
||||
'max_intensity_value': max(radial['radial_profile']) if radial['radial_profile'] else 0
|
||||
"r_centers": radial["r_centers"],
|
||||
"radial_profile": radial["radial_profile"],
|
||||
"center_intensity": radial["radial_profile"][0]
|
||||
if radial["radial_profile"]
|
||||
else 0,
|
||||
"edge_intensity": radial["radial_profile"][-1]
|
||||
if radial["radial_profile"]
|
||||
else 0,
|
||||
"max_intensity_radius": radial["r_centers"][np.argmax(radial["radial_profile"])]
|
||||
if radial["radial_profile"]
|
||||
else 0,
|
||||
"max_intensity_value": max(radial["radial_profile"])
|
||||
if radial["radial_profile"]
|
||||
else 0,
|
||||
}
|
||||
return None
|
||||
|
||||
@@ -311,21 +332,23 @@ class ImageStatsReceiver:
|
||||
filename = f"radial_profile_{int(time.time())}.txt"
|
||||
|
||||
with self.stats_lock:
|
||||
if self.latest_stats and 'radial' in self.latest_stats:
|
||||
radial = self.latest_stats['radial']
|
||||
if self.latest_stats and "radial" in self.latest_stats:
|
||||
radial = self.latest_stats["radial"]
|
||||
|
||||
with open(filename, 'w') as f:
|
||||
with open(filename, "w") as f:
|
||||
f.write("# Radial Integration Profile\n")
|
||||
f.write(f"# Timestamp: {time.ctime(self.latest_stats['timestamp'])}\n")
|
||||
f.write(f"# Image shape: {self.latest_stats['shape']}\n")
|
||||
f.write(f"# Center: {radial['center']}\n")
|
||||
f.write("# Radius(pixels)\tMean_Intensity\tStd_Intensity\tPixel_Count\n")
|
||||
|
||||
for i in range(len(radial['r_centers'])):
|
||||
f.write(f"{radial['r_centers'][i]:.2f}\t"
|
||||
f"{radial['radial_profile'][i]:.2f}\t"
|
||||
f"{radial['radial_std'][i]:.2f}\t"
|
||||
f"{radial['pixel_counts'][i]}\n")
|
||||
for i in range(len(radial["r_centers"])):
|
||||
f.write(
|
||||
f"{radial['r_centers'][i]:.2f}\t"
|
||||
f"{radial['radial_profile'][i]:.2f}\t"
|
||||
f"{radial['radial_std'][i]:.2f}\t"
|
||||
f"{radial['pixel_counts'][i]}\n"
|
||||
)
|
||||
|
||||
print(f"Radial profile saved to {filename}")
|
||||
return filename
|
||||
@@ -419,6 +442,7 @@ def save_current_radial_profile(filename: str = None):
|
||||
return stats_receiver.save_radial_profile_to_file(filename)
|
||||
return None
|
||||
|
||||
|
||||
# # Auto-start when daq.py is imported/run
|
||||
# if __name__ == "__main__":
|
||||
# # If running daq.py directly
|
||||
@@ -444,4 +468,4 @@ def save_current_radial_profile(filename: str = None):
|
||||
#
|
||||
# else:
|
||||
# # If daq.py is imported as a module, auto-start the receiver
|
||||
# start_image_stats_receiver()
|
||||
# start_image_stats_receiver()
|
||||
|
||||
Reference in New Issue
Block a user