Ruff lint/formatting

python/PiFinder/catalogs.py

@@ -271,7 +271,7 @@ def add_object(self, obj: CompositeObject):
 
     def _add_object(self, obj: CompositeObject):
         self.__objects.append(obj)
-        if (obj.sequence > self.max_sequence):
+        if obj.sequence > self.max_sequence:
             self.max_sequence = obj.sequence
 
     def add_objects(self, objects: List[CompositeObject]):
@@ -313,8 +313,7 @@ def _update_id_to_pos(self):
         self.id_to_pos = {obj.id: i for i, obj in enumerate(self.__objects)}
 
     def _update_sequence_to_pos(self):
-        self.sequence_to_pos = {obj.sequence: i for i,
-                                obj in enumerate(self.__objects)}
+        self.sequence_to_pos = {obj.sequence: i for i, obj in enumerate(self.__objects)}
 
     def __repr__(self):
         return f"Catalog({self.catalog_code=}, {self.max_sequence=}, count={self.get_count()})"
@@ -355,8 +354,10 @@ def filter_objects(self) -> List[CompositeObject]:
 
         self.filtered_objects = self.catalog_filter.apply(self.get_objects())
         logger.info(
-            "FILTERED %s %d/%d", self.catalog_code, len(
-                self.filtered_objects), len(self.get_objects())
+            "FILTERED %s %d/%d",
+            self.catalog_code,
+            len(self.filtered_objects),
+            len(self.get_objects()),
         )
         self.filtered_objects_seq = self._filtered_objects_to_seq()
         self.last_filtered = time.time()
@@ -369,7 +370,7 @@ def get_filtered_count(self):
         return len(self.filtered_objects)
 
     def get_age(self) -> Optional[int]:
-        """ If the catalog data is time-sensitive, return age in days. """
+        """If the catalog data is time-sensitive, return age in days."""
         return None
 
     def __repr__(self):
@@ -468,7 +469,8 @@ def add(self, catalog: Catalog, select: bool = False):
             self.__catalogs.append(catalog)
         else:
             logger.warning(
-                "Catalog %s already exists, not replaced (in Catalogs.add)", catalog.catalog_code
+                "Catalog %s already exists, not replaced (in Catalogs.add)",
+                catalog.catalog_code,
             )
 
     def remove(self, catalog_code: str):
@@ -477,8 +479,7 @@ def remove(self, catalog_code: str):
                 self.__catalogs.remove(catalog)
                 return
 
-        logger.warning(
-            "Catalog %s does not exist, cannot remove", catalog_code)
+        logger.warning("Catalog %s does not exist, cannot remove", catalog_code)
 
     def get_codes(self, only_selected: bool = True) -> List[str]:
         return_list = []
@@ -648,8 +649,7 @@ def do_timed_task(self):
                     planet = planet_dict[name]
                     obj.ra, obj.dec = planet["radec"]
                     obj.mag = MagnitudeObject([planet["mag"]])
-                    obj.const = sf_utils.radec_to_constellation(
-                        obj.ra, obj.dec)
+                    obj.const = sf_utils.radec_to_constellation(obj.ra, obj.dec)
                     obj.mag_str = obj.mag.calc_two_mag_representation()
 
 
@@ -664,7 +664,7 @@ def __init__(self, dt: datetime.datetime, shared_state: SharedStateObj):
         self._start_background_init(dt)
 
     def get_age(self) -> Optional[int]:
-        """ Return the age of the comet data in days """
+        """Return the age of the comet data in days"""
         return self.age
 
     def _start_background_init(self, dt):
@@ -676,7 +676,8 @@ def init_task():
                         self.initialised = self.calc_comet_first_time(dt)
                     with self.virtual_id_lock:
                         new_low = self.assign_virtual_object_ids(
-                            self, self.virtual_id_low)
+                            self, self.virtual_id_low
+                        )
                         self.virtual_id_low = new_low
                     break
                 time.sleep(60)  # retry every minute to download comet data
@@ -723,15 +724,16 @@ def add_comet(self, sequence: int, name: str, comet: Dict[str, Dict[str, float]]
         self.add_object(obj)
 
     def do_timed_task(self):
-        """ updating comet catalog data """
+        """updating comet catalog data"""
         with Timer("Comet Catalog periodic update"):
             with self._init_lock:
                 if not self.initialised:
                     logging.debug("Comets not yet initialised, skip periodic update...")
                     return
             dt = self.shared_state.datetime()
             comet_dict = comets.calc_comets(
-                dt, [x.names[0] for x in self._get_objects()])
+                dt, [x.names[0] for x in self._get_objects()]
+            )
             if not comet_dict:
                 return
             for obj in self._get_objects():
@@ -756,8 +758,7 @@ def build(self, shared_state) -> Catalogs:
         db: Database = ObjectsDatabase()
         obs_db: Database = ObservationsDatabase()
         # list of dicts, one dict for each entry in the catalog_objects table
-        catalog_objects: List[Dict] = [
-            dict(row) for row in db.get_catalog_objects()]
+        catalog_objects: List[Dict] = [dict(row) for row in db.get_catalog_objects()]
         objects = db.get_objects()
         common_names = Names()
         catalogs_info = db.get_catalogs_dict()
@@ -769,8 +770,7 @@ def build(self, shared_state) -> Catalogs:
         # to speed up repeated searches
         self.catalog_dicts = {}
         logger.debug("Loaded %i objects from database", len(composite_objects))
-        all_catalogs: Catalogs = self._get_catalogs(
-            composite_objects, catalogs_info)
+        all_catalogs: Catalogs = self._get_catalogs(composite_objects, catalogs_info)
         # Initialize planet catalog with whatever date we have for now
         # This will be re-initialized on activation of Catalog ui module
         # if we have GPS lock
@@ -792,8 +792,7 @@ def check_catalogs_sequences(self, catalogs: Catalogs):
         for catalog in catalogs.get_catalogs(only_selected=False):
             result = catalog.check_sequences()
             if not result:
-                logger.error("Duplicate sequence catalog %s!",
-                             catalog.catalog_code)
+                logger.error("Duplicate sequence catalog %s!", catalog.catalog_code)
                 return False
             return True
 

python/PiFinder/comets.py

@@ -2,9 +2,8 @@
 from datetime import datetime, timezone
 from skyfield.data import mpc
 from skyfield.constants import GM_SUN_Pitjeva_2005_km3_s2 as GM_SUN
-from skyfield.elementslib import osculating_elements_of
 from PiFinder.utils import Timer, comet_file
-from PiFinder.calc_utils import sf_utils, ra_to_hms, dec_to_dms
+from PiFinder.calc_utils import sf_utils
 import requests
 import os
 import logging
@@ -16,7 +15,7 @@
 def process_comet(comet_data, dt) -> Dict[str, Any]:
     name, row = comet_data
     t = sf_utils.ts.from_datetime(dt)
-    sun = sf_utils.eph['sun']
+    sun = sf_utils.eph["sun"]
     comet = sun + mpc.comet_orbit(row, sf_utils.ts, GM_SUN)
 
     # print(f"Processing comet: {name}, {sf_utils.observer_loc}")
@@ -26,10 +25,13 @@ def process_comet(comet_data, dt) -> Dict[str, Any]:
     ra, dec, earth_distance = topocentric.radec(sf_utils.ts.J2000)
     sun_distance = heliocentric.radec(sf_utils.ts.J2000)[2]
 
-    mag_g = float(row['magnitude_g'])
-    mag_k = float(row['magnitude_k'])
-    mag = mag_g + 2.5 * mag_k * math.log10(sun_distance.au) + \
-        5.0 * math.log10(earth_distance.au)
+    mag_g = float(row["magnitude_g"])
+    mag_k = float(row["magnitude_k"])
+    mag = (
+        mag_g
+        + 2.5 * mag_k * math.log10(sun_distance.au)
+        + 5.0 * math.log10(earth_distance.au)
+    )
     if mag > 15:
         return {}
 
@@ -45,11 +47,13 @@ def process_comet(comet_data, dt) -> Dict[str, Any]:
         "earth_distance": earth_distance.au,
         "sun_distance": sun_distance.au,
         "orbital_elements": None,  # could add this later
-        "row": row
+        "row": row,
     }
 
 
-def comet_data_download(local_filename, url=mpc.COMET_URL) -> Tuple[bool, Optional[float]]:
+def comet_data_download(
+    local_filename, url=mpc.COMET_URL
+) -> Tuple[bool, Optional[float]]:
     """
     Download the latest comet data from the Minor Planet Center.
     Return values are succes and the age of the file in days, if available.
@@ -62,15 +66,19 @@ def comet_data_download(local_filename, url=mpc.COMET_URL) -> Tuple[bool, Option
         response.raise_for_status()  # Raise an exception for bad responses
 
         # Try to get the Last-Modified header
-        last_modified = response.headers.get('Last-Modified')
+        last_modified = response.headers.get("Last-Modified")
 
         if last_modified:
-            remote_date = datetime.strptime(last_modified, '%a, %d %b %Y %H:%M:%S GMT').replace(tzinfo=timezone.utc)
+            remote_date = datetime.strptime(
+                last_modified, "%a, %d %b %Y %H:%M:%S GMT"
+            ).replace(tzinfo=timezone.utc)
             logger.debug(f"Remote Last-Modified: {remote_date}")
 
             # Check if local file exists and its modification time
             if os.path.exists(local_filename):
-                local_date = datetime.fromtimestamp(os.path.getmtime(local_filename)).replace(tzinfo=timezone.utc)
+                local_date = datetime.fromtimestamp(
+                    os.path.getmtime(local_filename)
+                ).replace(tzinfo=timezone.utc)
                 logger.debug(f"Local Last-Modified: {local_date}")
 
                 if remote_date <= local_date:
@@ -82,7 +90,7 @@ def comet_data_download(local_filename, url=mpc.COMET_URL) -> Tuple[bool, Option
             response = requests.get(url)
             response.raise_for_status()
 
-            with open(local_filename, 'wb') as f:
+            with open(local_filename, "wb") as f:
                 f.write(response.content)
 
             # Set the file's modification time to match the server's last-modified time
@@ -95,7 +103,7 @@ def comet_data_download(local_filename, url=mpc.COMET_URL) -> Tuple[bool, Option
             response = requests.get(url)
             response.raise_for_status()
 
-            with open(local_filename, 'wb') as f:
+            with open(local_filename, "wb") as f:
                 f.write(response.content)
 
             logger.debug("File downloaded successfully.")
@@ -110,15 +118,20 @@ def calc_comets(dt, comet_names=None) -> dict:
     with Timer("calc_comets()"):
         comet_dict: Dict[str, Any] = {}
         if sf_utils.observer_loc is None or dt is None:
-            logger.debug(f"calc_comets can't run: observer loc is None: {sf_utils.observer_loc is None}, dt is None: {dt is None}")
+            logger.debug(
+                f"calc_comets can't run: observer loc is None: {sf_utils.observer_loc is None}, dt is None: {dt is None}"
+            )
             return comet_dict
 
         with open(comet_file, "rb") as f:
             comets_df = mpc.load_comets_dataframe(f)
 
-        comets_df = (comets_df.sort_values('reference')
-                     .groupby('designation', as_index=False).last()
-                     .set_index('designation', drop=False))
+        comets_df = (
+            comets_df.sort_values("reference")
+            .groupby("designation", as_index=False)
+            .last()
+            .set_index("designation", drop=False)
+        )
 
         comet_data = list(comets_df.iterrows())
 

python/PiFinder/composite_object.py

@@ -27,7 +27,7 @@ def calc_filter_mag(self):
             self.filter_mag = self.UNKNOWN_MAG
 
     def _filter_floats(self) -> List[float]:
-        """ only used valid floats for magnitude"""
+        """only used valid floats for magnitude"""
         return [float(x) for x in self.mags if is_number(x)]
 
     def calc_two_mag_representation(self):

python/PiFinder/main.py

@@ -299,7 +299,9 @@ def main(
         shared_state.set_ui_state(ui_state)
         shared_state.set_arch(arch)  # Normal
         logger.debug("Ui state in main is" + str(shared_state.ui_state()))
-        console = UIConsole(display_device, None, shared_state, command_queues, cfg, Catalogs([]))
+        console = UIConsole(
+            display_device, None, shared_state, command_queues, cfg, Catalogs([])
+        )
         console.write("Starting....")
         console.update()
 

python/PiFinder/state.py

@@ -12,15 +12,14 @@
 from PiFinder import config
 import logging
 from typing import List
-import threading
 from PiFinder.composite_object import CompositeObject
 
 logger = logging.getLogger("SharedState")
 
 
 class RecentCompositeObjectList(list):
-    """ keeps the recent list of composite_objects,
-        handling duplicates """
+    """keeps the recent list of composite_objects,
+    handling duplicates"""
 
     def __init__(self):
         super().__init__()

python/PiFinder/sys_utils.py

@@ -5,6 +5,7 @@
 try:
     import sh
     from sh import wpa_cli, unzip, su, passwd
+
     REAL_SYS_UTILS = True
 except ImportError:
     REAL_SYS_UTILS = False

python/PiFinder/ui/base.py

@@ -8,7 +8,7 @@
 import time
 import uuid
 from itertools import cycle
-from typing import Type, Union, Optional
+from typing import Type, Union
 
 from PIL import Image, ImageDraw
 from PiFinder import utils
@@ -156,8 +156,7 @@ def message(self, message, timeout: float = 2, size=(5, 44, 123, 84)):
             fill=self.colors.get(0),
             outline=self.colors.get(0),
         )
-        self.draw.rectangle(size, fill=self.colors.get(0),
-                            outline=self.colors.get(128))
+        self.draw.rectangle(size, fill=self.colors.get(0), outline=self.colors.get(128))
 
         line_length = int((size[2] - size[0]) / self.fonts.bold.width)
         message = " " * int((line_length - len(message)) / 2) + message
@@ -190,8 +189,7 @@ def screen_update(self, title_bar=True, button_hints=True) -> None:
                     (6, 1), str(self.fps), font=self.fonts.bold.font, fill=fg
                 )
             else:
-                self.draw.text((6, 1), self.title,
-                               font=self.fonts.bold.font, fill=fg)
+                self.draw.text((6, 1), self.title, font=self.fonts.bold.font, fill=fg)
             imu = self.shared_state.imu()
             moving = True if imu and imu["pos"] and imu["moving"] else False
 
@@ -224,8 +222,7 @@ def screen_update(self, title_bar=True, button_hints=True) -> None:
                     # a fresh cam solve sets unmoved to True
                     self._unmoved = True if cam_active else self._unmoved
                     if self._unmoved:
-                        time_since_cam_solve = time.time() - \
-                            solution["cam_solve_time"]
+                        time_since_cam_solve = time.time() - solution["cam_solve_time"]
                         var_fg = min(64, int(time_since_cam_solve / 6 * 64))
                     # self.draw.rectangle([115, 2, 125, 14], fill=bg)
 

python/PiFinder/ui/callbacks.py

@@ -14,6 +14,7 @@
 from PiFinder import utils
 from PiFinder.ui.base import UIModule
 from PiFinder.catalogs import CatalogFilter
+
 sys_utils = utils.get_sys_utils()
 
 

python/PiFinder/ui/equipment.py

@@ -1,11 +1,8 @@
-import time
-
 from PiFinder.ui.base import UIModule
 from PiFinder import utils
 from PiFinder.state_utils import sleep_for_framerate
 from PiFinder.ui.marking_menus import MarkingMenuOption, MarkingMenu
 from PiFinder.ui.text_menu import UITextMenu
-from PiFinder.ui.ui_utils import TextLayouter, SpaceCalculatorFixed
 from PiFinder import config
 
 sys_utils = utils.get_sys_utils()
@@ -82,12 +79,16 @@ def update(self, force=False):
         selected_telescope = self.config_object.get_option("session.log_telescope", "")
 
         if selected_eyepiece != "":
-            cfg.equipment.set_active_eyepiece(cfg.equipment.eyepieces[selected_eyepiece])
+            cfg.equipment.set_active_eyepiece(
+                cfg.equipment.eyepieces[selected_eyepiece]
+            )
             cfg.save_equipment()
             self.config_object.set_option("session.log_eyepiece", "")
 
         if selected_telescope != "":
-            cfg.equipment.set_active_telescope(cfg.equipment.telescopes[selected_telescope])
+            cfg.equipment.set_active_telescope(
+                cfg.equipment.telescopes[selected_telescope]
+            )
             cfg.save_equipment()
             self.config_object.set_option("session.log_telescope", "")
 
@@ -121,7 +122,10 @@ def update(self, force=False):
                 fill=self.colors.get(128),
             )
 
-        if cfg.equipment.active_telescope is not None and cfg.equipment.active_eyepiece is not None:
+        if (
+            cfg.equipment.active_telescope is not None
+            and cfg.equipment.active_eyepiece is not None
+        ):
             mag = cfg.equipment.calc_magnification()
             if mag > 0:
                 self.draw.text(