Improve linting and commented code structure

Remove top level logging

jarvis/api/routers/investment.py

@@ -157,11 +157,12 @@ async def robinhood_path(request: Request, token: str = None):
         with open(models.fileio.robinhood) as static_file:
             html_content = static_file.read()
         content_type, _ = mimetypes.guess_type(html_content)
+        # serves as a static webpage
         return HTMLResponse(
             status_code=HTTPStatus.TEMPORARY_REDIRECT.real,
             content=html_content,
             media_type=content_type,
-        )  # serves as a static webpage
+        )
     else:
         raise APIResponse(
             status_code=HTTPStatus.EXPECTATION_FAILED.real,

jarvis/api/routers/stats.py

@@ -102,7 +102,7 @@ def get_files() -> Generator[FilePath]:
                 yield file_path
 
 
-@cache.timed_cache(max_age=900)  # Cache for 15 minutes
+@cache.timed_cache(max_age=900)
 def total_lines_of_code() -> PositiveInt:
     """Cached function to calculate the total lines of code.
 
@@ -114,7 +114,7 @@ def total_lines_of_code() -> PositiveInt:
     return sum(count_lines(file) for file in get_files())
 
 
-@cache.timed_cache(max_age=900)  # Cache for 15 minutes
+@cache.timed_cache(max_age=900)
 def total_files() -> PositiveInt:
     """Cached function to calculate the total number of files.
 

jarvis/api/routers/stock_monitor.py

@@ -162,9 +162,8 @@ async def stock_monitor_api(
     else:  # If apikey auth fails or unsupported
         sent_dict = settings.stock_monitor_helper.otp_sent
         recd_dict = settings.stock_monitor_helper.otp_recd
-        email_otp = email_otp or request.headers.get(
-            "email-otp"
-        )  # variable will be _ but headers should always be `-`
+        # variable will be _ but headers should always be `-`
+        email_otp = email_otp or request.headers.get("email-otp")
         if email_otp:
             recd_dict[input_data.email] = email_otp
         if secrets.compare_digest(
@@ -336,7 +335,8 @@ async def stock_monitor_api(
             "Please choose a lower 'Min' value or try at a later time.",
         )
 
-    stockmonitor_squire.insert_stock_userdata(params=new_entry)  # Store it in database
+    # Store it in database
+    stockmonitor_squire.insert_stock_userdata(params=new_entry)
 
     response = (
         f"Entry added to the database. Jarvis will notify you at {input_data.email!r} when a "

jarvis/api/squire/discover.py

@@ -75,9 +75,10 @@ def routes(routers: str) -> Generator[APIRouter]:
         APIRouter:
         API Router from scanned modules.
     """
+    # sort by name of the route
     entrypoints: List[Entrypoint] = sorted(
         get_entrypoints(routers=routers), key=lambda ent: ent.stem
-    )  # sort by name of the route
+    )
     for entrypoint in entrypoints:
         try:
             route = import_module(entrypoint.module)

jarvis/api/squire/stockmonitor_squire.py

@@ -96,13 +96,15 @@ def put_daily_alerts(
     with stock_db.connection:
         cursor = stock_db.connection.cursor()
         params = [(key, *values) for param in params for key, values in param.items()]
-        cursor.execute("DELETE FROM stock_daily")  # clear all existing data
+        # clear all existing data
+        cursor.execute("DELETE FROM stock_daily")
         query = (
             f"INSERT OR REPLACE INTO stock_daily {settings.stock_monitor.alerts} VALUES "
             f"{settings.stock_monitor.alert_values};"
         )
         for param in params:
-            cursor.execute(query, param)  # write new data in db
+            # write new data in db
+            cursor.execute(query, param)
         stock_db.connection.commit()
 
 

jarvis/api/squire/surveillance_squire.py

@@ -131,7 +131,8 @@ def gen_frames(manager: Queue, index: int, available_cameras: List[str]) -> None
             logger.info("Releasing camera: %s", available_cameras[index])
             cam.release()
             break
-        frame = cv2.flip(src=frame, flipCode=1)  # mirrors the frame
+        # mirrors the frame
+        frame = cv2.flip(src=frame, flipCode=1)
         ret, buffer = cv2.imencode(ext=".jpg", img=frame)
         frame = buffer.tobytes()
         manager.put(frame)

jarvis/api/triggers/stock_monitor.py

@@ -27,9 +27,10 @@ def generate_graph(logger: logging.Logger, ticker: str, bars: int = 300) -> str
         https://stackoverflow.com/a/49729752
     """
     logger.info("Generating price chart for '%s'", ticker)
+    # ~ 1 month
     dataframe = webull().get_bars(
         stock=ticker, interval="m60", count=bars, extendTrading=1
-    )  # ~ 1 month
+    )
     refined = dataframe[["close"]]
     if len(refined) == 0:
         refined = dataframe[["open"]]
@@ -205,7 +206,10 @@ def skip_signal(
                     if time.time() <= alert_time + hours * 60 * 60:
                         return True
                     else:
-                        self.repeat_alerts.remove({alert_time: alert_entry})
+                        try:
+                            self.repeat_alerts.remove({alert_time: alert_entry})
+                        except ValueError as err:
+                            self.logger.error(err)
                         return False  # notification should be triggered if condition matches
 
     def send_notification(self) -> None:

jarvis/executors/car.py

@@ -197,22 +197,18 @@ def enable_guard(self, phrase) -> str:
             or 1
         )
         if "hour" in phrase:
-            seconds = (
-                requested_expiry * 3_600
-            )  # Defaults to 1 hour if no numeric value in phrase
+            # Defaults to 1 hour if no numeric value in phrase
+            seconds = requested_expiry * 3_600
         elif "day" in phrase:
-            seconds = (
-                requested_expiry * 86_400
-            )  # Defaults to 1 day if no numeric value in phrase
+            # Defaults to 1 day if no numeric value in phrase
+            seconds = requested_expiry * 86_400
         elif "week" in phrase:
-            seconds = (
-                requested_expiry * 604_800
-            )  # Defaults to 1 week if no numeric value in phrase
+            # Defaults to 1 week if no numeric value in phrase
+            seconds = requested_expiry * 604_800
         else:
             seconds = 3_600  # Defaults to 1 hour if no datetime conversion was received
-        expire = int(
-            (time.time() + seconds) * 1000
-        )  # multiply by 1000 to including microseconds making it 13 digits
+        # multiply by 1000 to including microseconds making it 13 digits
+        expire = int((time.time() + seconds) * 1000)
         if response := self.object(operation="SECURE", end_time=expire):
             return response
         else:
@@ -554,9 +550,8 @@ def vehicle(
                         logger.error(lock_response)
                     else:
                         logger.info("Vehicle has been locked!")
-                        time.sleep(
-                            3
-                        )  # Wait before locking the car, so that there is no overlap in refresh token
+                        # Wait before locking the car, so that there is no overlap in refresh token
+                        time.sleep(3)
             response = control.remote_engine_start(
                 pin=models.env.car_pin, target_value=temp
             )

jarvis/executors/connection.py

@@ -21,9 +21,8 @@ def wifi(conn_object: classes.WiFiConnection) -> classes.WiFiConnection | None:
     socket_ = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
     try:
         if models.settings.os == models.supported_platforms.windows:
-            connection = HTTPSConnection(
-                "8.8.8.8", timeout=3
-            )  # Recreate a new connection everytime
+            # Recreate a new connection everytime
+            connection = HTTPSConnection("8.8.8.8", timeout=3)
             connection.request("HEAD", "/")
         else:
             socket_.connect(("8.8.8.8", 80))
@@ -37,7 +36,8 @@ def wifi(conn_object: classes.WiFiConnection) -> classes.WiFiConnection | None:
     except OSError as error:
         conn_object.os_errors += 1
         logger.error("OSError [%d]: %s", error.errno, error.strerror)
-        pywifi.ControlPeripheral(logger=logger).enable()  # Make sure Wi-Fi is enabled
+        # Make sure Wi-Fi is enabled
+        pywifi.ControlPeripheral(logger=logger).enable()
         connection_control = pywifi.ControlConnection(
             wifi_ssid=models.env.wifi_ssid,
             wifi_password=models.env.wifi_password,

jarvis/executors/controls.py

@@ -81,9 +81,8 @@ def exit_process() -> None:
             speaker.speak(
                 text=f"You have {len(reminders)} pending reminders {models.env.title}!"
             )
-        speaker.speak(
-            text=util.comma_separator(reminders)
-        )  # No need for string.capwords as speaker runs in a new loop
+        # No need for string.capwords as speaker runs in a new loop
+        speaker.speak(text=util.comma_separator(reminders))
     if alarms := alarm.get_alarm_state():
         if len(alarms) == 1:
             speaker.speak(text="You have a pending alarm at ")
@@ -173,7 +172,8 @@ def restart_control(phrase: str = None, quiet: bool = False) -> None:
                 speaker.speak(text="Restarting all background processes!")
                 return
             if avail := list(files.get_processes().keys()):
-                avail.remove("jarvis")  # cannot be restarted
+                # cannot be restarted
+                avail.remove("jarvis")
             else:
                 speaker.speak(
                     text="Unable to fetch background processes. Try specifying 'all'"
@@ -317,9 +317,8 @@ def delete_logs() -> None:
                 > models.env.log_retention
             ):
                 logger.debug("Deleting log file: %s", os.path.join(__path, file_))
-                os.remove(
-                    os.path.join(__path, file_)
-                )  # removes the file if it is older than log retention time
+                # removes the file if it is older than log retention time
+                os.remove(os.path.join(__path, file_))
 
 
 def delete_pycache() -> None:

jarvis/executors/crontab.py

@@ -9,9 +9,8 @@
 from jarvis.modules.logger import logger, multiprocessing_logger
 from jarvis.modules.models import models
 
-LOG_FILE = os.path.join(
-    "logs", "cron_%d-%m-%Y.log"
-)  # Used by api functions that run on cron schedule
+# Used by api functions that run on cron schedule
+LOG_FILE = os.path.join("logs", "cron_%d-%m-%Y.log")
 
 
 def executor(statement: str, log_file: str = None, process_name: str = None) -> None:

jarvis/executors/face.py

@@ -37,10 +37,10 @@ def detected_face() -> None:
         if not os.path.exists(os.path.join(TRAINING_DIR, phrase)):
             os.makedirs(os.path.join(TRAINING_DIR, phrase))
         img_name = f"{phrase}_{datetime.now().strftime('%B_%d_%Y_%I-%M_%p')}.jpg"  # adds datetime to image name
-        os.rename(FACE_DETECTION_TEMP_FILE, img_name)  # renames the files
-        shutil.move(
-            src=img_name, dst=os.path.join(TRAINING_DIR, phrase)
-        )  # move under TRAINING_DIR -> named directory
+        # renames the files
+        os.rename(FACE_DETECTION_TEMP_FILE, img_name)
+        # move under TRAINING_DIR -> named directory
+        shutil.move(src=img_name, dst=os.path.join(TRAINING_DIR, phrase))
         speaker.speak(
             text=f"Image has been added to known database. I will be able to recognize {phrase} in future."
         )

jarvis/executors/files.py

@@ -43,7 +43,7 @@ def put_frequent(data: Dict[str, int]) -> None:
     """
     with open(models.fileio.frequent, "w") as file:
         yaml.dump(data=data, stream=file, sort_keys=False)
-        file.flush()  # Write everything in buffer to file right away
+        file.flush()
 
 
 def get_location() -> DefaultDict[str, Dict | float | bool]:
@@ -89,7 +89,7 @@ def delete_secure_send(key: str) -> None:
         logger.critical("data for key [%s] was removed unprecedentedly", key)
     with open(models.fileio.secure_send, "w") as file:
         yaml.dump(data=current_data, stream=file, Dumper=yaml.Dumper)
-        file.flush()  # Write buffer to file immediately
+        file.flush()
 
 
 def put_secure_send(data: Dict[str, Dict[str, Any]]):
@@ -101,7 +101,7 @@ def put_secure_send(data: Dict[str, Dict[str, Any]]):
     existing = get_secure_send()
     with open(models.fileio.secure_send, "w") as file:
         yaml.dump(data={**existing, **data}, stream=file, Dumper=yaml.Dumper)
-        file.flush()  # Write buffer to file immediately
+        file.flush()
     logger.info(
         "Secure dict for [%s] will be cleared after 5 minutes",
         [*[*data.values()][0].keys()][0],
@@ -146,7 +146,7 @@ def put_restrictions(restrictions: List[str]) -> None:
     """
     with open(models.fileio.restrictions, "w") as file:
         yaml.dump(data=restrictions, stream=file, indent=2, sort_keys=False)
-        file.flush()  # Write buffer to file immediately
+        file.flush()
 
 
 def get_gpt_data() -> List[Dict[str, str]]:
@@ -171,7 +171,7 @@ def put_gpt_data(data: List[Dict[str, str]]) -> None:
     """
     with open(models.fileio.gpt_data, "w") as file:
         yaml.dump(data=data, stream=file, indent=4, Dumper=yaml.Dumper)
-        file.flush()  # Write buffer to file immediately
+        file.flush()
 
 
 def get_automation() -> Dict[str, List[Dict[str, str | bool]] | Dict[str, str | bool]]:
@@ -210,7 +210,7 @@ def put_automation(
 
     with open(models.fileio.automation, "w") as file:
         yaml.dump(data=sorted_data, stream=file, indent=2)
-        file.flush()  # Write buffer to file immediately
+        file.flush()
 
 
 def get_smart_devices() -> dict | bool | None:
@@ -243,7 +243,7 @@ def put_smart_devices(data: dict) -> None:
     """
     with open(models.fileio.smart_devices, "w") as file:
         yaml.dump(data=data, stream=file, indent=2, sort_keys=False)
-        file.flush()  # Write buffer to file immediately
+        file.flush()
 
 
 def get_processes() -> Dict[str, List[int | List[str]]]:
@@ -284,7 +284,7 @@ def put_reminders(data: List[Dict[str, str]]):
     """
     with open(models.fileio.reminders, "w") as file:
         yaml.dump(data=data, stream=file, indent=2, sort_keys=False)
-        file.flush()  # Write buffer to file immediately
+        file.flush()
 
 
 def get_alarms() -> List[Dict[str, str | bool]]:
@@ -310,7 +310,7 @@ def put_alarms(data: List[Dict[str, str | bool]]):
     """
     with open(models.fileio.alarms, "w") as file:
         yaml.dump(data=data, stream=file, indent=2, sort_keys=False)
-        file.flush()  # Write buffer to file immediately
+        file.flush()
 
 
 def get_recognizer() -> classes.RecognizerSettings:

jarvis/executors/ios_functions.py

@@ -146,9 +146,8 @@ def locate(phrase: str) -> None:
         return
     logger.info("Locating your %s", target_device)
     target_device.play_sound()
-    before_keyword, keyword, after_keyword = str(target_device).partition(
-        ":"
-    )  # partitions the hostname info
+    # partitions the hostname info
+    before_keyword, keyword, after_keyword = str(target_device).partition(":")
     if before_keyword == "Accessory":
         after_keyword = (
             after_keyword.replace(f"{models.env.name}’s", "")

jarvis/executors/lights.py

@@ -87,9 +87,8 @@ def avail_check(self, function_to_call: Callable) -> None:
         status = ThreadPool(processes=1).apply_async(
             func=self.thread_worker, args=(function_to_call,)
         )
-        speaker.speak(
-            run=True
-        )  # Speak the initial response when the work is happening behind the scenes
+        # Speak the initial response when the work is happening behind the scenes
+        speaker.speak(run=True)
         try:
             failed = status.get(5)
         except ThreadTimeoutError as error:

jarvis/executors/location.py

@@ -117,7 +117,7 @@ def distance(phrase) -> None:
     Args:
         phrase:Takes the phrase spoken as an argument.
     """
-    check = phrase.split()  # str to list
+    check = phrase.split()
     places = []
     for word in check:
         # looks for words that start with uppercase
@@ -195,9 +195,8 @@ def distance_controller(origin: str = None, destination: str = None) -> None:
         speaker.speak(text=f"I don't think {destination} exists {models.env.title}!")
         return
     if models.env.distance_unit == models.DistanceUnits.MILES:
-        dist = round(
-            geodesic(start, end).miles
-        )  # calculates miles from starting point to destination
+        # calculates miles from starting point to destination
+        dist = round(geodesic(start, end).miles)
     else:
         dist = round(geodesic(start, end).kilometers)
     if shared.called["directions"]: