2
\$\begingroup\$

I am working since a couple of weeks on my first Python application and I would love to have a review of the code I wrote so far.

The app has the purpose of controlling the cooling and lighting of an AIO water cooling kit from NZXT on Linux. The official software only supports Windows but the interface is USB so the community wrote some really nice CLI tools to control it on any OS. My goal is to write a GUI front-end for it.

This means that unfortunately to use the app you need to have both a GNU/Linux installation and a NZXT Kraken. If you can't run the app but want to see how it looks, you can check this video or watch this gif:

enter image description here

The app I think is already in a good stage: It can monitor the status (water temp and RPM of fan and pump) and I can set some predefined profile.

I develop applications for Android since 2010 so I tried to apply my mobile dev knowledge to this project.

I choose to support Python 3.6+ and I pick GTK as graphical toolkit because my main target is GNU/Linux.

For the architecture of the app I tried to go for some kind of MVP+Clean, but I am not super strict on respecting it.

I am using injector to handle the dependency injection, rx for the streams and async tasks, peewee for the persistency of the data and matplotlib to plot the charts. The communication with the kit is done via liquidctl.

As I said I am not a Python developer and this is my first app that goes somehow public (even if in early stage of development) so I would really appreciate if I could get some corrections and tips from more experienced Python devs.

I really like static analysis on Android so I am trying to use it also for Python. So far I am using PyLint and Mypy.

The entire project source is available here, but I am going to paste the full code on this post since I think is requested by the rules.


main.py

#!/usr/bin/env python3
import signal
import locale
import gettext
import logging
import sys
import gi
from os.path import abspath, join, dirname
from peewee import SqliteDatabase
from rx.disposables import CompositeDisposable

gi.require_version('Gtk', '3.0')
from gi.repository import GLib
from gkraken.repository import KrakenRepository
from gkraken.di import INJECTOR
from gkraken.app import Application

APP = "gkraken"
WHERE_AM_I = abspath(dirname(__file__))
LOCALE_DIR = join(WHERE_AM_I, 'mo')

FORMAT = '%(filename)15s:%(lineno)-4d %(asctime)-15s: %(levelname)s/%(threadName)s(%(process)d) %(message)s'
logging.basicConfig(level=logging.DEBUG, format=FORMAT)
logging.getLogger("Rx").setLevel(logging.INFO)
logging.getLogger('injector').setLevel(logging.INFO)
logging.getLogger('peewee').setLevel(logging.INFO)
logging.getLogger('matplotlib').setLevel(logging.INFO)

LOG = logging.getLogger(__name__)

# POSIX locale settings
locale.setlocale(locale.LC_ALL, locale.getlocale())
locale.bindtextdomain(APP, LOCALE_DIR)

gettext.bindtextdomain(APP, LOCALE_DIR)
gettext.textdomain(APP)


def __cleanup() -> None:
    LOG.debug("cleanup")
    composite_disposable: CompositeDisposable = INJECTOR.get(CompositeDisposable)
    composite_disposable.dispose()
    database = INJECTOR.get(SqliteDatabase)
    database.close()
    kraken_repository = INJECTOR.get(KrakenRepository)
    kraken_repository.cleanup()
    # futures.thread._threads_queues.clear()


def handle_exception(exc_type, exc_value, exc_traceback) -> None:
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return

    LOG.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
    __cleanup()
    sys.exit(1)


sys.excepthook = handle_exception


def main() -> int:
    LOG.debug("main")
    application: Application = INJECTOR.get(Application)
    GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, application.quit)
    exit_status = application.run(sys.argv)
    __cleanup()
    return sys.exit(exit_status)


if __name__ == "__main__":
    main()

app.py

import logging
from gettext import gettext as _
from typing import Any, Optional

from gi.repository import Gtk, Gio, GLib
from injector import inject
from peewee import SqliteDatabase

from gkraken.conf import APP_NAME, APP_ID
from gkraken.model import SpeedProfile, SpeedStep, Setting, CurrentSpeedProfile
from gkraken.presenter import Presenter
from gkraken.util import load_db_default_data
from gkraken.view import View

LOG = logging.getLogger(__name__)


class Application(Gtk.Application):
    @inject
    def __init__(self,
                 database: SqliteDatabase,
                 view: View,
                 presenter: Presenter,
                 builder: Gtk.Builder,
                 *args: Any,
                 **kwargs: Any) -> None:
        LOG.debug("init Application")
        GLib.set_application_name(_(APP_NAME))
        super().__init__(*args, application_id=APP_ID,
                         flags=Gio.ApplicationFlags.HANDLES_COMMAND_LINE,
                         **kwargs)

        database.connect()
        database.create_tables([SpeedProfile, SpeedStep, CurrentSpeedProfile, Setting])

        if SpeedProfile.select().count() == 0:
            load_db_default_data()

        self.add_main_option("test", ord("t"), GLib.OptionFlags.NONE,
                             GLib.OptionArg.NONE, "Command line test", None)
        self.__view = view
        self.__presenter = presenter
        self.__presenter.application_quit = self.quit
        self.__window: Optional[Gtk.ApplicationWindow] = None
        self.__builder: Gtk.Builder = builder

    def do_activate(self) -> None:
        if not self.__window:
            self.__builder.connect_signals(self.__presenter)
            self.__window: Gtk.ApplicationWindow = self.__builder.get_object("application_window")
            self.__window.set_application(self)
            self.__window.show_all()
            self.__view.show()
        self.__window.present()

    def do_startup(self) -> None:
        Gtk.Application.do_startup(self)

    def do_command_line(self, command_line: Gio.ApplicationCommandLine) -> int:
        options = command_line.get_options_dict()
        # convert GVariantDict -> GVariant -> dict
        options = options.end().unpack()

        if "test" in options:
            # This is printed on the main instance
            print("Test argument recieved: %s" % options["test"])

        self.activate()
        return 0

view.py

import gi
import logging
from typing import Optional, Dict, Any, List, Tuple
from injector import inject, singleton
from gi.repository import Gtk

# AppIndicator3 may not be installed
from gkraken.interactor import SettingsInteractor

try:
    gi.require_version('AppIndicator3', '0.1')
    from gi.repository import AppIndicator3
except (ImportError, ValueError):
    AppIndicator3 = None
from matplotlib.axes import Axes
from matplotlib.figure import Figure
from matplotlib.backends.backend_gtk3agg import FigureCanvasGTK3Agg as FigureCanvas

from gkraken.conf import APP_PACKAGE_NAME, APP_ID
from gkraken.model import Status, SpeedProfile, ChannelType
from gkraken.presenter import Presenter, ViewInterface

LOG = logging.getLogger(__name__)
if AppIndicator3 is None:
    LOG.warning("AppIndicator3 is not installed. The app indicator will not be shown.")


# pylint: disable=too-many-instance-attributes
@singleton
class View(ViewInterface):

    @inject
    def __init__(self,
                 presenter: Presenter,
                 builder: Gtk.Builder,
                 settings_interactor: SettingsInteractor,
                 ) -> None:
        LOG.debug('init View')
        self.__presenter: Presenter = presenter
        self.__presenter.view = self
        self.__builder: Gtk.Builder = builder
        self.__settings_interactor = settings_interactor
        self.__init_widgets()

    def __init_widgets(self) -> None:
        # self.refresh_content_header_bar_title()
        self.__window = self.__builder.get_object("application_window")
        self.__settings_dialog: Gtk.Dialog = self.__builder.get_object("settings_dialog")
        self.__app_indicator_menu = self.__builder.get_object("app_indicator_menu")
        self.__statusbar: Gtk.Statusbar = self.__builder.get_object('statusbar')
        self.__context = self.__statusbar.get_context_id(APP_PACKAGE_NAME)
        self.__cooling_fan_speed: Gtk.Label = self.__builder.get_object('cooling_fan_speed')
        self.__cooling_fan_rpm: Gtk.Label = self.__builder.get_object('cooling_fan_rpm')
        self.__cooling_liquid_temp: Gtk.Label = self.__builder.get_object('cooling_liquid_temp')
        self.__cooling_pump_rpm: Gtk.Label = self.__builder.get_object('cooling_pump_rpm')
        self.__firmware_version: Gtk.Label = self.__builder.get_object('firmware_version')
        self.__cooling_fan_combobox: Gtk.ComboBox = self.__builder.get_object('cooling_fan_profile_combobox')
        self.__cooling_fan_liststore: Gtk.ListStore = self.__builder.get_object('cooling_fan_profile_liststore')
        self.__cooling_pump_combobox: Gtk.ComboBox = self.__builder.get_object('cooling_pump_profile_combobox')
        self.__cooling_pump_liststore: Gtk.ListStore = self.__builder.get_object('cooling_pump_profile_liststore')
        cooling_fan_scrolled_window: Gtk.ScrolledWindow = self.__builder.get_object('cooling_fan_scrolled_window')
        cooling_pump_scrolled_window: Gtk.ScrolledWindow = self.__builder.get_object('cooling_pump_scrolled_window')
        self.__cooling_fan_apply_button: Gtk.Button = self.__builder.get_object('cooling_fan_apply_button')
        self.__cooling_pump_apply_button: Gtk.Button = self.__builder.get_object('cooling_pump_apply_button')
        self.__init_plot_charts(cooling_fan_scrolled_window, cooling_pump_scrolled_window)

    def show(self) -> None:
        self.__presenter.on_start()
        self.__init_app_indicator()

    def __init_app_indicator(self) -> None:
        self.__app_indicator: Optional[AppIndicator3.Indicator] = None
        if AppIndicator3:
            icon_theme = Gtk.IconTheme.get_default()
            icon_name = icon_theme.lookup_icon('weather-showers-symbolic', 16, 0).get_filename()
            self.__app_indicator = AppIndicator3.Indicator \
                .new(APP_ID, icon_name, AppIndicator3.IndicatorCategory.HARDWARE)
            self.__app_indicator.set_status(AppIndicator3.IndicatorStatus.PASSIVE)
            self.__app_indicator.set_menu(self.__app_indicator_menu)

    def toggle_window_visibility(self) -> None:
        if self.__window.props.visible:
            self.__window.hide()
        else:
            self.__window.show()

    def show_add_speed_profile_dialog(self, channel: ChannelType) -> None:
        LOG.debug("view show_add_speed_profile_dialog %s", channel.name)

    def show_settings_dialog(self) -> None:
        self.__settings_dialog.show()

    def hide_settings_dialog(self) -> None:
        self.__settings_dialog.hide()

    def set_statusbar_text(self, text: str) -> None:
        self.__statusbar.remove_all(self.__context)
        self.__statusbar.push(self.__context, text)

    def refresh_content_header_bar_title(self) -> None:
        #     contant_stack = self.__builder.get_object("content_stack")
        #     child = contant_stack.get_visible_child()
        #     if child is not None:
        #         title = contant_stack.child_get_property(child, "title")
        #         content_header_bar = self.__builder.get_object("content_header_bar")
        #         content_header_bar.set_title(title)
        pass

    def refresh_status(self, status: Optional[Status]) -> None:
        LOG.debug('view status')
        if status:
            self.__cooling_fan_rpm.set_markup("<span size=\"xx-large\">%s</span> RPM" % status.fan_rpm)
            self.__cooling_fan_speed.set_markup("<span size=\"xx-large\">%s</span> %%" %
                                                ('-' if status.fan_speed is None else status.fan_speed))
            self.__cooling_liquid_temp.set_markup("<span size=\"xx-large\">%s</span> °C" % status.liquid_temperature)
            self.__cooling_pump_rpm.set_markup("<span size=\"xx-large\">%s</span> RPM" % status.pump_rpm)
            self.__firmware_version.set_label("firmware version %s" % status.firmware_version)
            if self.__app_indicator:
                if self.__settings_interactor.get_bool('settings_show_app_indicator'):
                    self.__app_indicator.set_status(AppIndicator3.IndicatorStatus.ACTIVE)
                else:
                    self.__app_indicator.set_status(AppIndicator3.IndicatorStatus.PASSIVE)
                if self.__settings_interactor.get_bool('settings_app_indicator_show_water_temp'):
                    self.__app_indicator.set_label("  %s°C" % status.liquid_temperature, "  XX°C")
                else:
                    self.__app_indicator.set_label("", "")

    def refresh_chart(self, profile: SpeedProfile) -> None:
        if profile.channel == ChannelType.FAN.value:
            self.__plot_fan_chart(self.__get_speed_profile_data(profile))
        elif profile.channel == ChannelType.PUMP.value:
            self.__plot_pump_chart(self.__get_speed_profile_data(profile))
        else:
            raise ValueError("Unknown channel: %s" % profile.channel)

    @staticmethod
    def __get_speed_profile_data(profile: SpeedProfile) -> Dict[int, int]:
        data = {p.temperature: p.duty for p in profile.steps}
        if profile.single_step:
            data.update({60: profile.steps[0].duty})
        else:
            data.update({60: 100})
        return data

    def refresh_profile_combobox(self, channel: ChannelType, data: List[Tuple[int, str]],
                                 active: Optional[int]) -> None:
        if channel is ChannelType.FAN:
            for item in data:
                self.__cooling_fan_liststore.append([item[0], item[1]])
            self.__cooling_fan_combobox.set_model(self.__cooling_fan_liststore)
            self.__cooling_fan_combobox.set_sensitive(len(self.__cooling_fan_liststore) > 1)
            if active is not None:
                self.__cooling_fan_combobox.set_active(active)
        elif channel is ChannelType.PUMP:
            for item in data:
                self.__cooling_pump_liststore.append([item[0], item[1]])
            self.__cooling_pump_combobox.set_model(self.__cooling_pump_liststore)
            self.__cooling_pump_combobox.set_sensitive(len(self.__cooling_pump_liststore) > 1)
            if active is not None:
                self.__cooling_pump_combobox.set_active(active)
        else:
            raise ValueError("Unknown channel: %s" % channel.name)

    def set_apply_button_enabled(self, channel: ChannelType, enabled: bool) -> None:
        if channel is ChannelType.FAN:
            self.__cooling_fan_apply_button.set_sensitive(enabled)
        elif channel is ChannelType.PUMP:
            self.__cooling_pump_apply_button.set_sensitive(enabled)
        else:
            raise ValueError("Unknown channel: %s" % channel.name)

    # pylint: disable=attribute-defined-outside-init
    def __init_plot_charts(self,
                           fan_scrolled_window: Gtk.ScrolledWindow,
                           pump_scrolled_window: Gtk.ScrolledWindow) -> None:
        self.__fan_figure = Figure(figsize=(8, 6), dpi=72, facecolor='#00000000')
        self.__fan_canvas = FigureCanvas(self.__fan_figure)  # a Gtk.DrawingArea+
        self.__fan_axis = self.__fan_figure.add_subplot(111)
        self.__fan_line, = self.__init_plot_chart(
            fan_scrolled_window,
            self.__fan_figure,
            self.__fan_canvas,
            self.__fan_axis
        )

        self.__pump_figure = Figure(figsize=(8, 6), dpi=72, facecolor='#00000000')
        self.__pump_canvas = FigureCanvas(self.__pump_figure)  # a Gtk.DrawingArea+
        self.__pump_axis = self.__pump_figure.add_subplot(111)
        self.__pump_line, = self.__init_plot_chart(
            pump_scrolled_window,
            self.__pump_figure,
            self.__pump_canvas,
            self.__pump_axis
        )

    def refresh_settings(self, settings: Dict[str, Any]) -> None:
        for key, value in settings.items():
            if isinstance(value, bool):
                switch: Gtk.Switch = self.__builder.get_object(key + '_switch')
                switch.set_active(value)

    @staticmethod
    def __init_plot_chart(fan_scrolled_window: Gtk.ScrolledWindow,
                          figure: Figure,
                          canvas: FigureCanvas,
                          axis: Axes) -> Any:
        axis.grid(True, linestyle=':')
        axis.margins(x=0, y=0.05)
        axis.set_facecolor('#00000000')
        axis.set_xlabel('Liquid temperature [°C]')
        axis.set_ylabel('Fan speed [%]')
        figure.subplots_adjust(top=1)
        canvas.set_size_request(400, 300)
        fan_scrolled_window.add_with_viewport(canvas)
        # Returns a tuple of line objects, thus the comma
        lines = axis.plot([], [], 'o-', linewidth=3.0, markersize=10, antialiased=True)
        axis.set_ybound(lower=0, upper=105)
        axis.set_xbound(20, 60)
        figure.canvas.draw()
        return lines

    def __plot_fan_chart(self, data: Dict[int, int]) -> None:
        temperature = list(data.keys())
        duty = list(data.values())
        self.__fan_line.set_xdata(temperature)
        self.__fan_line.set_ydata(duty)
        self.__fan_canvas.draw()
        self.__fan_canvas.flush_events()

    def __plot_pump_chart(self, data: Dict[int, int]) -> None:
        temperature = list(data.keys())
        duty = list(data.values())
        self.__pump_line.set_xdata(temperature)
        self.__pump_line.set_ydata(duty)
        self.__pump_canvas.draw()
        self.__pump_canvas.flush_events()

presenter.py

import logging
import multiprocessing
import re
from typing import Optional, Any, List, Tuple, Dict, Callable

from gi.repository import Gtk
from injector import inject, singleton
from rx import Observable
from rx.concurrency import GtkScheduler, ThreadPoolScheduler
from rx.concurrency.schedulerbase import SchedulerBase
from rx.disposables import CompositeDisposable

from gkraken.conf import SETTINGS_DEFAULTS
from gkraken.interactor import GetStatusInteractor, SetSpeedProfileInteractor, SettingsInteractor
from gkraken.model import Status, SpeedProfile, ChannelType, CurrentSpeedProfile

LOG = logging.getLogger(__name__)
_REFRESH_INTERVAL_IN_MS = 3000
_ADD_NEW_PROFILE_INDEX = -10


class ViewInterface:
    def toggle_window_visibility(self) -> None:
        raise NotImplementedError()

    def refresh_status(self, status: Optional[Status]) -> None:
        raise NotImplementedError()

    def refresh_profile_combobox(self, channel: ChannelType, data: List[Tuple[int, str]],
                                 active: Optional[int]) -> None:
        raise NotImplementedError()

    def refresh_chart(self, profile: SpeedProfile) -> None:
        raise NotImplementedError()

    def set_apply_button_enabled(self, channel: ChannelType, enabled: bool) -> None:
        raise NotImplementedError()

    def set_statusbar_text(self, text: str) -> None:
        raise NotImplementedError()

    def refresh_content_header_bar_title(self) -> None:
        raise NotImplementedError()

    def refresh_settings(self, settings: Dict[str, Any]) -> None:
        raise NotImplementedError()

    def show_add_speed_profile_dialog(self, channel: ChannelType) -> None:
        raise NotImplementedError()

    def show_settings_dialog(self) -> None:
        raise NotImplementedError()

    def hide_settings_dialog(self) -> None:
        raise NotImplementedError()


@singleton
class Presenter:
    @inject
    def __init__(self,
                 get_status_interactor: GetStatusInteractor,
                 set_speed_profile_interactor: SetSpeedProfileInteractor,
                 settings_interactor: SettingsInteractor,
                 composite_disposable: CompositeDisposable,
                 ) -> None:
        LOG.debug("init Presenter ")
        self.view: ViewInterface = ViewInterface()
        self.__scheduler: SchedulerBase = ThreadPoolScheduler(multiprocessing.cpu_count())
        self.__get_status_interactor: GetStatusInteractor = get_status_interactor
        self.__set_speed_profile_interactor: SetSpeedProfileInteractor = set_speed_profile_interactor
        self.__settings_interactor = settings_interactor
        self.__composite_disposable: CompositeDisposable = composite_disposable
        self.__profile_selected: Dict[str, SpeedProfile] = {}
        self.__should_update_fan_speed: bool = False
        self.application_quit: Callable = lambda *args: None  # will be set by the Application

    def on_start(self) -> None:
        self.__init_speed_profiles()
        self.__init_settings()
        self.__start_refresh()

    def __start_refresh(self) -> None:
        LOG.debug("start refresh")
        self.__composite_disposable \
            .add(Observable
                 .interval(_REFRESH_INTERVAL_IN_MS, scheduler=self.__scheduler)
                 .start_with(0)
                 .subscribe_on(self.__scheduler)
                 .flat_map(lambda _: self.__get_status())
                 .observe_on(GtkScheduler())
                 .subscribe(on_next=self.__update_status,
                            on_error=lambda e: LOG.exception("Refresh error: %s", str(e)))
                 )

    def __update_status(self, status: Optional[Status]) -> None:
        if status is not None:
            if self.__should_update_fan_speed:
                last_applied: CurrentSpeedProfile = CurrentSpeedProfile.get_or_none(channel=ChannelType.FAN.value)
                if last_applied is not None:
                    duties = [i.duty for i in last_applied.profile.steps if status.liquid_temperature >= i.temperature]
                    if duties:
                        status.fan_speed = duties[-1]

            self.view.refresh_status(status)

    # def __load_last_profile(self) -> None:
    #     for current in CurrentSpeedProfile.select():

    @staticmethod
    def __get_profile_list(channel: ChannelType) -> List[Tuple[int, str]]:
        return [(p.id, p.name) for p in SpeedProfile.select().where(SpeedProfile.channel == channel.value)]

    def __init_speed_profiles(self) -> None:
        for channel in ChannelType:
            data = self.__get_profile_list(channel)

            active = None
            if self.__settings_interactor.get_bool('settings_load_last_profile'):
                self.__should_update_fan_speed = True
                current: CurrentSpeedProfile = CurrentSpeedProfile.get_or_none(channel=channel.value)
                if current is not None:
                    active = next(i for i, item in enumerate(data) if item[0] == current.profile.id)
                    self.__set_speed_profile(current.profile)

            data.append((_ADD_NEW_PROFILE_INDEX, "<span style='italic' alpha='50%'>Add new profile...</span>"))

            self.view.refresh_profile_combobox(channel, data, active)

    def __init_settings(self) -> None:
        settings = {}
        for key, default_value in SETTINGS_DEFAULTS.items():
            if isinstance(default_value, bool):
                settings[key] = self.__settings_interactor.get_bool(key)
        self.view.refresh_settings(settings)

    def on_menu_settings_clicked(self, *_: Any) -> None:
        self.view.show_settings_dialog()

    def on_settings_dialog_closed(self, *_: Any) -> bool:
        self.view.hide_settings_dialog()
        return True

    def on_setting_changed(self, widget: Any, *args: Any) -> None:
        if isinstance(widget, Gtk.Switch):
            state = args[0]
            key = re.sub('_switch$', '', widget.get_name())
            self.__settings_interactor.set_bool(key, state)

    def on_stack_visible_child_changed(self, *_: Any) -> None:
        self.view.refresh_content_header_bar_title()

    def on_fan_profile_selected(self, widget: Any, *_: Any) -> None:
        profile_id = widget.get_model()[widget.get_active()][0]
        self.__select_speed_profile(profile_id, ChannelType.FAN)

    def on_pump_profile_selected(self, widget: Any, *_: Any) -> None:
        profile_id = widget.get_model()[widget.get_active()][0]
        self.__select_speed_profile(profile_id, ChannelType.PUMP)

    def on_quit_clicked(self, *_: Any) -> None:
        self.application_quit()

    def on_toggle_app_window_clicked(self, *_: Any) -> None:
        self.view.toggle_window_visibility()

    def __select_speed_profile(self, profile_id: int, channel: ChannelType) -> None:
        if profile_id == _ADD_NEW_PROFILE_INDEX:
            self.view.set_apply_button_enabled(channel, False)
            self.view.show_add_speed_profile_dialog(channel)
        else:
            profile: SpeedProfile = SpeedProfile.get(id=profile_id)
            self.__profile_selected[profile.channel] = profile
            self.view.set_apply_button_enabled(channel, True)
            self.view.refresh_chart(profile)

    @staticmethod
    def __get_profile_data(profile: SpeedProfile) -> List[Tuple[int, int]]:
        return [(p.temperature, p.duty) for p in profile.steps]

    def on_fab_apply_button_clicked(self, *_: Any) -> None:
        self.__set_speed_profile(self.__profile_selected[ChannelType.FAN.value])
        self.__should_update_fan_speed = True

    def on_pump_apply_button_clicked(self, *_: Any) -> None:
        self.__set_speed_profile(self.__profile_selected[ChannelType.PUMP.value])

    def __set_speed_profile(self, profile: SpeedProfile) -> None:
        observable = self.__set_speed_profile_interactor \
            .execute(profile.channel, self.__get_profile_data(profile))
        self.__composite_disposable \
            .add(observable
                 .subscribe_on(self.__scheduler)
                 .observe_on(GtkScheduler())
                 .subscribe(on_next=lambda _: self.__update_current_speed_profile(profile),
                            on_error=lambda e: (LOG.exception("Set cooling error: %s", str(e)),
                                                self.view.set_statusbar_text('Error applying %s speed profile!'
                                                                             % profile.channel))))

    def __update_current_speed_profile(self, profile: SpeedProfile) -> None:
        current: CurrentSpeedProfile = CurrentSpeedProfile.get_or_none(channel=profile.channel)
        if current is None:
            CurrentSpeedProfile.create(channel=profile.channel, profile=profile)
        else:
            current.profile = profile
            current.save()
        self.view.set_statusbar_text('%s cooling profile applied' % profile.channel.capitalize())

    def __log_exception_return_empty_observable(self, ex: Exception) -> Observable:
        LOG.exception("Err = %s", ex)
        self.view.set_statusbar_text(str(ex))
        return Observable.just(None)

    def __get_status(self) -> Observable:
        return self.__get_status_interactor.execute() \
            .catch_exception(lambda ex: self.__log_exception_return_empty_observable(ex))

interactor.py

LOG = logging.getLogger(__name__)


@singleton
class GetStatusInteractor:
    @inject
    def __init__(self,
                 kraken_repository: KrakenRepository,
                 ) -> None:
        self.__kraken_repository = kraken_repository

    def execute(self) -> Observable:
        LOG.debug("GetStatusInteractor.execute()")
        return Observable.defer(lambda: Observable.just(self.__kraken_repository.get_status()))


@singleton
class SetSpeedProfileInteractor:
    @inject
    def __init__(self,
                 kraken_repository: KrakenRepository,
                 ) -> None:
        self.__kraken_repository = kraken_repository

    def execute(self, channel_value: str, profile_data: List[Tuple[int, int]]) -> Observable:
        LOG.debug("SetSpeedProfileInteractor.execute()")
        return Observable.defer(lambda: Observable.just(
            self.__kraken_repository.set_speed_profile(channel_value, profile_data)))


@singleton
class SettingsInteractor:
    @inject
    def __init__(self) -> None:
        pass

    @staticmethod
    def get_bool(key: str, default: Optional[bool] = None) -> bool:
        if default is None:
            default = SETTINGS_DEFAULTS[key]
        setting: Setting = Setting.get_or_none(key=key)
        if setting is not None:
            return bool(setting.value)
        return bool(default)

    @staticmethod
    def set_bool(key: str, value: bool) -> None:
        setting: Setting = Setting.get_or_none(key=key)
        if setting is not None:
            setting.value = value
            setting.save()
        else:
            Setting.create(key=key, value=value)

    @staticmethod
    def get_str(key: str, default: Optional[str] = None) -> str:
        if default is None:
            default = SETTINGS_DEFAULTS[key]
        setting: Setting = Setting.get_or_none(key=key)
        if setting is not None:
            return str(setting.value.decode("utf-8"))
        return str(default)

    @staticmethod
    def set_str(key: str, value: str) -> None:
        setting: Setting = Setting.get_or_none(key=key)
        if setting is not None:
            setting.value = value.encode("utf-8")
            setting.save()
        else:
            Setting.create(key=key, value=value.encode("utf-8"))

repository.py

import logging
import threading
from enum import Enum
from typing import Optional, List, Tuple

import liquidctl
from injector import singleton, inject
from liquidctl.driver.kraken_two import KrakenTwoDriver

from gkraken.model import Status
from gkraken.di import INJECTOR
from gkraken.util import synchronized_with_attr

LOG = logging.getLogger(__name__)


@singleton
class KrakenRepository:
    @inject
    def __init__(self) -> None:
        self.lock = threading.RLock()
        self.__driver: Optional[KrakenTwoDriver] = None

    def cleanup(self) -> None:
        LOG.debug("KrakenRepository cleanup")
        if self.__driver:
            self.__driver.finalize()
            self.__driver = None

    @synchronized_with_attr("lock")
    def get_status(self) -> Optional[Status]:
        self.__load_driver()
        if self.__driver:
            try:
                status = [v for k, v, u in self.__driver.get_status()]
                return Status(
                    status[_StatusType.LIQUID_TEMPERATURE.value],
                    status[_StatusType.FAN_RPM.value],
                    status[_StatusType.PUMP_RPM.value],
                    status[_StatusType.FIRMWARE_VERSION.value]
                )
            # pylint: disable=bare-except
            except:
                LOG.exception("Error getting the status")
                self.cleanup()
        return None

    @synchronized_with_attr("lock")
    def set_speed_profile(self, channel_value: str, profile_data: List[Tuple[int, int]]) -> None:
        self.__load_driver()
        if self.__driver and profile_data:
            try:
                if len(profile_data) == 1:
                    self.__driver.set_fixed_speed(channel_value, profile_data[0][1])
                else:
                    self.__driver.set_speed_profile(channel_value, profile_data)
            # pylint: disable=bare-except
            except:
                LOG.exception("Error getting the status")
                self.cleanup()

    def __load_driver(self) -> None:
        if not self.__driver:
            self.__driver = INJECTOR.get(Optional[KrakenTwoDriver])

            if self.__driver:
                liquidctl.util.verbose = True
                self.__driver.initialize()
            else:
                raise ValueError("Kraken USB interface error (check USB cable connection)")


class _StatusType(Enum):
    LIQUID_TEMPERATURE = 0
    FAN_RPM = 1
    PUMP_RPM = 2
    FIRMWARE_VERSION = 3

model.py

# pylint: disable=too-many-locals,too-many-instance-attributes
from enum import Enum
from typing import Optional

from peewee import Model, CharField, DateTimeField, SqliteDatabase, SQL, IntegerField, Check, \
    ForeignKeyField, BooleanField, BlobField
from playhouse.sqlite_ext import AutoIncrementField

from gkraken.di import INJECTOR


class Status:
    def __init__(self,
                 liquid_temperature: float,
                 fan_rpm: int,
                 pump_rpm: int,
                 firmware_version: str
                 ) -> None:
        self.liquid_temperature: float = liquid_temperature
        self.fan_rpm: int = fan_rpm
        self.fan_speed: Optional[int] = None
        self.pump_rpm: int = pump_rpm
        self.firmware_version: str = firmware_version


class ChannelType(Enum):
    FAN = 'fan'
    PUMP = 'pump'


class SpeedProfile(Model):
    id = AutoIncrementField()
    channel = CharField(constraints=[Check("channel='%s' OR channel='%s'"
                                           % (ChannelType.FAN.value, ChannelType.PUMP.value))])
    name = CharField()
    read_only = BooleanField(default=False)
    single_step = BooleanField(default=False)
    timestamp = DateTimeField(constraints=[SQL('DEFAULT CURRENT_TIMESTAMP')])

    class Meta:
        legacy_table_names = False
        database = INJECTOR.get(SqliteDatabase)


class SpeedStep(Model):
    profile = ForeignKeyField(SpeedProfile, backref='steps')
    temperature = IntegerField()
    duty = IntegerField()
    timestamp = DateTimeField(constraints=[SQL('DEFAULT CURRENT_TIMESTAMP')])

    class Meta:
        legacy_table_names = False
        database = INJECTOR.get(SqliteDatabase)


class CurrentSpeedProfile(Model):
    channel = CharField(primary_key=True, constraints=[Check("channel='%s' OR channel='%s'"
                                                             % (ChannelType.FAN.value, ChannelType.PUMP.value))])
    profile = ForeignKeyField(SpeedProfile, unique=True)
    timestamp = DateTimeField(constraints=[SQL('DEFAULT CURRENT_TIMESTAMP')])

    class Meta:
        legacy_table_names = False
        database = INJECTOR.get(SqliteDatabase)


class Setting(Model):
    key = CharField(primary_key=True)
    value = BlobField()

    class Meta:
        legacy_table_names = False
        database = INJECTOR.get(SqliteDatabase)

di.py

import logging

from typing import Optional
from gi.repository import Gtk
from injector import Module, provider, singleton, Injector
from liquidctl.cli import find_all_supported_devices
from liquidctl.driver.kraken_two import KrakenTwoDriver
from peewee import SqliteDatabase
from rx.disposables import CompositeDisposable

from gkraken.conf import APP_PACKAGE_NAME, get_data_path, APP_UI_NAME, get_config_path, APP_DB_NAME

LOG = logging.getLogger(__name__)


# pylint: disable=no-self-use
class ProviderModule(Module):
    @singleton
    @provider
    def provide_builder(self) -> Gtk.Builder:
        LOG.debug("provide Gtk.Builder")
        builder = Gtk.Builder()
        builder.set_translation_domain(APP_PACKAGE_NAME)
        builder.add_from_file(get_data_path(APP_UI_NAME))
        return builder

    @singleton
    @provider
    def provide_thread_pool_scheduler(self) -> CompositeDisposable:
        LOG.debug("provide CompositeDisposable")
        return CompositeDisposable()

    @singleton
    @provider
    def provide_database(self) -> SqliteDatabase:
        LOG.debug("provide CompositeDisposable")
        return SqliteDatabase(get_config_path(APP_DB_NAME))

    @provider
    def provide_kraken_two_driver(self) -> Optional[KrakenTwoDriver]:
        LOG.debug("provide KrakenTwoDriver")
        return next((dev for dev in find_all_supported_devices() if isinstance(dev, KrakenTwoDriver)), None)


INJECTOR = Injector(ProviderModule)
\$\endgroup\$
  • \$\begingroup\$ "to use the app you need to have ... a NZXT Kraken" Here's an argument for a separable design, with a clean interface so you can provide some sort of simulation in place of the real hardware. Then only a small part of the testing needs access to the device. If that idea is new to you, say so, and I'll consider writing it up as an answer (time permitting, of course!). \$\endgroup\$ – Toby Speight Oct 10 '18 at 17:31
  • \$\begingroup\$ Welcome to Code Review! I changed the title so that it describes what the code does per site goals: "State what your code does in your title, not your main concerns about it.". Please check that I haven't misrepresented your code, and correct it if I have. \$\endgroup\$ – Toby Speight Oct 10 '18 at 17:32
  • \$\begingroup\$ Hi Toby, thank you for your comment, I think I have already something that can be considered an interface, since I don't access directly the USB interface of the Kraken but I access it via the liquidctl library (self.__driver: Optional[KrakenTwoDriver]). All my interactions with the Kraken go from it and it should be possible to simulate it providing a mock for the set_speed_profile and get_status methods of the instance returned by the def provide_kraken_two_driver() method in di.py. \$\endgroup\$ – Roberto Leinardi Oct 10 '18 at 17:41

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.