diff --git a/src/assistant.py b/src/assistant.py index d36922c..0809cde 100755 --- a/src/assistant.py +++ b/src/assistant.py @@ -13,10 +13,16 @@ import subprocess from datetime import datetime -import commands -from infra import clear_screen -from utils import load_email_config -from voice_interface import VoiceInterface +from commands import scroller +from commands.basic_features import * +from commands.infra import clear_screen +from commands.modify_settings import * +from commands.news_reporter import fetch_news +from commands.send_email import send_email +from commands.utils import load_email_config +from commands.voice_interface import VoiceInterface +from commands.weather_reporter import weather_reporter +from commands.website_oppener import * LISTENING_ERROR = "Say that again please..." MAX_FETCHED_HEADLINES = ( @@ -69,19 +75,19 @@ def execute_query(self, query: str) -> None: print("No query detected. Please provide an input.") elif "what can you do" in query: - commands.explain_features(self.__voice_interface) + explain_features(self.__voice_interface) elif re.search(r"search .* (in google)?", query): # to convert to a generalized format query = query.replace(" in google", "") search_query = re.findall(r"search (.*)", query)[0] - commands.run_search_query(self.__voice_interface, search_query) + run_search_query(self.__voice_interface, search_query) elif "wikipedia" in query: # replace it only once to prevent changing the query query = query.replace("wikipedia", "", 1) search_query = query.replace("search", "", 1) - commands.wikipedia_search(self.__voice_interface, search_query, 3) + wikipedia_search(self.__voice_interface, search_query, 3) elif re.search("open .*", query): application = re.findall(r"open (.*)", query) @@ -90,7 +96,7 @@ def execute_query(self, query: str) -> None: return application = application[0] try: - commands.open_application_website(self.__voice_interface, application) + open_application_website(self.__voice_interface, application) except ValueError as ve: print( f"Error occurred while opening {application}: {ve.__class__.__name__}: {ve}" @@ -100,7 +106,7 @@ def execute_query(self, query: str) -> None: ) elif any(text in query for text in ["the time", "time please"]): - commands.tell_time(self.__voice_interface) + tell_time(self.__voice_interface) elif "scroll" in query: direction = re.search(r"(up|down|left|right|top|bottom)", query) @@ -114,27 +120,27 @@ def execute_query(self, query: str) -> None: self.__scrolling_thread is None ): # Only start if not already scrolling self.__scrolling_thread, self.__stop_scrolling_event = ( - commands.start_scrolling(direction) + scroller.start_scrolling(direction) ) elif "stop scrolling" in query: if self.__scrolling_thread is None: # Only stop if already scrolling return - commands.stop_scrolling( + scroller.stop_scrolling( self.__scrolling_thread, self.__stop_scrolling_event ) del self.__scrolling_thread self.__scrolling_thread = None elif re.search(r"scroll to (up|down|left|right|top|bottom)", query): - commands.scroll_to(direction) + scroller.scroll_to(direction) elif re.search(r"scroll (up|down|left|right)", query): - commands.simple_scroll(direction) + scroller.simple_scroll(direction) else: print("Scroll command not recognized") elif "weather" in query: cities = re.findall( r"\b(?:of|in|at)\s+(\w+)", query ) # Extract the city name just after the word 'of' - commands.weather_reporter(self.__voice_interface, cities[0]) + weather_reporter(self.__voice_interface, cities[0]) elif "email" in query: query = query.lower() @@ -191,7 +197,7 @@ def execute_query(self, query: str) -> None: response = self.listen_for_query() if "yes" in response.lower() or "sure" in response.lower(): self.__voice_interface.speak("Sending the email") - commands.send_email(self.__voice_interface, receiver, subject, body) + send_email(self.__voice_interface, receiver, subject, body) else: self.__voice_interface.speak("Request aborted by user") @@ -204,11 +210,11 @@ def execute_query(self, query: str) -> None: else: value = min(max(0, int(value[0])), 100) if "set" in query: - commands.brightness_control(value, False, False) + brightness_control(value, False, False) else: toDecrease = "decrease" in query or "reduce" in query relative = "by" in query - commands.brightness_control(value, relative, toDecrease) + brightness_control(value, relative, toDecrease) elif "volume" in query: query = query.lower() @@ -219,11 +225,11 @@ def execute_query(self, query: str) -> None: else: value = min(max(0, int(value[0])), 100) if "set" in query: - commands.volume_control(value, False, False) + volume_control(value, False, False) else: toDecrease = "decrease" in query or "reduce" in query relative = "by" in query - commands.volume_control(value, relative, toDecrease) + volume_control(value, relative, toDecrease) elif "shutdown" in query or "shut down" in query: self.__voice_interface.speak("Are you sure you want to shut down your PC?") @@ -247,7 +253,7 @@ def execute_query(self, query: str) -> None: else: self.__voice_interface.speak("Request aborted by user") elif "news" in query: - commands.fetch_news(self.__voice_interface, MAX_FETCHED_HEADLINES) + fetch_news(self.__voice_interface, MAX_FETCHED_HEADLINES) else: self.__voice_interface.speak("could not interpret the query") @@ -257,7 +263,7 @@ def close(self): self.__voice_interface.close() del self.__voice_interface if self.__scrolling_thread: - commands.stop_scrolling( + scroller.stop_scrolling( self.__scrolling_thread, self.__stop_scrolling_event ) del self.__scrolling_thread diff --git a/src/commands.py b/src/commands.py deleted file mode 100644 index 96b7321..0000000 --- a/src/commands.py +++ /dev/null @@ -1,490 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -Features -=============== - -This module contains all the functions pertaining to implementing the -individual features of the Assistant. - -""" - -import smtplib -import ssl -import subprocess -import threading -import time -from datetime import datetime -from email.message import EmailMessage -from subprocess import CalledProcessError, TimeoutExpired - -import feedparser -import googlesearch -import pyautogui as pag -import pygetwindow -import requests -import wikipedia -import wmi -from comtypes import CLSCTX_ALL -from dotenv import dotenv_values -from PIL import ImageGrab -from pycaw.pycaw import AudioUtilities, IAudioEndpointVolume - -from infra import __is_darwin, __is_posix, __is_windows, __system_os -from utils import load_email_config -from voice_interface import VoiceInterface - -SUPPORTED_FEATURES = { - "search your query in google and return upto 10 results", - "get a wikipedia search summary of upto 3 sentences", - "open applications or websites", - "tell you the time of the day", - "scroll the screen with active cursor", -} - -########## Conditional Imports ########## -if __is_windows(): - from AppOpener import open as open_app -########## Conditional Imports ########## - -ENVIRONMENT_VARIABLES = dotenv_values(".env") - - -def explain_features(vi: VoiceInterface) -> None: - """Explains the features available - - Args: - vi (VoiceInterface): The voice interface instance used to speak the text - """ - vi.speak("Here's what I can do...\n") - for feature in SUPPORTED_FEATURES: - vi.speak(f"--> {feature}") - - -def run_search_query(vi: VoiceInterface, search_query: str) -> None: - """Performs google search based on some terms - - Args: - vi (VoiceInterface): VoiceInterface instance used to speak - search_query (str): the query term to be searched in google - """ - if not search_query: - vi.speak("Invalid Google Search Query Found!!") - return - - results = googlesearch.search(term=search_query) - if not results: - vi.speak("No Search Result Found!!") - else: - results = list(results) - vi.speak("Found Following Results: ") - for i, result in enumerate(results): - print(i + 1, ")", result.title) - - -def wikipedia_search( - vi: VoiceInterface, search_query: str, sentence_count: int = 3 -) -> None: - """Searches wikipedia for the given query and returns fixed number of statements in response. - Disambiguation Error due to multiple similar results is handled. - Speaks the options in this case. - - Args: - vi (VoiceInterface): VoiceInterface instance used to speak. - search_query (str): The query term to search in wikipedia - sentence_count (int, optional): The number of sentences to speak in case of direct match. - Default is 3. - """ - try: - vi.speak("Searching Wikipedia...") - results = wikipedia.summary(search_query, sentences=sentence_count) - - vi.speak("According to wikipedia...") - vi.speak(results) - except wikipedia.DisambiguationError as de: - vi.speak(f"\n{de.__class__.__name__}") - options = str(de).split("\n") - if len(options) < 7: - for option in options: - vi.speak(option) - else: - for option in options[0:6]: - vi.speak(option) - vi.speak("... and more") - - -def open_application_website(vi: VoiceInterface, search_query: str) -> None: - """ - open the application/website using a matching path from AppPath/WebPath dictionaries. - - Args: - vi (VoiceInterface): VoiceInterface instance used to speak. - search_query (str): The website or application name - - Raises: - ValueError: Throws exception in case neither app nor web access-point is present. - """ - vi.speak(f"Attempting to open {search_query}...") - - search_query = search_query.strip().lower() - - # use appopener to open the application only if os is windows - if __is_windows(): - __open_application_website_windows(vi, search_query) - if __is_darwin(): - __open_application_website_darwin(vi, search_query) - elif __is_posix(): - __open_application_website_posix(vi, search_query) - else: - raise ValueError(f"Unsupported OS: {__system_os()}") - - -def __open_application_website_windows(vi: VoiceInterface, search_query: str) -> None: - """handle the opening of application/website for Windows OS - - Args: - vi (VoiceInterface): VoiceInterface instance used to speak. - search_query (str): The website or application name - - Raises: - ValueError: Throws exception in case neither app nor web access-point is present. - """ - try: - open_app(search_query, match_closest=True) # attempt to open as application - except Exception as error: - vi.speak(f"Error: {error}: Failed to open {search_query}") - - -def __open_application_website_darwin(vi: VoiceInterface, search_query: str) -> None: - """handle the opening of application/website for Darwin OS - - Args: - vi (VoiceInterface): VoiceInterface instance used to speak. - search_query (str): The website or application name - - Raises: - ValueError: Throws exception in case neither app nor web access-point is present. - """ - try: - subprocess.run( - ["open", "-a", search_query], capture_output=True, check=True - ) # attempt to open as application - except CalledProcessError: - try: - subprocess.run( - ["open", search_query], capture_output=True, check=True - ) # attempt to open as website - except CalledProcessError as error: - return_code = error.returncode - stdout_text = error.stdout.decode("utf-8") - stderr_text = error.stderr.decode("utf-8") - vi.speak( - f"Error: {error}: Failed to open {search_query}: error code {return_code}" - ) - if stdout_text: - print("stdout:", stdout_text) - if stderr_text: - print("stderr:", stderr_text) - except TimeoutExpired as error: - vi.speak(f"Error: {error}: Call to open {search_query} timed out.") - - except TimeoutExpired as error: - vi.speak(f"Error: {error}: Call to open {search_query} timed out.") - - -def __open_application_website_posix(vi: VoiceInterface, search_query: str) -> None: - """handle the opening of application/website for POSIX OS - - Args: - vi (VoiceInterface): VoiceInterface instance used to speak. - search_query (str): The website or application name - - Raises: - ValueError: Throws exception in case neither app nor web access-point is present. - """ - try: - subprocess.run( - ["xdg-open", search_query], capture_output=True, check=True - ) # attempt to open website/application - except CalledProcessError as error: - vi.speak( - f"Error: {error}: Failed to open {search_query}: error code {error.returncode}" - ) - stdout_text = error.stdout.decode("utf-8") - stderr_text = error.stderr.decode("utf-8") - if stdout_text: - print("stdout:", stdout_text) - if stderr_text: - print("stderr:", stderr_text) - - except TimeoutExpired as error: - vi.speak(f"Error: {error}: Call to open {search_query} timed out.") - - -def tell_time(vi: VoiceInterface) -> None: - """Tells the time of the day with timezone - - Args: - vi (VoiceInterface): Voice interface instance used to speak - """ - date_time = datetime.now() - hour, minute, second = date_time.hour, date_time.minute, date_time.second - tmz = date_time.tzname() - - vi.speak(f"Current time is {hour}:{minute}:{second} {tmz}") - - -def start_gradual_scroll(direction: str, stop_event: threading.Event) -> None: - """Gradually scroll in the given direction until stop_event is set.""" - active_window = pygetwindow.getActiveWindow() - if not active_window: - return - - # Capture a portion of the window to ensure scrolling - left, top, right, bottom = 0, 0, 100, 100 - width = right - left - height = bottom - top - previous_image = ImageGrab.grab(bbox=(left, top, left + width, top + height)) - while not stop_event.is_set(): - pag.scroll(clicks=1) - current_image = ImageGrab.grab(bbox=(left, top, left + width, top + height)) - - if current_image.getdata() == previous_image.getdata(): - print("Reached to extreme") - stop_event.set() - break - previous_image = current_image - - print(f"Stopped scrolling {direction}.") - - -def start_scrolling(direction: str) -> tuple[threading.Thread, threading.Event]: - """Start a new scroll thread.""" - stop_scrolling_event = threading.Event() - scrolling_thread = threading.Thread( - target=start_gradual_scroll, args=(direction, stop_scrolling_event) - ) - scrolling_thread.start() - return scrolling_thread, stop_scrolling_event - - -def stop_scrolling( - scrolling_thread: threading.Thread, scrolling_thread_event: threading.Event -) -> None: - """Stop the scrolling thread if not already stopped.""" - if scrolling_thread is not None: - scrolling_thread_event.set() - scrolling_thread.join() - - -def scroll_to(direction: str) -> None: - """Scroll to the extreme in the given direction.""" - active_window = pygetwindow.getActiveWindow() - if not active_window: - return - time.sleep(0.5) - if direction == "top": - pag.press("home") - elif direction == "bottom": - pag.press("end") - elif direction == "right": - pag.press("right", presses=9999) - elif direction == "left": - pag.press("left", presses=9999) - else: - print("Invalid Command") - - -def simple_scroll(direction: str) -> None: - """Simple scroll in the given direction by a fixed number of steps.""" - active_window = pygetwindow.getActiveWindow() - if not active_window: - return - time.sleep(0.5) - if direction in ["up", "down", "left", "right"]: - pag.press(keys=direction, presses=25) - else: - print("Invalid direction") - - -def brightness_control(value: int, relative: bool, toDecrease: bool): - """ - Adjusts the brightness of the monitor. - - Args: - value (int): The brightness level to set or adjust by. Should be between 0 and 100. - relative (bool): If True, the brightness change is relative to the current brightness. - If False, the brightness is set to the specified value. - toDecrease (bool): If True, decreases the brightness by the specified value. - If False, increases the brightness by the specified value. Only applicable when `relative` is True. - - Raises: - RuntimeError: If there is an issue with accessing the brightness control methods. - - Returns: - None - """ - - brightness_ctrl = wmi.WMI(namespace="root\\wmi") - methods = brightness_ctrl.WmiMonitorBrightnessMethods()[0] - - if relative: - current_brightness = brightness_ctrl.WmiMonitorBrightness()[0].CurrentBrightness - set_brightnes = ( - current_brightness - int(value) - if toDecrease - else current_brightness + int(value) - ) - methods.WmiSetBrightness(set_brightnes, 0) - else: - methods.WmiSetBrightness(value, 0) - - -def volume_control(value: int, relative: bool, toDecrease: bool): - """ - Adjusts the master volume of the system. - - Args: - value (int): The volume level to set or adjust by. Should be between 0 and 100. - relative (bool): If True, the volume change is relative to the current volume. - If False, the volume is set to the specified value. - toDecrease (bool): If True, decreases the volume by the specified value. - If False, increases the volume by the specified value. Only applicable when `relative` is True. - - Raises: - RuntimeError: If there is an issue with accessing the audio endpoint. - - Returns: - None - """ - - devices = AudioUtilities.GetSpeakers() - interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None) - volume = interface.QueryInterface(IAudioEndpointVolume) - - if relative: - current_volume = volume.GetMasterVolumeLevelScalar() * 100 - set_volume = ( - current_volume - int(value) if toDecrease else current_volume + int(value) - ) - print(set_volume) - volume.SetMasterVolumeLevelScalar(min(max(0, set_volume), 100) / 100, None) - else: - volume.SetMasterVolumeLevelScalar(min(max(0, value), 100) / 100, None) - - -def fetch_news(vi: VoiceInterface, max_fetched_headlines: int) -> None: - """ - Fetches and reads out the top 5 headlines from the Google News RSS feed. - - This function fetches news headlines from the Google News RSS feed (specific to India in English). - It then reads out the top 5 headlines using the provided VoiceInterface instance. If the feed fetch is successful, - it reads the headlines one by one. If the fetch fails, it informs the user that the news couldn't be fetched. - - Args: - vi (VoiceInterface): The VoiceInterface instance used to speak the news headlines. - - Raises: - requests.exceptions.RequestException: If there is an issue while fetching the RSS feed. - AttributeError: If the feed does not contain expected attributes or entries. - """ - - feed_url = "https://news.google.com/rss?hl=en-IN&gl=IN&ceid=IN:en" - - vi.speak("Fetching news from servers.") - feed = feedparser.parse(feed_url) - if feed.status == 200: - headlines_list = [] - for entry in feed.entries[:max_fetched_headlines]: - headlines_list.append((entry.title).split(" -")[0]) - vi.speak("Here are some recent news headlines.") - for headline in headlines_list: - vi.speak(headline) - else: - vi.speak("Failed to fetch the news.") - - -def weather_reporter(vi: VoiceInterface, city_name: str) -> None: - """ - Fetches and reports the weather conditions for a given city. - - This function retrieves the latitude and longitude of the specified city using the API Ninjas City API. - It then fetches the current weather data from the Open-Meteo API and reports the temperature, humidity, - apparent temperature, rain probability, cloud cover, and wind speed using the VoiceInterface instance. - - Args: - vi (VoiceInterface): The VoiceInterface instance used to speak the weather report. - city_name (str): The name of the city for which to fetch weather data. - - Raises: - requests.exceptions.RequestException: If there is an issue with the API request. - IndexError: If the city name is not found in the API response. - KeyError: If expected weather data fields are missing from the response. - """ - # Fetch latitude and longitude for the given city to be used by open-metro api - params = { - "name": city_name, - } - geo_codes = requests.get( - "https://api.api-ninjas.com/v1/city", - params=params, - headers={"origin": "https://www.api-ninjas.com"}, - ).json() - - # Fetch weather data from Open-Meteo using the obtained coordinates - weather_data_response = requests.get( - f'https://api.open-meteo.com/v1/forecast?latitude={geo_codes[0].get("latitude")}&longitude={geo_codes[0].get("longitude")}¤t=temperature_2m,relative_humidity_2m,apparent_temperature,rain,showers,cloud_cover,wind_speed_10m&forecast_days=1' - ).json() - - weather_data = weather_data_response.get("current") - weather_units = weather_data_response.get("current_units") - - vi.speak( - f"The current temperature in {city_name} is {weather_data.get('temperature_2m')}{weather_units.get('temperature_2m')}. " - f"However, due to a relative humidity of {weather_data.get('relative_humidity_2m')}{weather_units.get('relative_humidity_2m')}, " - f"it feels like {weather_data.get('apparent_temperature')}{weather_units.get('apparent_temperature')}." - ) - - if weather_data.get("rain") == 0: - vi.speak("The skies will be clear, with no chance of rain.") - else: - cloud_cover = weather_data.get("cloud_cover") - vi.speak( - f"The sky will be {cloud_cover}{weather_units.get('cloud_cover')} cloudy, " - f"and there's a predicted rainfall of {weather_data.get('rain')}{weather_units.get('rain')}." - ) - - vi.speak( - f"The wind speed is expected to be {weather_data.get('wind_speed_10m')}{weather_units.get('wind_speed_10m')}, " - "so plan accordingly." - ) - - -def send_email(vi: VoiceInterface, toEmail: str, subject: str, body: str): - """ - Send an email to the specified recipient. - - Args: - vi (VoiceInterface): VoiceInterface instance used to speak. - toEmail (str): The recipient's email address. - subject (str): The subject of the email. - body (str): The body content of the email. - - Raises: - ValueError: If any required parameters are missing or invalid. - """ - - data = load_email_config() - CONTEXT = ssl.create_default_context() - msg = EmailMessage() - msg["Subject"] = subject - msg["From"] = data.get("username") - msg["To"] = [toEmail] - msg.set_content(body) - server = smtplib.SMTP_SSL(data.get("server"), data.get("port"), context=CONTEXT) - server.login( - data.get("username"), ENVIRONMENT_VARIABLES.get("DESKTOP_ASSISTANT_SMTP_PWD") - ) - server.send_message(msg) - server.quit() - vi.speak(f"Email sent to {toEmail}") diff --git a/src/commands/__init__.py b/src/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/commands/basic_features.py b/src/commands/basic_features.py new file mode 100644 index 0000000..47f216a --- /dev/null +++ b/src/commands/basic_features.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Basic Features +=============== + +This module contains basic features for the desktop assistant, including: + +- Explaining supported features +- Running Google search queries +- Telling the current time +- Performing Wikipedia searches + +Functions: + explain_features(vi: VoiceInterface) -> None: + Explains the features available. + + run_search_query(vi: VoiceInterface, search_query: str) -> None: + Performs a Google search based on some terms. + + tell_time(vi: VoiceInterface) -> None: + Tells the current time of the day with timezone. + + wikipedia_search(vi: VoiceInterface, search_query: str, sentence_count: int = 3) -> None: + Searches Wikipedia for the given query and returns a fixed number of sentences in response. +""" + +from datetime import datetime + +import googlesearch +import wikipedia + +from .voice_interface import VoiceInterface + +SUPPORTED_FEATURES = { + "search your query in google and return upto 10 results", + "get a wikipedia search summary of upto 3 sentences", + "open applications or websites", + "tell you the time of the day", + "scroll the screen with active cursor", +} + + +def explain_features(vi: VoiceInterface) -> None: + """Explains the features available + + Args: + vi (VoiceInterface): The voice interface instance used to speak the text + """ + vi.speak("Here's what I can do...\n") + for feature in SUPPORTED_FEATURES: + vi.speak(f"--> {feature}") + + +def run_search_query(vi: VoiceInterface, search_query: str) -> None: + """Performs google search based on some terms + + Args: + vi (VoiceInterface): VoiceInterface instance used to speak + search_query (str): the query term to be searched in google + """ + if not search_query: + vi.speak("Invalid Google Search Query Found!!") + return + + results = googlesearch.search(term=search_query) + if not results: + vi.speak("No Search Result Found!!") + else: + results = list(results) + vi.speak("Found Following Results: ") + for i, result in enumerate(results): + print(i + 1, ")", result.title) + + +def tell_time(vi: VoiceInterface) -> None: + """Tells the time of the day with timezone + + Args: + vi (VoiceInterface): Voice interface instance used to speak + """ + date_time = datetime.now() + hour, minute, second = date_time.hour, date_time.minute, date_time.second + tmz = date_time.tzname() + + vi.speak(f"Current time is {hour}:{minute}:{second} {tmz}") + + +def wikipedia_search( + vi: VoiceInterface, search_query: str, sentence_count: int = 3 +) -> None: + """Searches wikipedia for the given query and returns fixed number of statements in response. + Disambiguation Error due to multiple similar results is handled. + Speaks the options in this case. + + Args: + vi (VoiceInterface): VoiceInterface instance used to speak. + search_query (str): The query term to search in wikipedia + sentence_count (int, optional): The number of sentences to speak in case of direct match. + Default is 3. + """ + try: + vi.speak("Searching Wikipedia...") + results = wikipedia.summary(search_query, sentences=sentence_count) + + vi.speak("According to wikipedia...") + vi.speak(results) + except wikipedia.DisambiguationError as de: + vi.speak(f"\n{de.__class__.__name__}") + options = str(de).split("\n") + if len(options) < 7: + for option in options: + vi.speak(option) + else: + for option in options[0:6]: + vi.speak(option) + vi.speak("... and more") diff --git a/src/config/email_config.json b/src/commands/config/email_config.json similarity index 100% rename from src/config/email_config.json rename to src/commands/config/email_config.json diff --git a/src/infra.py b/src/commands/infra.py similarity index 100% rename from src/infra.py rename to src/commands/infra.py diff --git a/src/commands/modify_settings.py b/src/commands/modify_settings.py new file mode 100644 index 0000000..34c6ae9 --- /dev/null +++ b/src/commands/modify_settings.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Modify Settings +=============== + +This module contains functions to adjust system settings such as brightness and volume. + +Functions: + brightness_control(value: int, relative: bool, toDecrease: bool) -> None: + Adjusts the brightness of the monitor. + + volume_control(value: int, relative: bool, toDecrease: bool) -> None: + Adjusts the master volume of the system. +""" + +import wmi +from comtypes import CLSCTX_ALL +from pycaw.pycaw import AudioUtilities, IAudioEndpointVolume + + +def brightness_control(value: int, relative: bool, toDecrease: bool): + """ + Adjusts the brightness of the monitor. + + Args: + value (int): The brightness level to set or adjust by. Should be between 0 and 100. + relative (bool): If True, the brightness change is relative to the current brightness. + If False, the brightness is set to the specified value. + toDecrease (bool): If True, decreases the brightness by the specified value. + If False, increases the brightness by the specified value. Only applicable when `relative` is True. + + Raises: + RuntimeError: If there is an issue with accessing the brightness control methods. + + Returns: + None + """ + + brightness_ctrl = wmi.WMI(namespace="root\\wmi") + methods = brightness_ctrl.WmiMonitorBrightnessMethods()[0] + + if relative: + current_brightness = brightness_ctrl.WmiMonitorBrightness()[0].CurrentBrightness + set_brightnes = ( + current_brightness - int(value) + if toDecrease + else current_brightness + int(value) + ) + methods.WmiSetBrightness(set_brightnes, 0) + else: + methods.WmiSetBrightness(value, 0) + + +def volume_control(value: int, relative: bool, toDecrease: bool): + """ + Adjusts the master volume of the system. + + Args: + value (int): The volume level to set or adjust by. Should be between 0 and 100. + relative (bool): If True, the volume change is relative to the current volume. + If False, the volume is set to the specified value. + toDecrease (bool): If True, decreases the volume by the specified value. + If False, increases the volume by the specified value. Only applicable when `relative` is True. + + Raises: + RuntimeError: If there is an issue with accessing the audio endpoint. + + Returns: + None + """ + + devices = AudioUtilities.GetSpeakers() + interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None) + volume = interface.QueryInterface(IAudioEndpointVolume) + + if relative: + current_volume = volume.GetMasterVolumeLevelScalar() * 100 + set_volume = ( + current_volume - int(value) if toDecrease else current_volume + int(value) + ) + print(set_volume) + volume.SetMasterVolumeLevelScalar(min(max(0, set_volume), 100) / 100, None) + else: + volume.SetMasterVolumeLevelScalar(min(max(0, value), 100) / 100, None) diff --git a/src/commands/news_reporter.py b/src/commands/news_reporter.py new file mode 100644 index 0000000..94eafc2 --- /dev/null +++ b/src/commands/news_reporter.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +News Reporter +=============== + +This module contains functions to fetch and read out the top news headlines. + +Functions: + fetch_news(vi: VoiceInterface, max_fetched_headlines: int) -> None: + Fetches and reads out the top headlines from the Google News RSS feed. + + Args: + vi (VoiceInterface): The VoiceInterface instance used to speak the news headlines. + max_fetched_headlines (int): The maximum number of headlines to fetch and read out. + + Raises: + requests.exceptions.RequestException: If there is an issue while fetching the RSS feed. + AttributeError: If the feed does not contain expected attributes or entries. +""" +import feedparser + +from .voice_interface import VoiceInterface + + +def fetch_news(vi: VoiceInterface, max_fetched_headlines: int) -> None: + """ + Fetches and reads out the top 5 headlines from the Google News RSS feed. + + This function fetches news headlines from the Google News RSS feed (specific to India in English). + It then reads out the top 5 headlines using the provided VoiceInterface instance. If the feed fetch is successful, + it reads the headlines one by one. If the fetch fails, it informs the user that the news couldn't be fetched. + + Args: + vi (VoiceInterface): The VoiceInterface instance used to speak the news headlines. + + Raises: + requests.exceptions.RequestException: If there is an issue while fetching the RSS feed. + AttributeError: If the feed does not contain expected attributes or entries. + """ + + feed_url = "https://news.google.com/rss?hl=en-IN&gl=IN&ceid=IN:en" + + vi.speak("Fetching news from servers.") + feed = feedparser.parse(feed_url) + if feed.status == 200: + headlines_list = [] + for entry in feed.entries[:max_fetched_headlines]: + headlines_list.append((entry.title).split(" -")[0]) + vi.speak("Here are some recent news headlines.") + for headline in headlines_list: + vi.speak(headline) + else: + vi.speak("Failed to fetch the news.") diff --git a/src/commands/scroller.py b/src/commands/scroller.py new file mode 100644 index 0000000..952b202 --- /dev/null +++ b/src/commands/scroller.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Scroller +=============== + +This module contains functions to handle scrolling actions for the desktop assistant. + +Functions: + start_gradual_scroll(direction: str, stop_event: threading.Event) -> None: + Gradually scrolls in the given direction until the stop_event is set. + + start_scrolling(direction: str) -> tuple[threading.Thread, threading.Event]: + Starts a new thread to handle gradual scrolling. + + stop_scrolling(scrolling_thread: threading.Thread, scrolling_thread_event: threading.Event) -> None: + Stops the scrolling thread if it is not already stopped. + + scroll_to(direction: str) -> None: + Scrolls to the extreme in the given direction. + + simple_scroll(direction: str) -> None: + Performs a simple scroll in the given direction by a fixed number of steps. +""" + +import threading +import time + +import pyautogui as pag +import pygetwindow +from PIL import ImageGrab + + +def start_gradual_scroll(direction: str, stop_event: threading.Event) -> None: + """Gradually scroll in the given direction until stop_event is set.""" + active_window = pygetwindow.getActiveWindow() + if not active_window: + return + + # Capture a portion of the window to ensure scrolling + left, top, right, bottom = 0, 0, 100, 100 + width = right - left + height = bottom - top + previous_image = ImageGrab.grab(bbox=(left, top, left + width, top + height)) + while not stop_event.is_set(): + pag.scroll(clicks=1) + current_image = ImageGrab.grab(bbox=(left, top, left + width, top + height)) + + if current_image.getdata() == previous_image.getdata(): + print("Reached to extreme") + stop_event.set() + break + previous_image = current_image + + print(f"Stopped scrolling {direction}.") + + +def start_scrolling(direction: str) -> tuple[threading.Thread, threading.Event]: + """Start a new scroll thread.""" + stop_scrolling_event = threading.Event() + scrolling_thread = threading.Thread( + target=start_gradual_scroll, args=(direction, stop_scrolling_event) + ) + scrolling_thread.start() + return scrolling_thread, stop_scrolling_event + + +def stop_scrolling( + scrolling_thread: threading.Thread, scrolling_thread_event: threading.Event +) -> None: + """Stop the scrolling thread if not already stopped.""" + if scrolling_thread is not None: + scrolling_thread_event.set() + scrolling_thread.join() + + +def scroll_to(direction: str) -> None: + """Scroll to the extreme in the given direction.""" + active_window = pygetwindow.getActiveWindow() + if not active_window: + return + time.sleep(0.5) + if direction == "top": + pag.press("home") + elif direction == "bottom": + pag.press("end") + elif direction == "right": + pag.press("right", presses=9999) + elif direction == "left": + pag.press("left", presses=9999) + else: + print("Invalid Command") + + +def simple_scroll(direction: str) -> None: + """Simple scroll in the given direction by a fixed number of steps.""" + active_window = pygetwindow.getActiveWindow() + if not active_window: + return + time.sleep(0.5) + if direction in ["up", "down", "left", "right"]: + pag.press(keys=direction, presses=25) + else: + print("Invalid direction") diff --git a/src/commands/send_email.py b/src/commands/send_email.py new file mode 100644 index 0000000..240de52 --- /dev/null +++ b/src/commands/send_email.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Send Email +=============== + +This module contains functions to send emails using SMTP. + +Functions: + send_email(vi: VoiceInterface, toEmail: str, subject: str, body: str) -> None: + Sends an email to the specified recipient. + + Args: + vi (VoiceInterface): VoiceInterface instance used to speak. + toEmail (str): The recipient's email address. + subject (str): The subject of the email. + body (str): The body content of the email. + + Raises: + ValueError: If any required parameters are missing or invalid. +""" + +import smtplib +import ssl +from email.message import EmailMessage + +from dotenv import dotenv_values + +from .utils import load_email_config +from .voice_interface import VoiceInterface + +ENVIRONMENT_VARIABLES = dotenv_values(".env") + + +def send_email(vi: VoiceInterface, toEmail: str, subject: str, body: str): + """ + Send an email to the specified recipient. + + Args: + vi (VoiceInterface): VoiceInterface instance used to speak. + toEmail (str): The recipient's email address. + subject (str): The subject of the email. + body (str): The body content of the email. + + Raises: + ValueError: If any required parameters are missing or invalid. + """ + + data = load_email_config() + CONTEXT = ssl.create_default_context() + msg = EmailMessage() + msg["Subject"] = subject + msg["From"] = data.get("username") + msg["To"] = [toEmail] + msg.set_content(body) + server = smtplib.SMTP_SSL(data.get("server"), data.get("port"), context=CONTEXT) + server.login( + data.get("username"), ENVIRONMENT_VARIABLES.get("DESKTOP_ASSISTANT_SMTP_PWD") + ) + server.send_message(msg) + server.quit() + vi.speak(f"Email sent to {toEmail}") diff --git a/src/commands/utils.py b/src/commands/utils.py new file mode 100644 index 0000000..93d7e37 --- /dev/null +++ b/src/commands/utils.py @@ -0,0 +1,27 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Utils +=============== + +This module contains utility functions for the desktop assistant. + +Functions: + load_email_config() -> dict: + Loads the email configuration from a JSON file. + + Returns: + dict: The email configuration data. +""" + +import json +import os + +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) + +CONFIG_PATH = os.path.join(BASE_DIR, "config", "email_config.json") + + +def load_email_config(): + with open(CONFIG_PATH, "r") as f: + return json.load(f) diff --git a/src/voice_interface.py b/src/commands/voice_interface.py similarity index 100% rename from src/voice_interface.py rename to src/commands/voice_interface.py diff --git a/src/commands/weather_reporter.py b/src/commands/weather_reporter.py new file mode 100644 index 0000000..72721b3 --- /dev/null +++ b/src/commands/weather_reporter.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Weather Reporter +=============== + +This module contains functions to fetch and report the weather conditions for a given city. + +Functions: + weather_reporter(vi: VoiceInterface, city_name: str) -> None: + Fetches and reports the weather conditions for a given city. + + Args: + vi (VoiceInterface): The VoiceInterface instance used to speak the weather report. + city_name (str): The name of the city for which to fetch weather data. + + Raises: + requests.exceptions.RequestException: If there is an issue with the API request. + IndexError: If the city name is not found in the API response. + KeyError: If expected weather data fields are missing from the response. +""" + +import requests + +from .voice_interface import VoiceInterface + + +def weather_reporter(vi: VoiceInterface, city_name: str) -> None: + """ + Fetches and reports the weather conditions for a given city. + + This function retrieves the latitude and longitude of the specified city using the API Ninjas City API. + It then fetches the current weather data from the Open-Meteo API and reports the temperature, humidity, + apparent temperature, rain probability, cloud cover, and wind speed using the VoiceInterface instance. + + Args: + vi (VoiceInterface): The VoiceInterface instance used to speak the weather report. + city_name (str): The name of the city for which to fetch weather data. + + Raises: + requests.exceptions.RequestException: If there is an issue with the API request. + IndexError: If the city name is not found in the API response. + KeyError: If expected weather data fields are missing from the response. + """ + # Fetch latitude and longitude for the given city to be used by open-metro api + params = { + "name": city_name, + } + geo_codes = requests.get( + "https://api.api-ninjas.com/v1/city", + params=params, + headers={"origin": "https://www.api-ninjas.com"}, + ).json() + + # Fetch weather data from Open-Meteo using the obtained coordinates + weather_data_response = requests.get( + f'https://api.open-meteo.com/v1/forecast?latitude={geo_codes[0].get("latitude")}&longitude={geo_codes[0].get("longitude")}¤t=temperature_2m,relative_humidity_2m,apparent_temperature,rain,showers,cloud_cover,wind_speed_10m&forecast_days=1' + ).json() + + weather_data = weather_data_response.get("current") + weather_units = weather_data_response.get("current_units") + + vi.speak( + f"The current temperature in {city_name} is {weather_data.get('temperature_2m')}{weather_units.get('temperature_2m')}. " + f"However, due to a relative humidity of {weather_data.get('relative_humidity_2m')}{weather_units.get('relative_humidity_2m')}, " + f"it feels like {weather_data.get('apparent_temperature')}{weather_units.get('apparent_temperature')}." + ) + + if weather_data.get("rain") == 0: + vi.speak("The skies will be clear, with no chance of rain.") + else: + cloud_cover = weather_data.get("cloud_cover") + vi.speak( + f"The sky will be {cloud_cover}{weather_units.get('cloud_cover')} cloudy, " + f"and there's a predicted rainfall of {weather_data.get('rain')}{weather_units.get('rain')}." + ) + + vi.speak( + f"The wind speed is expected to be {weather_data.get('wind_speed_10m')}{weather_units.get('wind_speed_10m')}, " + "so plan accordingly." + ) diff --git a/src/commands/website_oppener.py b/src/commands/website_oppener.py new file mode 100644 index 0000000..b7c1f1a --- /dev/null +++ b/src/commands/website_oppener.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Website Opener +=============== + +This module contains functions to open applications or websites based on the user's query. +It supports different operating systems including Windows, Darwin (macOS), and POSIX (Linux). + +Functions: + open_application_website(vi: VoiceInterface, search_query: str) -> None: + Opens the application or website using a matching path from AppPath/WebPath dictionaries. + + __open_application_website_windows(vi: VoiceInterface, search_query: str) -> None: + Handles the opening of application/website for Windows OS. + + __open_application_website_darwin(vi: VoiceInterface, search_query: str) -> None: + Handles the opening of application/website for Darwin OS. + + __open_application_website_posix(vi: VoiceInterface, search_query: str) -> None: + Handles the opening of application/website for POSIX OS. +""" + +import subprocess +from subprocess import CalledProcessError, TimeoutExpired + +from commands.infra import __is_darwin, __is_posix, __is_windows, __system_os +from commands.voice_interface import VoiceInterface + +if __is_windows(): + from AppOpener import open as open_app + + +def open_application_website(vi: VoiceInterface, search_query: str) -> None: + """ + open the application/website using a matching path from AppPath/WebPath dictionaries. + + Args: + vi (VoiceInterface): VoiceInterface instance used to speak. + search_query (str): The website or application name + + Raises: + ValueError: Throws exception in case neither app nor web access-point is present. + """ + vi.speak(f"Attempting to open {search_query}...") + + search_query = search_query.strip().lower() + + # use appopener to open the application only if os is windows + if __is_windows(): + __open_application_website_windows(vi, search_query) + if __is_darwin(): + __open_application_website_darwin(vi, search_query) + elif __is_posix(): + __open_application_website_posix(vi, search_query) + else: + raise ValueError(f"Unsupported OS: {__system_os()}") + + +def __open_application_website_windows(vi: VoiceInterface, search_query: str) -> None: + """handle the opening of application/website for Windows OS + + Args: + vi (VoiceInterface): VoiceInterface instance used to speak. + search_query (str): The website or application name + + Raises: + ValueError: Throws exception in case neither app nor web access-point is present. + """ + try: + open_app(search_query, match_closest=True) # attempt to open as application + except Exception as error: + vi.speak(f"Error: {error}: Failed to open {search_query}") + + +def __open_application_website_darwin(vi: VoiceInterface, search_query: str) -> None: + """handle the opening of application/website for Darwin OS + + Args: + vi (VoiceInterface): VoiceInterface instance used to speak. + search_query (str): The website or application name + + Raises: + ValueError: Throws exception in case neither app nor web access-point is present. + """ + try: + subprocess.run( + ["open", "-a", search_query], capture_output=True, check=True + ) # attempt to open as application + except CalledProcessError: + try: + subprocess.run( + ["open", search_query], capture_output=True, check=True + ) # attempt to open as website + except CalledProcessError as error: + return_code = error.returncode + stdout_text = error.stdout.decode("utf-8") + stderr_text = error.stderr.decode("utf-8") + vi.speak( + f"Error: {error}: Failed to open {search_query}: error code {return_code}" + ) + if stdout_text: + print("stdout:", stdout_text) + if stderr_text: + print("stderr:", stderr_text) + except TimeoutExpired as error: + vi.speak(f"Error: {error}: Call to open {search_query} timed out.") + + except TimeoutExpired as error: + vi.speak(f"Error: {error}: Call to open {search_query} timed out.") + + +def __open_application_website_posix(vi: VoiceInterface, search_query: str) -> None: + """handle the opening of application/website for POSIX OS + + Args: + vi (VoiceInterface): VoiceInterface instance used to speak. + search_query (str): The website or application name + + Raises: + ValueError: Throws exception in case neither app nor web access-point is present. + """ + try: + subprocess.run( + ["xdg-open", search_query], capture_output=True, check=True + ) # attempt to open website/application + except CalledProcessError as error: + vi.speak( + f"Error: {error}: Failed to open {search_query}: error code {error.returncode}" + ) + stdout_text = error.stdout.decode("utf-8") + stderr_text = error.stderr.decode("utf-8") + if stdout_text: + print("stdout:", stdout_text) + if stderr_text: + print("stderr:", stderr_text) + + except TimeoutExpired as error: + vi.speak(f"Error: {error}: Call to open {search_query} timed out.") diff --git a/src/utils.py b/src/utils.py deleted file mode 100644 index 09cbd27..0000000 --- a/src/utils.py +++ /dev/null @@ -1,11 +0,0 @@ -import json -import os - -BASE_DIR = os.path.dirname(os.path.abspath(__file__)) - -CONFIG_PATH = os.path.join(BASE_DIR, "config", "email_config.json") - - -def load_email_config(): - with open(CONFIG_PATH, "r") as f: - return json.load(f)