Reformat flow description files to comply with Python design guidelines.

hydrogym/firedrake/envs/cavity/flow.py

@@ -16,16 +16,6 @@ class Cavity(FlowConfig):
   yp = np.linspace(-0.05, 0.05, 4)
   X, Y = np.meshgrid(xp, yp)
   DEFAULT_VEL_PROBES = [(x, y) for x, y in zip(X.ravel(), Y.ravel())]
-
-  # # Pressure probes (spaced equally around the cylinder)
-  # xp = np.linspace(0.25, 0.95, 4)
-  # yp = np.linspace(-0.075, 0.075, 3)
-  # X, Y = np.meshgrid(xp, yp)
-  # DEFAULT_PRES_PROBES = [(x, y) for x, y in zip(X.ravel(), Y.ravel())]
-  # DEFAULT_PRES_PROBES.extend([(0.4833333, -0.15), (0.4833333, -0.225),
-  #                            (0.7166666, -0.15), (0.7166666, -0.225)])
-  # DEFAULT_PRES_PROBES.extend([(1.05, 0.005), (1.2, 0.005), (1.35, 0.005), (1.5, 0.005)])
-
   xp = np.linspace(0.1, 0.5, 4)
   yp = np.linspace(-0.025, 0.025, 3)
   X, Y = np.meshgrid(xp, yp)
@@ -128,7 +118,12 @@ def wall_stress_sensor(self, q=None):
     m = fd.assemble(-dot(grad(u[0]), self.n) * ds(self.SENSOR))
     return (m,)
 
-  def evaluate_objective(self, q=None, qB=None, averaged_objective_values=None, return_objective_values=False):
+  def evaluate_objective(
+      self,
+      q=None,
+      qB=None,
+      averaged_objective_values=None,
+      return_objective_values=False):
     if averaged_objective_values is None:
         if q is None:
             q = self.q

hydrogym/firedrake/envs/cylinder/flow.py

@@ -10,24 +10,6 @@
 
 from hydrogym.firedrake import FlowConfig, ObservationFunction, ScaledDirichletBC
 
-# # Velocity probes
-# xp = np.linspace(1.0, 10.0, 16)
-# yp = np.linspace(-2.0, 2.0, 4)
-# X, Y = np.meshgrid(xp, yp)
-# DEFAULT_VEL_PROBES = [(x, y) for x, y in zip(X.ravel(), Y.ravel())]
-
-# # Pressure probes (spaced equally around the cylinder)
-# xp = np.linspace(1.0, 4.0, 4)
-# yp = np.linspace(-0.66, 0.66, 3)
-# X, Y = np.meshgrid(xp, yp)
-# DEFAULT_PRES_PROBES = [(x, y) for x, y in zip(X.ravel(), Y.ravel())]
-
-# RADIUS = 0.5
-# DEFAULT_PRES_PROBES = [
-#     (RADIUS * np.cos(theta), RADIUS * np.sin(theta))
-#     for theta in np.linspace(0, 2 * np.pi, 20, endpoint=False)
-# ]
-
 RADIUS = 0.5
 
 
@@ -55,8 +37,6 @@ class CylinderBase(FlowConfig):
   DEFAULT_DT = 1e-2
 
   # MAX_CONTROL = 0.5 * np.pi
-  # TAU = 0.556  # Time constant for controller damping (0.1*vortex shedding period)
-  # TAU = 0.278  # Time constant for controller damping (0.05*vortex shedding period)
   TAU = 0.0556  # Time constant for controller damping (0.01*vortex shedding period)
 
   # Domain labels
@@ -72,9 +52,10 @@ class CylinderBase(FlowConfig):
   def num_inputs(self) -> int:
     return 1  # Rotary control
 
-  def configure_observations(self,
-                             obs_type=None,
-                             probe_obs_types={}) -> ObservationFunction:
+  def configure_observations(
+      self,
+      obs_type=None,
+      probe_obs_types={}) -> ObservationFunction:
     if obs_type is None:
       obs_type = "lift_drag"
 
@@ -158,7 +139,7 @@ def shear_force(self, q: fd.Function = None) -> float:
     (u, p) = fd.split(q)
     (v, s) = fd.TestFunctions(self.mixed_space)
 
-    # der of velocity wrt to the unit normal at the surface of the cylinder
+    # Order of velocity wrt to the unit normal at the surface of the cylinder
     # equivalent to directional derivative along normal:
     du_dn = dot(self.epsilon(u), self.n)
 
@@ -175,32 +156,17 @@ def shear_force(self, q: fd.Function = None) -> float:
         (direction / self.Re * sqrt(du_dn_t[0]**2 + du_dn_t[1]**2)) *
         ds(self.CYLINDER))
 
-  # TODO: Add back in when linearization is fixed
-  # def linearize_control(self, qB=None):
-  #     """
-  #     Solve linear problem with nonzero Dirichlet BCs to derive forcing term for unsteady DNS
-  #     """
-  #     if qB is None:
-  #         qB = self.solve_steady()
-
-  #     A = self.linearize_dynamics(qB, adjoint=False)
-  #     # M = self.mass_matrix()
-  #     self.linearize_bcs()  # Linearize BCs first (sets freestream to zero)
-  #     self.set_control([1.0])  # Now change the cylinder rotation  TODO: FIX
-
-  #     (v, _) = fd.TestFunctions(self.mixed_space)
-  #     zero = fd.inner(fd.Constant((0, 0)), v) * fd.dx  # Zero RHS for linear form
-
-  #     f = fd.Function(self.mixed_space)
-  #     fd.solve(A == zero, f, bcs=self.collect_bcs())
-  #     return f
-
   def linearize_bcs(self, function_spaces=None):
     self.reset_controls(function_spaces=function_spaces)
     self.bcu_inflow.set_value(fd.Constant((0, 0)))
     self.bcu_freestream.set_value(fd.Constant(0.0))
 
-  def evaluate_objective(self, q: fd.Function = None, averaged_objective_values=None, lambda_value=0.2, return_objective_values=False) -> float:
+  def evaluate_objective(
+      self,
+      q: fd.Function = None,
+      averaged_objective_values=None,
+      lambda_value=0.2,
+      return_objective_values=False) -> float:
     """The objective function for this flow is the drag coefficient"""
     if averaged_objective_values is None:
         CL, CD = self.compute_forces(q=q)

hydrogym/firedrake/envs/pinball/flow.py

@@ -98,9 +98,10 @@ def init_bcs(self, function_spaces=None):
   def num_inputs(self) -> int:
     return len(self.CYLINDER)
 
-  def configure_observations(self,
-                             obs_type=None,
-                             probe_obs_types={}) -> ObservationFunction:
+  def configure_observations(
+      self,
+      obs_type=None,
+      probe_obs_types={}) -> ObservationFunction:
     if obs_type is None:
       obs_type = "lift_drag"
 
@@ -141,22 +142,20 @@ def linearize_bcs(self, function_spaces=None):
     self.bcu_inflow.set_value(fd.Constant((0, 0)))
     self.bcu_freestream.set_value(fd.Constant(0.0))
 
-  # def get_observations(self):
-  #   CL, CD = self.compute_forces()
-  #   return [*CL, *CD]
-
-  def evaluate_objective(self, q: fd.Function = None, averaged_objective_values=None, return_objective_values=False) -> float:
+  def evaluate_objective(
+      self, 
+      q: fd.Function = None,
+      averaged_objective_values=None,
+      return_objective_values=False) -> float:
     """The objective function for this flow is the drag coefficient"""
     if averaged_objective_values is None:
         CL, CD = self.compute_forces(q=q)
         if return_objective_values:
           return [*CL, *CD]
     else:
         CL, CD = averaged_objective_values[:3], averaged_objective_values[3:]
-    # print('pinball lambda', self.reward_lambda, CD, sum(np.square(CD)), flush=True)
-    # return sum(np.square(CD)) + self.reward_lambda * sum(np.square(CL))
+
     return sum(CD)
-    # return -(1.5 - (sum(CD) + self.reward_lambda * np.abs(sum(CL))))
 
   def render(self, mode="human", clim=None, levels=None, cmap="RdBu", **kwargs):
     if clim is None:

hydrogym/firedrake/envs/step/flow.py

@@ -26,8 +26,6 @@ class Step(FlowConfig):
     """
 
   # Velocity probes
-  # xp = np.linspace(3, 8.5, 7)
-  # yp = np.linspace(-0.475, -0.2, 3)
   xp = np.linspace(0.1, 3, 5)
   yp = np.linspace(-0.1, 0.2, 4)
   X, Y = np.meshgrid(xp, yp)
@@ -99,9 +97,10 @@ def body_force(self):
         exp(-((self.x - x0)**2 + (self.y - y0)**2) / delta**2),
     ))
 
-  def configure_observations(self,
-                             obs_type=None,
-                             probe_obs_types={}) -> ObservationFunction:
+  def configure_observations(
+      self,
+      obs_type=None,
+      probe_obs_types={}) -> ObservationFunction:
     if obs_type is None:
       obs_type = "stress_sensor"  # Shear stress on downstream wall
 
@@ -125,13 +124,18 @@ def init_bcs(self, function_spaces=None):
     # Define static boundary conditions
     self.U_inf = ufl.as_tensor(
         (1.0 - ((self.y - 0.25) / 0.25)**2, 0.0 * self.y))
-    self.bcu_inflow = fd.DirichletBC(V, self.U_inf, self.INLET)
-    # self.bcu_noslip = fd.DirichletBC(V, fd.Constant((0, 0)),
-    #                                  (self.WALL, self.SENSOR))
-    self.bcu_noslip = fd.DirichletBC(V, fd.Constant((0, 0)),
-                                    #  (self.WALL, self.SENSOR))
-                                     (self.WALL, 6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30))
-    self.bcp_outflow = fd.DirichletBC(Q, fd.Constant(0), self.OUTLET)
+    self.bcu_inflow = fd.DirichletBC(
+      V,
+      self.U_inf,
+      self.INLET)
+    self.bcu_noslip = fd.DirichletBC(
+      V,
+      fd.Constant((0, 0)),
+      (self.WALL, 6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30))
+    self.bcp_outflow = fd.DirichletBC(
+      Q,
+      fd.Constant(0),
+      self.OUTLET)
 
     # Define time-varying boundary conditions for actuation
     u_bc = ufl.as_tensor((0.0 * self.x, -self.x * (1600 * self.x + 560) / 147))
@@ -196,7 +200,12 @@ def wall_stress_sensor(self, q=None, sensor=None):
     m = fd.assemble(-dot(grad(u[0]), self.n) * ds(sensor))
     return (m,)
 
-  def evaluate_objective(self, q=None, qB=None, averaged_objective_values=None, return_objective_values=False):
+  def evaluate_objective(
+      self,
+      q=None,
+      qB=None,
+      averaged_objective_values=None,
+      return_objective_values=False):
     if averaged_objective_values is None:
         if q is None:
             q = self.q
@@ -207,9 +216,6 @@ def evaluate_objective(self, q=None, qB=None, averaged_objective_values=None, re
         KE = 0.5 * fd.assemble(fd.inner(u - uB, u - uB) * fd.dx)
     else:
         KE = averaged_objective_values[0]
-    # print("KE", KE, flush=True)
-    # ReLen = self.reattachment_length(q)
-    # print("Reattachment length", ReLen[0], flush=True)
     return KE
 
   def render(