Merge pull request #16 from QuentinWach/codex/improve-topology-optimization-in-4_topo.py

Remove generated topology optimization images

Author: QuentinWach

beamz/devices/monitors.py

@@ -1,17 +1,24 @@
+from typing import Callable, Optional
+
 import numpy as np
 import matplotlib.pyplot as plt
 from matplotlib.patches import Rectangle as MatplotlibRectangle
 
-class Monitor():    
-    def __init__(self, design=None, start=(0,0), end=None, plane_normal=None, plane_position=0, 
-                 size=None, record_fields=True, accumulate_power=True, live_update=False, 
-                 record_interval=1, max_history_steps=None):
+class Monitor():
+    def __init__(self, design=None, start=(0,0), end=None, plane_normal=None, plane_position=0,
+                 size=None, record_fields=True, accumulate_power=True, live_update=False,
+                 record_interval=1, max_history_steps=None,
+                 objective_function: Optional[Callable[["Monitor"], float]] = None,
+                 name: Optional[str] = None):
         self.design = design
         self.should_record_fields = record_fields
         self.accumulate_power = accumulate_power
         self.live_update = live_update
         self.record_interval = record_interval
         self.max_history_steps = max_history_steps
+        self.objective_function = objective_function
+        self.objective_value: Optional[float] = None
+        self.name = name
 
         # Determine if this is a 3D monitor based on input parameters
         self.is_3d = self._determine_3d_mode(start, end, design)
@@ -44,8 +51,26 @@ def __init__(self, design=None, start=(0,0), end=None, plane_normal=None, plane_
 
         if self.is_3d: 
             self._init_3d_monitor(start, end, plane_normal, plane_position, size)
-        else: 
+        else:
             self._init_2d_monitor(start, end)
+
+    def evaluate_objective(self) -> Optional[float]:
+        """Evaluate the objective function associated with this monitor, if any."""
+        if self.objective_function is None:
+            return None
+        try:
+            value = self.objective_function(self)
+        except Exception as exc:
+            print(f"Warning: monitor objective evaluation failed: {exc}")
+            return None
+        if value is None:
+            return None
+        try:
+            self.objective_value = float(value)
+        except (TypeError, ValueError):
+            print(f"Warning: monitor objective returned non-numeric value: {value}")
+            return None
+        return self.objective_value
 
     def _determine_3d_mode(self, start, end, design):
         """Determine if this should be a 3D monitor based on inputs."""

beamz/simulation/fdtd.py

@@ -1,4 +1,4 @@
-from typing import Dict, Optional
+from typing import Dict, Optional, Sequence
 import datetime
 import numpy as np
 
@@ -309,12 +309,12 @@ def _update_3d_fields(self):
         Ey[1:-1, :, 1:-1] = factor_ey * Ey[1:-1, :, 1:-1] + source_ey * (curlH_y)
         Ez[:, 1:-1, 1:-1] = factor_ez * Ez[:, 1:-1, 1:-1] + source_ez * (curlH_z)
 
-    def initialize_simulation(self, save=True, live=True, axis_scale=[-1,1], save_animation=False, 
-                             animation_filename='fdtd_animation.mp4', clean_visualization=True, 
+    def initialize_simulation(self, save=True, live=True, axis_scale=[-1,1], save_animation=False,
+                             animation_filename='fdtd_animation.mp4', clean_visualization=True,
                              save_fields=None, decimate_save=1, accumulate_power=False,
-                             save_memory_mode=False):
+                             save_memory_mode=False, fields_to_cache: Optional[Sequence[str]] = None):
         """Initialize the simulation before running steps.
-        
+
         Args:
             save: Whether to save field data at each step.
             live: Whether to show live animation of the simulation.
@@ -325,14 +325,16 @@ def initialize_simulation(self, save=True, live=True, axis_scale=[-1,1], save_an
             decimate_save: Save only every nth time step (1 = save all, 10 = save every 10th step)
             accumulate_power: Instead of saving all fields, accumulate power and save that
             save_memory_mode: If True, avoid storing all field data and only keep monitors/power
+            fields_to_cache: Fields that should always be cached (complex values preserved)
         """
         # Set default save_fields based on dimensionality
         if save_fields is None:
             if self.is_3d:
                 save_fields = ['Ex', 'Ey', 'Ez', 'Hx', 'Hy', 'Hz']
             else:
                 save_fields = ['Ez', 'Hx', 'Hy']
-        
+        self._cache_fields = list(fields_to_cache) if fields_to_cache else []
+
         # Record start time
         self.start_time = datetime.datetime.now()
         # Initialize simulation state
@@ -344,11 +346,15 @@ def initialize_simulation(self, save=True, live=True, axis_scale=[-1,1], save_an
         self._decimate_save = decimate_save
         self._live = live
         self._axis_scale = axis_scale
-        
+
+        # Reset stored results for a new run
+        for key in list(self.results.keys()):
+            self.results[key] = []
+
         # Save mode flags as class attributes for monitor access
         self.save_memory_mode = save_memory_mode
         self.accumulate_power = accumulate_power
-        
+
         # Display simulation header and parameters
         sim_params = {
             "Domain size": f"{self.design.width:.2e} x {self.design.height:.2e} m",
@@ -392,9 +398,10 @@ def initialize_simulation(self, save=True, live=True, axis_scale=[-1,1], save_an
         self._effective_save_freq = save_freq * decimate_save
 
         # If in save_memory_mode, clear any existing results to start fresh
-        if save_memory_mode: 
+        if save_memory_mode and not self._cache_fields:
             for field in self.results:
-                if field != 't': self.results[field] = []
+                if field != 't':
+                    self.results[field] = []
             display_status("Memory-saving mode active: Only storing monitor data and/or power accumulation", "info")
 
     def step(self) -> bool:
@@ -429,14 +436,29 @@ def finalize_simulation(self):
         # Display memory usage estimate
         memory_usage = self.estimate_memory_usage(time_steps=self.num_steps, save_fields=self._save_fields)
         display_status(f"Estimated memory usage: {memory_usage['Full simulation']['Total memory (MB)']:.2f} MB", "info")
+        objective_results: Dict[str, float] = {}
+        for idx, monitor in enumerate(self.monitors):
+            if hasattr(monitor, 'evaluate_objective'):
+                value = monitor.evaluate_objective()
+            else:
+                value = None
+            if value is None:
+                continue
+            key = getattr(monitor, 'name', None) or f"monitor_{idx}"
+            objective_results[key] = value
+        if objective_results:
+            self.results['objectives'] = objective_results
+        else:
+            self.results.pop('objectives', None)
+        self.last_objectives = objective_results
         return self.results
 
-    def run(self, steps: Optional[int] = None, save=True, live=True, axis_scale=[-1,1], save_animation=False, 
-            animation_filename='fdtd_animation.mp4', clean_visualization=True, 
+    def run(self, steps: Optional[int] = None, save=True, live=True, axis_scale=[-1,1], save_animation=False,
+            animation_filename='fdtd_animation.mp4', clean_visualization=True,
             save_fields=None, decimate_save=1, accumulate_power=False,
-            save_memory_mode=False) -> Dict:
+            save_memory_mode=False, fields_to_cache: Optional[Sequence[str]] = None) -> Dict:
         """Run the complete simulation using the new step-by-step approach.
-        
+
         Args:
             steps: Number of steps to run. If None, run until the end of the time array.
             save: Whether to save field data at each step.
@@ -448,18 +470,20 @@ def run(self, steps: Optional[int] = None, save=True, live=True, axis_scale=[-1,
             decimate_save: Save only every nth time step (1 = save all, 10 = save every 10th step)
             accumulate_power: Instead of saving all fields, accumulate power and save that
             save_memory_mode: If True, avoid storing all field data and only keep monitors/power
-        
+            fields_to_cache: Fields that should always be cached even when memory saving
+
         Returns:
             Dictionary containing the simulation results.
         """
         # Initialize the simulation
-        self.initialize_simulation(save=save, live=live, axis_scale=axis_scale, 
-                                  save_animation=save_animation, 
+        self.initialize_simulation(save=save, live=live, axis_scale=axis_scale,
+                                  save_animation=save_animation,
                                   animation_filename=animation_filename,
                                   clean_visualization=clean_visualization,
                                   save_fields=save_fields, decimate_save=decimate_save,
-                                  accumulate_power=accumulate_power, 
-                                  save_memory_mode=save_memory_mode)
+                                  accumulate_power=accumulate_power,
+                                  save_memory_mode=save_memory_mode,
+                                  fields_to_cache=fields_to_cache)
 
         # Run the simulation with progress tracking
         with create_rich_progress() as progress:

beamz/simulation/helper.py

@@ -184,15 +184,39 @@ def accumulate_power(fdtd) -> None:
 
 def save_step_results(fdtd) -> None:
     """Save results for this time step if requested and at the right frequency."""
-    if (fdtd._save_results and not fdtd.save_memory_mode and 
-        (fdtd.current_step % fdtd._effective_save_freq == 0 or fdtd.current_step == fdtd.num_steps - 1)):
-        fdtd.results['t'].append(fdtd.t)
-        for field in fdtd._save_fields:
-            arr = getattr(fdtd, field)
-            arr_np = fdtd.backend.to_numpy(fdtd.backend.copy(arr))
-            if np.iscomplexobj(arr_np):
-                arr_np = np.abs(arr_np)
-            fdtd.results[field].append(arr_np)
+    should_save_full = (
+        fdtd._save_results
+        and not fdtd.save_memory_mode
+        and (fdtd.current_step % fdtd._effective_save_freq == 0 or fdtd.current_step == fdtd.num_steps - 1)
+    )
+    should_cache = (
+        fdtd._save_results
+        and fdtd._cache_fields
+        and (fdtd.current_step % fdtd._effective_save_freq == 0 or fdtd.current_step == fdtd.num_steps - 1)
+    )
+    if not should_save_full and not should_cache:
+        return
+
+    if 't' not in fdtd.results:
+        fdtd.results['t'] = []
+    fdtd.results['t'].append(fdtd.t)
+
+    fields_to_store = []
+    if should_save_full:
+        fields_to_store.extend(fdtd._save_fields)
+    if should_cache:
+        for field in fdtd._cache_fields:
+            if field not in fields_to_store:
+                fields_to_store.append(field)
+
+    for field in fields_to_store:
+        arr = getattr(fdtd, field)
+        arr_np = fdtd.backend.to_numpy(fdtd.backend.copy(arr))
+        if np.iscomplexobj(arr_np) and (field not in fdtd._cache_fields):
+            arr_np = np.abs(arr_np)
+        if field not in fdtd.results:
+            fdtd.results[field] = []
+        fdtd.results[field].append(arr_np)
 
 def record_monitor_data(fdtd, step: int) -> None:
     """Record field data at monitor locations for current step."""