Welcome to Jarvis UI’s documentation!

Read Me:

Jarvis

jarvis_ui.main.initiator(status_manager: Optional[DictProxy] = None) → None

Starts main process to activate Jarvis and process requests via API calls.

Parameters:

status_manager – Shared multiprocessing dict to update in case of failed health check.

jarvis_ui.main.start() → None

Initiates Jarvis as a child process.

jarvis_ui.main.terminator(process: Process) → None

Terminates the process.

Parameters:

process – Takes the process object as an argument.

Logger

Initiates a custom logger to be accessed across modules.

>>> Logger

Disables loggers from imported modules, while using the root logger without having to load an external file.

jarvis_ui.logger.console_logger() → Logger

Create custom stream logger.

Returns:

Returns the logger object.

Return type:

Logger

jarvis_ui.logger.file_logger() → Logger

Create custom file logger.

Returns:

Returns the logger object.

Return type:

Logger

Executables

API Handler

Makes a post call to Jarvis API running in the backend to process a request and return the response.

>>> APIHandler
class jarvis_ui.executables.api_handler.BearerAuth(token: str)

Instantiates BearerAuth object.

>>> BearerAuth

References

New Forms of Authentication

Initializes the class and assign object members.

Parameters:

token – Token for bearer auth.

jarvis_ui.executables.api_handler.make_request(path: str, data: Optional[dict] = None, method: str = 'POST') → Union[dict, bool]

Makes a requests call to the API running on the backend to execute a said task.

Parameters:

  • data – Takes the command to be executed as an argument.

  • path – Path to make the api call.

  • method – HTTP methods, GET/POST.

Returns:

Returns the JSON response if request was successful.

Return type:

dict

Display

OnScreen display functions for the UI.

>>> Display
jarvis_ui.executables.display.flush_screen() → None

Flushes the screen output.

See also

  • Writes white spaces to the window size in a terminal.

  • Writes recursive empty text in IDE to flush screen.

jarvis_ui.executables.display.write_screen(text: Any) → None

Write text to a screen that can be cleared later.

Parameters:

text – Text to be written on screen..

Helper

jarvis_ui.executables.helper.extract_nos(input_: str, method: type = <class 'float'>) → Union[int, float]

Extracts number part from a string.

Parameters:

  • input_ – Takes string as an argument.

  • method – Takes a type to return a float or int value.

Returns:

Integer/float values from the phrase.

Return type:

Union[int, float]

jarvis_ui.executables.helper.heart_beat(status_manager: Optional[DictProxy] = None) → None

Initiate health check with the server.

Parameters:

status_manager – Shared multiprocessing dict to update in case of failed health check.

See also

  • Heart beat should be set no lesser than 5 seconds to avoid throttling and no longer than an hour.

  • Maintains a consecutive failure threshold of 5, as a single failed health check doesn’t warrant a restart.

jarvis_ui.executables.helper.linux_restart() → NoReturn

Restarts the base script on Linux OS.

See also

  • In Linux, it is not possible to trigger port audio on multiple processes.

  • To overcome this problem, JarvisUI on Linux is set to restart from self.

  • Since restarting executable triggers the base script, explicit reload of env vars are not required.

Raises:

- KeyboardInterrupt – To stop the current process to avoid recursion.

Listener

Module for listener and speech recognition.

>>> Listener
jarvis_ui.executables.listener.listen(timeout: Union[int, float] = 3.0, phrase_time_limit: Union[int, float] = 5) → Optional[str]

Function to activate listener and get the user input.

Parameters:

  • timeout – Time in seconds to wait for a phrase/sound to begin.

  • phrase_time_limit – Time in seconds to await user input. Anything spoken beyond this limit will be excluded.

Returns:

Returns the recognized statement listened via microphone.

Return type:

str

Processor

jarvis_ui.executables.processor.process(phrase: Optional[str] = None, status_manager: Optional[DictProxy] = None) → None

Handles request and response.

Parameters:

  • phrase – Takes existing phrase as an argument in case a previous failure is pending tobe addressed.

  • status_manager – Multiprocessing dictionary to set restarts.

jarvis_ui.executables.processor.process_request(phrase: str) → Optional[str]

Process request from the user.

Parameters:

phrase – Takes the phrase spoken as an argument.

Returns:

Returns the appropriate action to be taken.

Return type:

str

jarvis_ui.executables.processor.process_response(response: Union[dict, bool]) → None

Processes response from the server.

Parameters:

response – Takes either a boolean flag or a dictionary from the server as an argument.

Speaker

Module for speaker and voice options.

>>> Speaker
jarvis_ui.executables.speaker.speak(text: str) → None

Speak the received text using audio driver.

Parameters:

text – Takes the text that has to be spoken as an argument.

Starter

Module to kick off wake word detection.

>>> Starter
class jarvis_ui.executables.starter.Activator

Awaits for the keyword Jarvis and triggers initiator when heard.

>>> Activator

See also

  • Creates an input audio stream from a microphone, monitors it, and detects the specified wake word.

  • After processing the phrase, the converted text is sent as response to the API.

Initiates Porcupine object for hot word detection.

See also

  • Instantiates an instance of Porcupine object and monitors audio stream for occurrences of keywords.

  • A higher sensitivity results in fewer misses at the cost of increasing the false alarm rate.

  • sensitivity: Tolerance/Sensitivity level. Takes argument or env var sensitivity or defaults to 0.5

References

at_exit() → None

Invoked when the run loop is exited or manual interrupt.

See also

  • Releases resources held by porcupine.

  • Closes audio stream.

  • Releases port audio resources.

executor(status_manager: Optional[DictProxy] = None)

Closes the audio stream and calls the processor.

open_stream() → Stream

Initializes an audio stream.

Returns:

PyAudio stream.

Return type:

Stream

start(status_manager: Optional[DictProxy] = None) → None

Runs audio_stream in a forever loop and calls initiator when the phrase Jarvis is heard.

jarvis_ui.executables.starter.constructor() → Dict[str, Union[str, List[float], List[str]]]

Construct arguments for wake word detector.

Returns:

Arguments for wake word detector constructed as a dictionary based on the system and dependency version.

Return type:

Dict[str, Union[str, List[float], List[str]]]

Modules

Config

class jarvis_ui.modules.config.Config(BaseConfig)

Gets keywords during start up. Runs custom validations on env-vars.

>>> Config
Raises:

  • InvalidEnvVars –

  • If the voice name is not present for the OperatingSystem. –

Exceptions

exception jarvis_ui.modules.exceptions.DependencyError

Custom base exception for dependency errors.

>>> DependencyError
exception jarvis_ui.modules.exceptions.InvalidEnvVars

Custom InvalidEnvVars raised when invalid env vars are passed.

>>> InvalidEnvVars
exception jarvis_ui.modules.exceptions.UnsupportedOS

Custom OSError raised when initiated in an unsupported operating system.

>>> UnsupportedOS
jarvis_ui.modules.exceptions.no_alsa_err() → Iterable

Wrapper to suppress ALSA error messages when PyAudio module is called.

Notes

  • This happens specifically for Linux based operating systems.

  • There are usually multiple sound APIs to choose from but not all of them might be configured correctly.

  • PyAudio goes through “ALSA”, “PulseAudio” and “Jack” looking for audio hardware and that triggers warnings.

  • None of the options below seemed to work in all given conditions, so the approach taken was to hide them.

Options:

  • Comment off the ALSA devices where the error is triggered.

  • Set energy threshold to the output from python -m speech_recognition

  • Setting dynamic energy threshold to True

References

jarvis_ui.modules.exceptions.py_error_handler(filename: ByteString, line: int, function: ByteString, err: int, fmt: ByteString) → None

Handles errors from pyaudio module especially for Linux based operating systems.

Models

class jarvis_ui.modules.models.EnvConfig(BaseSettings)

Configure all env vars and validate using pydantic to share across modules.

>>> EnvConfig

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

__init__ uses __pydantic_self__ instead of the more common self for the first arg to allow self as a field name.

class Config

Environment variables configuration.

env_file = '.env'
env_prefix = ''
debug: bool
heart_beat: Optional[int]
listener_phrase_limit: Union[float, int]
listener_timeout: Union[float, int]
microphone_index: Optional[Union[int, int]]
model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'env_file': '.env', 'env_file_encoding': None, 'env_nested_delimiter': None, 'env_prefix': '', 'extra': 'forbid', 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'debug': FieldInfo(annotation=bool, required=False, default=False), 'heart_beat': FieldInfo(annotation=Union[int, NoneType], required=False, metadata=[Ge(ge=5), Le(le=3600)]), 'listener_phrase_limit': FieldInfo(annotation=Union[float, Annotated[int, Gt(gt=0)]], required=False, default=5), 'listener_timeout': FieldInfo(annotation=Union[float, Annotated[int, Gt(gt=0)]], required=False, default=2), 'microphone_index': FieldInfo(annotation=Union[int, Annotated[int, Gt(gt=0)], NoneType], required=False, metadata=[Ge(ge=0)]), 'native_audio': FieldInfo(annotation=bool, required=False, default=False), 'porcupine_key': FieldInfo(annotation=Union[str, NoneType], required=False), 'recognizer_settings': FieldInfo(annotation=RecognizerSettings, required=False, default=RecognizerSettings(energy_threshold=1100, pause_threshold=1, phrase_threshold=0.1, dynamic_energy_threshold=False, non_speaking_duration=1)), 'sensitivity': FieldInfo(annotation=Union[Annotated[int, Gt(gt=0)], Annotated[float, Gt(gt=0)], List[Annotated[int, Gt(gt=0)]], List[Annotated[float, Gt(gt=0)]]], required=False, default=0.5, metadata=[Ge(ge=0), Le(le=1)]), 'server_host': FieldInfo(annotation=Union[str, NoneType], required=False), 'server_ip': FieldInfo(annotation=Union[IPv4Address, NoneType], required=False), 'server_port': FieldInfo(annotation=Union[Annotated[int, Gt(gt=0)], NoneType], required=False), 'server_url': FieldInfo(annotation=Union[Annotated[pydantic_core._pydantic_core.Url, UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'], host_required=None, default_host=None, default_port=None, default_path=None)], NoneType], required=False), 'speech_timeout': FieldInfo(annotation=Union[int, Annotated[float, Gt(gt=0)], Annotated[int, Gt(gt=0)]], required=False, default=0), 'token': FieldInfo(annotation=str, required=True), 'voice_name': FieldInfo(annotation=Union[str, NoneType], required=False), 'voice_pitch': FieldInfo(annotation=Union[Annotated[int, Gt(gt=0)], Annotated[float, Gt(gt=0)], NoneType], required=False), 'voice_rate': FieldInfo(annotation=Union[Annotated[int, Gt(gt=0)], Annotated[float, Gt(gt=0)], NoneType], required=False), 'volume': FieldInfo(annotation=int, required=False, default=70, metadata=[Gt(gt=0)]), 'wake_words': FieldInfo(annotation=List[str], required=False, default=['jarvis'])}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

native_audio: bool
classmethod parse_microphone_index(idx: int) → Optional[int]

Validate microphone index.

classmethod parse_server_url(url: Optional[Url]) → Optional[str]

Validate server_url and return as string.

porcupine_key: Optional[str]
recognizer_settings: RecognizerSettings
sensitivity: Union[int, float, List[int], List[float]]
server_host: Optional[str]
server_ip: Optional[IPv4Address]
server_port: Optional[int]
server_url: Optional[Url]
speech_timeout: Union[int, float, int]
token: str
voice_name: Optional[str]
voice_pitch: Optional[Union[int, float]]
voice_rate: Optional[Union[int, float]]
volume: int
wake_words: List[str]
class jarvis_ui.modules.models.FileIO(BaseSettings)

Loads all the mp3 files’ path and log file path required by Jarvis.

>>> FileIO

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

__init__ uses __pydantic_self__ instead of the more common self for the first arg to allow self as a field name.

acknowledgement: Path
base_log_file: Union[Path, str]
connection_restart: Path
extn_: Dict[str, str]
failed: Path
model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'env_file': None, 'env_file_encoding': None, 'env_nested_delimiter': None, 'env_prefix': '', 'extra': 'forbid', 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'acknowledgement': FieldInfo(annotation=Path, required=False, default='/Users/vicky/Desktop/git/Jarvis_UI/jarvis_ui/indicators/acknowledgement.wav', metadata=[PathType(path_type='file')]), 'base_log_file': FieldInfo(annotation=Union[Annotated[pathlib.Path, PathType(path_type='file')], str], required=False, default='logs/jarvis_03-08-2024.log'), 'connection_restart': FieldInfo(annotation=Path, required=False, default='/Users/vicky/Desktop/git/Jarvis_UI/jarvis_ui/indicators/connection_restart_mac.wav', metadata=[PathType(path_type='file')]), 'extn_': FieldInfo(annotation=Dict[str, str], required=False, default={'Darwin': 'mac', 'Windows': 'win', 'Linux': 'ss'}), 'failed': FieldInfo(annotation=Path, required=False, default='/Users/vicky/Desktop/git/Jarvis_UI/jarvis_ui/indicators/failed_mac.wav', metadata=[PathType(path_type='file')]), 'path': FieldInfo(annotation=str, required=False, default='/Users/vicky/Desktop/git/Jarvis_UI/jarvis_ui/indicators'), 'restart': FieldInfo(annotation=Path, required=False, default='/Users/vicky/Desktop/git/Jarvis_UI/jarvis_ui/indicators/restart_mac.wav', metadata=[PathType(path_type='file')]), 'shutdown': FieldInfo(annotation=Path, required=False, default='/Users/vicky/Desktop/git/Jarvis_UI/jarvis_ui/indicators/shutdown_mac.wav', metadata=[PathType(path_type='file')]), 'speech_wav_file': FieldInfo(annotation=Union[Annotated[pathlib.Path, PathType(path_type='file')], str], required=False, default='/Users/vicky/Desktop/git/Jarvis_UI/jarvis_ui/indicators/speech-synthesis.wav')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

path: str
restart: Path
shutdown: Path
speech_wav_file: Union[Path, str]
class jarvis_ui.modules.models.Settings(BaseSettings)

Loads most common system values.

>>> Settings
Raises:

  • UnsupportedOS –

  • If the host operating system is other than Linux, macOS or Windows. –

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

__init__ uses __pydantic_self__ instead of the more common self for the first arg to allow self as a field name.

interactive: bool
legacy: bool
model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'env_file': None, 'env_file_encoding': None, 'env_nested_delimiter': None, 'env_prefix': '', 'extra': 'forbid', 'protected_namespaces': ('model_', 'settings_'), 'secrets_dir': None, 'validate_default': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'interactive': FieldInfo(annotation=bool, required=False, default=False), 'legacy': FieldInfo(annotation=bool, required=False, default=False), 'operating_system': FieldInfo(annotation=str, required=False, default='Darwin')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

operating_system: str
class jarvis_ui.modules.models.Sensitivity(float or PositiveInt, Enum)

Allowed values for sensitivity.

>>> Sensitivity
sensitivity: Union[float, int]

Peripherals

class jarvis_ui.modules.peripherals.ChannelType(value)

Allowed values for channel types.

>>> ChannelType
input_channels: str = 'maxInputChannels'
output_channels: str = 'maxOutputChannels'
jarvis_ui.modules.peripherals.channel_type

alias of ChannelType

jarvis_ui.modules.peripherals.get_audio_devices(channels: str) → Iterable[Dict[str, Union[str, int, float]]]

Iterates over all devices and yields the device that has input channels.

Parameters:

channels – Takes an argument to determine whether to yield input or output channels.

Yields:

Iterable – Yields a dictionary with all the input devices available.

Repeated Timer

class jarvis_ui.modules.timer.RepeatedTimer(interval: int, function: Callable, args: Optional[Tuple] = None, kwargs: Optional[Dict[str, Any]] = None)

Instantiates RepeatedTimer object to kick off the threading.Timer object with custom intervals.

>>> RepeatedTimer

Repeats the Timer object from threading.

Parameters:

  • interval – Interval in seconds.

  • function – Function to trigger with intervals.

  • args – Arguments for the function.

  • kwargs – Keyword arguments for the function.

cancel()

Initiate cancellation.

start()

Trigger target function if timer isn’t running already.

stop()

Stop the timer and cancel all futures.

Indices and tables