Removing the random_seed parameter, as is no longer necessary, thanks to the working synchronization mechanism

Author: afmurillo

dhalsim/physical_process.py

@@ -120,8 +120,8 @@ def __init__(self, intermediate_yaml):
             self.master_time = -1
 
         self.db_update_string = "UPDATE plant SET value = ? WHERE name = ?"
-        self.db_sleep_time = random.uniform(0.01, 0.1)
 
+        self.db_sleep_time = random.uniform(0.01, 0.1)
         self.logger.info("DB Sleep time: " + str(self.db_sleep_time))
 
 
@@ -260,7 +260,6 @@ def create_link_header(a_list):
     def create_attack_header(self):
         """
         Function that creates csv list headers for device and network attacks
-
         :return: list of attack names starting with device and ending with network
         """
         result = []
@@ -311,19 +310,25 @@ def register_initial_results(self):
                     [self.wn.junctions[junction].pressure])
         elif self.simulator == 'wntr':
             for junction in self.junction_list:
-                self.values_list.extend(
-                    [self.wn.get_node(junction).head - self.wn.get_node(junction).elevation])
+                # toDo: Check in wntr 0.4.2 a new way of getting the initial junction pressure
+                self.values_list.extend([0])
 
         if self.simulator == 'epynet':
             # Get pumps flows and status
+            self.logger.debug('Registering initial results of pumps: ' + str(self.pump_list))
             for pump in self.pump_list:
-                self.values_list.extend([self.wn.pumps[pump].flow, self.wn.pumps[pump].status])
+
+                if pump in self.wn.pumps:
+                    self.values_list.extend([self.wn.pumps[pump].flow, self.wn.pumps[pump].status])
+                elif pump in self.wn.valves:
+                    self.values_list.extend([self.wn.valves[pump].flow, self.wn.valves[pump].status])
+                else:
+                    self.logger.error("Error. Actuator " + str(pump)  + " not found in EPANET file")
 
         elif self.simulator == 'wntr':
 
             for pump in self.pump_list:
                 self.values_list.extend([self.wn.get_link(pump).flow])
-
                 if type(self.wn.get_link(pump).status) is int:
                     self.values_list.extend([self.wn.get_link(pump).status])
                 else:
@@ -518,7 +523,6 @@ def get_plcs_ready(self, flag):
     def get_attack_flag(self, name):
         """
         Get the attack flag of this attack.
-
         :return: False if attack not running, true otherwise
         """
 
@@ -701,51 +705,66 @@ def simulate_with_epynet(self, iteration_limit, p_bar):
             #time.sleep(0.3)
 
     def simulate_with_wntr(self, iteration_limit, p_bar):
-        self.logger.info("Starting wntr simulation")
+        self.logger.info("Starting WNTR simulation")
         self.wn.options.time.duration = self.wn.options.time.hydraulic_timestep
 
+        self.register_initial_results()
+        self.results_list.append(self.values_list)
+
         while self.master_time < iteration_limit:
-            conn = sqlite3.connect(self.data["db_path"])
-            c = conn.cursor()
-            c.execute("REPLACE INTO master_time (id, time) VALUES(1, ?)", (str(self.master_time),))
-            conn.commit()
 
-            self.master_time = self.master_time + 1
+            # We check that all PLCs updated their local caches and local CPPPO
+            while not self.get_plcs_ready(1):
+                time.sleep(self.WAIT_FOR_FLAG)
 
-            while not self.get_plcs_ready():
-                time.sleep(0.01)
+            # Notify the PLCs they can start receiving remote values
+            with sqlite3.connect(self.data["db_path"]) as conn:
+                c = conn.cursor()
+                c.execute("UPDATE sync SET flag=2")
+                conn.commit()
 
-            self.update_controls()
+            # Wait for the PLCs to apply control logic
+            while not self.get_plcs_ready(3):
+                time.sleep(self.WAIT_FOR_FLAG)
 
-            self.logger.debug("Iteration {x} out of {y}.".format(x=str(self.master_time),
-                                                                 y=str(iteration_limit)))
+            self.update_controls()
 
-            if p_bar:
-                p_bar.update(self.master_time)
+            self.logger.debug("Iteration {x} out of {y}.".format(x=str(self.master_time), y=str(iteration_limit)))
 
-            # Check for simulation error, print output on exception
             try:
                 self.sim.run_sim(convergence_error=True)
             except Exception as exp:
                 self.logger.error(f"Error in WNTR simulation: {exp}")
                 self.finish()
 
-            self.register_results()
-            self.results_list.append(self.values_list)
-
+            # Updates the SQLite DB
             self.update_tanks()
             self.update_pumps()
             self.update_valves()
             self.update_junctions()
 
+            self.master_time = self.master_time + 1
+            conn = sqlite3.connect(self.data["db_path"])
+            c = conn.cursor()
+            c.execute("REPLACE INTO master_time (id, time) VALUES(1, ?)", (str(self.master_time),))
+            conn.commit()
+
+            if p_bar:
+                p_bar.update(self.master_time)
+
+            self.register_results()
+            self.results_list.append(self.values_list)
+
             # Write results of this iteration if needed
             if 'saving_interval' in self.data and self.master_time != 0 and \
                     self.master_time % self.data['saving_interval'] == 0:
                 self.write_results(self.results_list)
 
             # Set sync flags for nodes
-            c.execute("UPDATE sync SET flag=0")
-            conn.commit()
+            with sqlite3.connect(self.data["db_path"]) as conn:
+                c = conn.cursor()
+                c.execute("UPDATE sync SET flag=0")
+                conn.commit()
 
     def update_tanks(self, network_state=None):
         """Update tanks in database."""
@@ -893,4 +912,4 @@ def is_valid_file(test_parser, arg):
     args = parser.parse_args()
 
     simulation = PhysicalPlant(Path(args.intermediate_yaml))
-    simulation.main()
+    simulation.main()

dhalsim/python2/generic_plc.py

@@ -18,7 +18,6 @@
 import thread
 
 
-
 class Error(Exception):
     """Base class for exceptions in this module."""
 
@@ -112,8 +111,6 @@ def __init__(self, intermediate_yaml_path, yaml_index):
         self.plcs_ready = False
 
         for tag in set(dependant_sensors) - set(plc_sensors):
-
-            #todo: We should query the DB for this
             self.cache[tag] = Decimal(0)
             self.tag_fresh[tag] = False
 
@@ -132,7 +129,6 @@ def generate_real_tags(sensors, dependants, actuators):
         """
         Generates real tags with all sensors, dependants, and actuators
         attached to the plc.
-
         :param sensors: list of sensors attached to the plc
         :param dependants: list of dependant sensors (from other plcs)
         :param actuators: list of actuators controlled by the plc
@@ -155,7 +151,6 @@ def generate_real_tags(sensors, dependants, actuators):
     def generate_tags(taggable):
         """
         Generates tags from a list of taggable entities (sensor or actuator)
-
         :param taggable: a list of strings containing names of things like tanks, pumps, and valves
         """
         tags = []
@@ -171,7 +166,6 @@ def generate_tags(taggable):
     def create_controls(controls_list):
         """
         Generates list of control objects for a plc
-
         :param controls_list: a list of the control dicts to be converted to Control objects
         """
         ret = []
@@ -195,7 +189,6 @@ def create_controls(controls_list):
     @staticmethod
     def create_attacks(attack_list):
         """This function will create an array of DeviceAttacks
-
         :param attack_list: A list of attack dicts that need to be converted to DeviceAttacks
         """
         attacks = []
@@ -226,7 +219,6 @@ def pre_loop(self, sleep=0.5):
         """
         The pre loop of a PLC. In everything is setup. Like starting the sending thread through
         the :class:`~dhalsim.python2.basePLC` class.
-
         :param sleep:  (Default value = 0.5) The time to sleep after setting everything up
         """
         self.logger.debug(self.intermediate_plc['name'] + ' enters pre_loop')
@@ -259,7 +251,6 @@ def pre_loop(self, sleep=0.5):
     def get_tag(self, tag):
         """
         Get the value of a tag that is connected to this PLC or over the network.
-
         :param tag: The tag to get
         :type tag: str
         :return: value of that tag
@@ -317,15 +308,14 @@ def update_cache(self, a, cache_update_time):
                         res = self.get_tag_for_cache(cached_tag, plc_data["public_ip"], cache_update_time)
                         if self.get_master_clock() == start_iteration:
                             self.tag_fresh[cached_tag] = res
-                            
+
                         if not self.tag_fresh[cached_tag]:
                             self.logger.info("Warning: Cache for tag " + str(cached_tag) + " could not be updated")
                             self.tag_fresh[cached_tag] = True
 
     def set_tag(self, tag, value):
         """
         Set a tag that is connected to this PLC to a value.
-
         :param tag: Which tag to set
         :type tag: str
         :param value: value to set the Tag to
@@ -349,14 +339,10 @@ def db_query(self, query, write=False, parameters=None):
         On a :code:`sqlite3.OperationalError` it will retry with a max of :code:`DB_TRIES` tries.
         Before it reties, it will sleep for :code:`DB_SLEEP_TIME` seconds.
         This is necessary because of the limited concurrency in SQLite.
-
         :param query: The SQL query to execute in the db
         :type query: str
-
         :param write: Boolean flag to indicate if this query will write into the database
-
         :param parameters: The parameters to put in the query. This must be a tuple.
-
         :raise DatabaseError: When a :code:`sqlite3.OperationalError` is still raised after
            :code:`DB_TRIES` tries.
         """
@@ -389,9 +375,7 @@ def get_master_clock(self):
         Get the value of the master clock of the physical process through the database.
         On a :code:`sqlite3.OperationalError` it will retry with a max of :code:`DB_TRIES` tries.
         Before it reties, it will sleep for :code:`DB_SLEEP_TIME` seconds.
-
         :return: Iteration in the physical process.
-
         :raise DatabaseError: When a :code:`sqlite3.OperationalError` is still raised after
            :code:`DB_TRIES` tries.
         """
@@ -403,9 +387,7 @@ def get_sync(self, flag):
         Get the sync flag of this plc.
         On a :code:`sqlite3.OperationalError` it will retry with a max of :code:`DB_TRIES` tries.
         Before it reties, it will sleep for :code:`DB_SLEEP_TIME` seconds.
-
         :return: False if physical process wants the plc to do a iteration, True if not.
-
         :raise DatabaseError: When a :code:`sqlite3.OperationalError` is still raised after
            :code:`DB_TRIES` tries.
         """
@@ -418,10 +400,8 @@ def set_sync(self, flag):
         knows this plc finished the requested iteration.
         On a :code:`sqlite3.OperationalError` it will retry with a max of :code:`DB_TRIES` tries.
         Before it reties, it will sleep for :code:`DB_SLEEP_TIME` seconds.
-
         :param flag: True for sync to 1, False for sync to 0
         :type flag: bool
-
         :raise DatabaseError: When a :code:`sqlite3.OperationalError` is still raised after
            :code:`DB_TRIES` tries.
         """
@@ -433,13 +413,10 @@ def set_attack_flag(self, flag, attack_name):
         provided name is currently running. When it is 0, it is not.
         On a :code:`sqlite3.OperationalError` it will retry with a max of :code:`DB_TRIES` tries.
         Before it reties, it will sleep for :code:`DB_SLEEP_TIME` seconds.
-
         :param flag: True for running to 1, False for running to 0
         :type flag: bool
-
         :param attack_name: The name of the attack
         :type attack_name: str
-
         :raise DatabaseError: When a :code:`sqlite3.OperationalError` is still raised after
            :code:`DB_TRIES` tries.
         """
@@ -451,7 +428,6 @@ def stop_cache_update(self):
     def main_loop(self, sleep=0.5, test_break=False):
         """
         The main loop of a PLC. In here all the controls will be applied.
-
         :param sleep:  (Default value = 0.5) Not used
         :param test_break:  (Default value = False) used for unit testing, breaks the loop after one iteration
         """
@@ -498,6 +474,7 @@ def main_loop(self, sleep=0.5, test_break=False):
             if test_break:
                 break
 
+
 def is_valid_file(parser_instance, arg):
     if not os.path.exists(arg):
         parser_instance.error(arg + " does not exist")
@@ -516,4 +493,4 @@ def is_valid_file(parser_instance, arg):
     args = parser.parse_args()
     plc = GenericPLC(
         intermediate_yaml_path=Path(args.intermediate_yaml),
-        yaml_index=args.index)
+        yaml_index=args.index)