Fixed ruff check issues

Fixed issue where exceptions raised from within exceptions where not passed the context. Added defaults for stacklevel when using warn(). Removed redundant comparison for void test function. Disabled warning for use of private functions in tests. Disabled warning for unneeded list comprehension which I believe is required. Fixed line too long

Author: ptsOSL

atip/__init__.py

@@ -1,6 +1,7 @@
 """ATIP: Accelerator Toolbox Interface for Pytac.
 See README.rst & INSTALL.rst for more information.
 """
+
 from . import load_sim, sim_data_sources, simulator, utils
 
 __all__ = ["load_sim", "sim_data_sources", "simulator", "utils"]

src/atip/simulator.py

@@ -197,7 +197,7 @@ def _recalculate_phys_data(self, callback):
                         self._at_lat, self._rp, self._disable_emittance
                     )
                 except Exception as e:
-                    warn(at.AtWarning(e))
+                    warn(at.AtWarning(e), stacklevel=1)
                 # Signal the up to date flag since the physics data is now up to date.
                 # We do this before the callback is executed in case the callback
                 # checks the flag.

src/atip/utils.py

@@ -73,7 +73,7 @@ def preload_at(at_lat):
     class elems:
         pass
 
-    elems.all = [elem for elem in at_lat]
+    elems.all = [elem for elem in at_lat]  # noqa: C416
     elems_dict = {
         type_: []
         for type_ in [
@@ -142,7 +142,7 @@ def get_atsim(target):
     if isinstance(target, atip.simulator.ATSimulator):
         return target
     else:  # Pytac lattice
-        return target._data_source_manager._data_sources[pytac.SIM]._atsim
+        return target._data_source_manager._data_sources[pytac.SIM]._atsim  # noqa: SLF001
 
 
 def get_sim_lattice(target):

src/virtac/atip_ioc_entry.py

@@ -78,17 +78,22 @@ def main():
     if 5064 in ports_list or 5065 in ports_list:
         warn(
             f"At least one of {epics_env_vars} is set to 5064 or 5065"
-            + conflict_warning
+            + conflict_warning,
+            stacklevel=1,
         )
     elif all(port == 0 for port in ports_list):
         warn(
             "No EPICS port set, default base port (5064) will be used"
-            + conflict_warning
+            + conflict_warning,
+            stacklevel=1,
         )
     # Avoid PV conflict between multiple IP interfaces on the same machine.
     primary_ip = socket.gethostbyname(socket.getfqdn())
     if "EPICS_CAS_INTF_ADDR_LIST" in os.environ.keys():
-        warn("Pre-existing 'EPICS_CAS_INTF_ADDR_LIST' value" + conflict_warning)
+        warn(
+            "Pre-existing 'EPICS_CAS_INTF_ADDR_LIST' value" + conflict_warning,
+            stacklevel=1,
+        )
     else:
         os.environ["EPICS_CAS_INTF_ADDR_LIST"] = primary_ip
         os.environ["EPICS_CAS_BEACON_ADDR_LIST"] = primary_ip

src/virtac/atip_server.py

@@ -309,10 +309,10 @@ def _create_feedback_records(self, feedback_csv, disable_emittance):
             try:
                 readonly = ast.literal_eval(line["read-only"])
                 assert isinstance(readonly, bool)
-            except (ValueError, AssertionError):
+            except (ValueError, AssertionError) as exc:
                 raise ValueError(
                     f"Unable to evaluate {line['read-only']} as a boolean."
-                )
+                ) from exc
             prefix, suffix = line["pv"].split(":", 1)
             builder.SetDeviceName(prefix)
             if readonly:
@@ -462,7 +462,7 @@ def monitor_mirrored_pvs(self):
             try:
                 self._monitored_pvs[pv] = camonitor(pv, mask.callback)
             except Exception as e:
-                warn(e)
+                warn(e, stacklevel=1)
 
     def refresh_record(self, pv_name):
         """For a given PV refresh the time-stamp of the associated record,
@@ -473,10 +473,10 @@ def refresh_record(self, pv_name):
         """
         try:
             record = self.all_record_names[pv_name]
-        except KeyError:
+        except KeyError as exc:
             raise ValueError(
                 f"{pv_name} is not the name of a record created by this server."
-            )
+            ) from exc
         else:
             record.set(record.get())
 
@@ -513,7 +513,7 @@ def setup_tune_feedback(self, tune_csv=None):
                     line["delta"], mask.callback
                 )
             except Exception as e:
-                warn(e)
+                warn(e, stacklevel=1)
 
     def stop_all_monitoring(self):
         """Stop monitoring mirrored records and tune feedback offsets."""
@@ -546,12 +546,13 @@ def set_feedback_record(self, index, field, value):
         """
         try:
             self._feedback_records[(index, field)].set(value)
-        except KeyError:
+        except KeyError as exc:
             if index == 0:
                 raise FieldException(
                     f"Simulated lattice {self.lattice} does not have field {field}."
-                )
+                ) from exc
             else:
                 raise FieldException(
-                    f"Simulated element {self.lattice[index]} does not have field {field}."
-                )
+                    f"Simulated element {self.lattice[index]} does not have "
+                    f"field {field}."
+                ) from exc

tests/test_at_simulator_object.py

@@ -199,8 +199,7 @@ def test_toggle_calculations_and_wait_for_calculations(atsim, initial_phys_data)
     atsim._at_lat[5].PolynomB[1] = 2.5
     atsim.queue_set(mock.Mock(), "f", 0)
     assert atsim.wait_for_calculations(2) is False
-    _check_initial_phys_data(atsim, initial_phys_data) is True
-    atsim.toggle_calculations()
+    _check_initial_phys_data(atsim, initial_phys_data)
     atsim.queue_set(mock.Mock(), "f", 0)
     assert atsim.wait_for_calculations() is True
     # Physics data has changed.