');
+ }
+ }
+ // Currently, this event is equivalent to the complete initialization of the backend side of the project
+ onNameChanged: {
+ for (let i = 0; i < buttonsModel.count; ++i) {
+ projActionsRow.children[i].enabled = true;
+ }
+
+ const state = project.state;
+ const completedStages = Object.keys(state).filter(stateName => state[stateName]);
+ if (completedStages.length === 1 && completedStages[0] === 'EMPTY') {
+ initScreenLoader.active = true;
+ mainOrInitScreen.currentIndex = 0; // show init dialog
+ } else {
+ mainOrInitScreen.currentIndex = 1; // show main view
+ }
+ }
+ }
+
+ /*
+ Prompt a user to perform initial setup
+ */
+ Loader {
+ id: initScreenLoader
+ active: false
+ sourceComponent: Column {
+ Text {
+ text: "To complete initialization you can provide PlatformIO name of the board"
+ padding: 10
+ }
+ Row {
+ padding: 10
+ spacing: 10
+ ComboBox {
+ id: board
+ editable: true
+ model: boardsModel // backend-side (simple string model)
+ textRole: 'display'
+ onAccepted: {
+ focus = false;
+ }
+ onActivated: {
+ focus = false;
+ }
+ onFocusChanged: {
+ if (!focus) {
+ if (find(editText) === -1) {
+ editText = textAt(0); // should be 'None' at index 0
+ }
+ } else {
+ selectAll();
+ }
+ }
+ }
+ /*
+ Trigger full run
+ */
+ CheckBox {
+ id: runCheckBox
+ text: 'Run'
+ enabled: false
+ ToolTip {
+ visible: runCheckBox.hovered
+ Component.onCompleted: {
+ const actions = [];
+ for (let i = 3; i < buttonsModel.count; ++i) {
+ actions.push(`${buttonsModel.get(i).name}`);
+ }
+ text = `Do: ${actions.join(' → ')}`;
+ }
+ }
+ Connections {
+ target: board
+ onFocusChanged: {
+ if (!board.focus) {
+ if (board.editText === board.textAt(0)) { // should be 'None' at index 0
+ runCheckBox.checked = false;
+ runCheckBox.enabled = false;
+ } else {
+ runCheckBox.enabled = true;
+ }
+ }
+ }
+ }
+ }
+ CheckBox {
+ id: openEditor
+ text: 'Open editor'
+ ToolTip {
+ text: 'Start the editor specified in the Settings after the completion'
+ visible: openEditor.hovered
+ }
+ }
+ }
+ Button {
+ text: 'OK'
+ topInset: 15
+ leftInset: 10
+ topPadding: 20
+ leftPadding: 18
+ onClicked: {
+ // All operations will be queued
+ projectsListView.currentItem.item.actionRunning = true;
+
+ project.run('save_config', [{
+ 'project': {
+ 'board': board.editText === board.textAt(0) ? '' : board.editText
+ }
+ }]);
+
+ if (runCheckBox.checked) {
+ for (let i = 3; i < buttonsModel.count - 1; ++i) {
+ buttonsModel.setProperty(i, 'shouldRunNext', true);
+ }
+ projActionsRow.children[3].clicked();
+ }
+
+ if (openEditor.checked) {
+ if (runCheckBox.checked) {
+ buttonsModel.setProperty(buttonsModel.count - 1, 'shouldStartEditor', true);
+ } else {
+ projActionsRow.children[1].clicked();
+ }
+ }
+
+ mainOrInitScreen.currentIndex = 1; // go to main screen
+ initScreenLoader.sourceComponent = undefined; // destroy init screen
+ }
+ }
+ }
+ }
+
+ ColumnLayout {
+ Layout.fillWidth: true
+ Layout.fillHeight: true
+
+ /*
+ Detect and reflect changes of a project outside of the app
+ */
+ QtDialogs.MessageDialog {
+ // TODO: case: .ioc file can be removed on init stage too (i.e. when initDialog is active)
+ id: projectIncorrectDialog
+ text: `The project was modified outside of the stm32pio and .ioc file is no longer present.
+ The project will be removed from the app. It will not affect any real content`
+ icon: QtDialogs.StandardIcon.Critical
+ onAccepted: {
+ moveToNextAndRemove();
+ projActionsButtonGroup.lock = false;
+ }
+ }
+
+ /*
+ Show this or action buttons
+ */
+ Text {
+ id: initErrorMessage
+ visible: false
+ padding: 10
+ text: "The project cannot be initialized"
+ color: 'red'
+ }
+
+ /*
+ The core widget - a group of buttons mapping all main actions that can be performed on the given project.
+ They also serve the project state displaying - each button indicates a stage associated with it:
+ - green: done
+ - yellow: in progress right now
+ - red: an error has occured during the last execution
+ */
+ ButtonGroup {
+ id: projActionsButtonGroup
+ buttons: projActionsRow.children
+ signal stateReceived()
+ signal actionDone(string actionDone, bool success)
+ property bool lock: false // TODO: is it necessary? mb make a dialog modal or smth.
+ onStateReceived: { // TODO: cache state!
+ if (mainWindow.active && (index === projectsWorkspaceView.currentIndex) && !lock) {
+ const state = project.state;
+ project.stageChanged();
+
+ if (state['LOADING']) {
+ //
+ } else if (state['INIT_ERROR']) {
+ projActionsRow.visible = false;
+ initErrorMessage.visible = true;
+ } else if (!state['EMPTY']) {
+ lock = true; // projectIncorrectDialog.visible is not working correctly (seems like delay or smth.)
+ projectIncorrectDialog.open();
+ } else if (state['EMPTY']) {
+ for (let i = 0; i < buttonsModel.count; ++i) {
+ projActionsRow.children[i].palette.button = 'lightgray';
+ if (state[buttonsModel.get(i).state]) {
+ projActionsRow.children[i].palette.button = 'lightgreen';
+ }
+ }
+ }
+ }
+ }
+ onActionDone: {
+ for (let i = 0; i < buttonsModel.count; ++i) {
+ projActionsRow.children[i].enabled = true;
+ }
+ }
+ onClicked: {
+ for (let i = 0; i < buttonsModel.count; ++i) {
+ projActionsRow.children[i].enabled = false;
+ projActionsRow.children[i].glowVisible = false;
+ }
+ }
+ Component.onCompleted: {
+ // Several events lead to a single handler:
+ // - the state has changed and explicitly informs about it
+ // - the project was selected in the list
+ // - the app window has got the focus
+ project.stateChanged.connect(stateReceived);
+ projectsWorkspaceView.currentIndexChanged.connect(stateReceived);
+ mainWindow.activeChanged.connect(stateReceived);
+
+ project.actionDone.connect(actionDone);
+ }
+ }
+ RowLayout {
+ id: projActionsRow
+ Layout.fillWidth: true
+ Layout.bottomMargin: 7
+ z: 1 // for the glowing animation
+ Repeater {
+ model: ListModel {
+ id: buttonsModel
+ ListElement {
+ name: 'Clean'
+ action: 'clean'
+ shouldStartEditor: false
+ }
+ ListElement {
+ name: 'Open editor'
+ action: 'start_editor'
+ margin: 15 // margin to visually separate first 2 actions as they doesn't represent any state
+ }
+ ListElement {
+ name: 'Initialize'
+ state: 'INITIALIZED'
+ action: 'save_config'
+ shouldRunNext: false
+ shouldStartEditor: false
+ }
+ ListElement {
+ name: 'Generate'
+ state: 'GENERATED'
+ action: 'generate_code'
+ shouldRunNext: false
+ shouldStartEditor: false
+ }
+ ListElement {
+ name: 'Init PlatformIO'
+ state: 'PIO_INITIALIZED'
+ action: 'pio_init'
+ shouldRunNext: false
+ shouldStartEditor: false
+ }
+ ListElement {
+ name: 'Patch'
+ state: 'PATCHED'
+ action: 'patch'
+ shouldRunNext: false
+ shouldStartEditor: false
+ }
+ ListElement {
+ name: 'Build'
+ state: 'BUILT'
+ action: 'build'
+ shouldRunNext: false
+ shouldStartEditor: false
+ }
+ }
+ delegate: Button {
+ text: name
+ Layout.rightMargin: model.margin
+ enabled: false // turn on after project initialization
+ property alias glowVisible: glow.visible
+ function runOwnAction() {
+ projectsListView.currentItem.item.actionRunning = true;
+ palette.button = 'gold';
+ let args = []; // JS array cannot be attached to a ListElement (at least in a non-hacky manner)
+ if (model.action === 'start_editor') {
+ args.push(settings.get('editor'));
+ }
+ project.run(model.action, args);
+ }
+ onClicked: {
+ runOwnAction();
+ }
+ /*
+ Detect modifier keys:
+ - Ctrl: start the editor after an operation(s)
+ - Shift: continuous actions run
+ */
+ MouseArea {
+ anchors.fill: parent
+ hoverEnabled: true
+ property bool ctrlPressed: false
+ property bool ctrlPressedLastState: false
+ property bool shiftPressed: false
+ property bool shiftPressedLastState: false
+ function shiftHandler() {
+ for (let i = 2; i <= index; ++i) { // TODO: magic number, actually...
+ if (shiftPressed) {
+ projActionsRow.children[i].palette.button = Qt.lighter('lightgreen', 1.2);
+ } else {
+ projActionsButtonGroup.stateReceived();
+ }
+ }
+ }
+ onClicked: {
+ if (ctrlPressed && model.action !== 'start_editor') {
+ model.shouldStartEditor = true;
+ }
+ if (shiftPressed && index >= 2) {
+ // run all actions in series
+ for (let i = 2; i < index; ++i) {
+ buttonsModel.setProperty(i, 'shouldRunNext', true);
+ }
+ projActionsRow.children[2].clicked();
+ return;
+ }
+ parent.clicked(); // propagateComposedEvents doesn't work...
+ }
+ onPositionChanged: {
+ ctrlPressed = mouse.modifiers & Qt.ControlModifier; // bitwise AND
+ if (ctrlPressedLastState !== ctrlPressed) {
+ ctrlPressedLastState = ctrlPressed;
+ }
+
+ shiftPressed = mouse.modifiers & Qt.ShiftModifier; // bitwise AND
+ if (shiftPressedLastState !== shiftPressed) {
+ shiftPressedLastState = shiftPressed;
+ shiftHandler();
+ }
+ }
+ onEntered: {
+ statusBar.text =
+ `Ctrl-click to open the editor specified in the Settings after the operation,
+ Shift-click to perform all actions prior this one (including).
+ Ctrl-Shift-click for both`;
+ }
+ onExited: {
+ statusBar.text = '';
+
+ ctrlPressed = false;
+ ctrlPressedLastState = false;
+
+ if (shiftPressed || shiftPressedLastState) {
+ shiftPressed = false;
+ shiftPressedLastState = false;
+ shiftHandler();
+ }
+ }
+ }
+ Connections {
+ target: projActionsButtonGroup
+ onActionDone: {
+ if (actionDone === model.action) {
+ if (success) {
+ glow.color = 'lightgreen';
+ } else {
+ palette.button = 'lightcoral';
+ glow.color = 'lightcoral';
+ }
+ glow.visible = true;
+
+ if (model.shouldRunNext) {
+ model.shouldRunNext = false;
+ projActionsRow.children[index + 1].clicked();
+ }
+
+ if (model.shouldStartEditor) {
+ model.shouldStartEditor = false;
+ for (let i = 0; i < buttonsModel.count; ++i) {
+ if (buttonsModel.get(i).action === 'start_editor') {
+ // Use runOwnAction for no additional actions in parent handlers
+ projActionsRow.children[i].runOwnAction();
+ break;
+ }
+ }
+ }
+ }
+ }
+ }
+ /*
+ Blinky glowing
+ */
+ RectangularGlow {
+ id: glow
+ visible: false
+ anchors.fill: parent
+ cornerRadius: 25
+ glowRadius: 20
+ spread: 0.25
+ onVisibleChanged: {
+ visible ? glowAnimation.start() : glowAnimation.complete();
+ }
+ SequentialAnimation {
+ id: glowAnimation
+ loops: 3
+ OpacityAnimator {
+ target: glow
+ from: 0
+ to: 1
+ duration: 1000
+ }
+ OpacityAnimator {
+ target: glow
+ from: 1
+ to: 0
+ duration: 1000
+ }
+ }
+ }
+ }
+ }
+ }
+
+ Rectangle {
+ Layout.fillWidth: true
+ Layout.fillHeight: true
+
+ ScrollView {
+ anchors.fill: parent
+ TextArea {
+ id: log
+ readOnly: true
+ selectByMouse: true
+ wrapMode: Text.WordWrap
+ font.family: 'Courier'
+ font.pointSize: 10 // different on different platforms, Qt's bug
+ textFormat: TextEdit.RichText
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ /*
+ Simple text line. Currently, doesn't support smart intrinsic properties as a fully-fledged status bar,
+ but is used only for a single feature so not a big deal
+ */
+ footer: Text {
+ id: statusBar
+ padding: 10
+ }
+}
diff --git a/stm32pio-test-project/stm32pio-test-project.ioc b/stm32pio-test-project/stm32pio-test-project.ioc
index 834d9de..90016d5 100644
--- a/stm32pio-test-project/stm32pio-test-project.ioc
+++ b/stm32pio-test-project/stm32pio-test-project.ioc
@@ -1,6 +1,6 @@
#MicroXplorer Configuration settings - do not modify
File.Version=6
-KeepUserPlacement=true
+KeepUserPlacement=false
Mcu.Family=STM32F0
Mcu.IP0=NVIC
Mcu.IP1=RCC
@@ -19,8 +19,8 @@ Mcu.PinsNb=6
Mcu.ThirdPartyNb=0
Mcu.UserConstants=
Mcu.UserName=STM32F031K6Tx
-MxCube.Version=5.4.0
-MxDb.Version=DB.5.0.40
+MxCube.Version=5.6.0
+MxDb.Version=DB.5.0.60
NVIC.ForceEnableDMAVector=true
NVIC.HardFault_IRQn=true\:0\:0\:false\:false\:true\:false\:false
NVIC.NonMaskableInt_IRQn=true\:0\:0\:false\:false\:true\:false\:false
@@ -47,14 +47,6 @@ PA2.GPIO_Label=VCP_TX
PA2.Locked=true
PA2.Mode=Asynchronous
PA2.Signal=USART1_TX
-PCC.Checker=false
-PCC.Line=STM32F0x1
-PCC.MCU=STM32F031K6Tx
-PCC.PartNumber=STM32F031K6Tx
-PCC.Seq0=0
-PCC.Series=STM32F0
-PCC.Temperature=25
-PCC.Vdd=3.6
PF0-OSC_IN.Locked=true
PF0-OSC_IN.Mode=HSE-External-Clock-Source
PF0-OSC_IN.Signal=RCC_OSC_IN
diff --git a/stm32pio/app.py b/stm32pio/app.py
index bff187f..d61b0c9 100755
--- a/stm32pio/app.py
+++ b/stm32pio/app.py
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
-__version__ = '0.96'
+__version__ = '1.0'
import argparse
import logging
@@ -36,11 +36,12 @@ def parse_args(args: list) -> Optional[argparse.Namespace]:
"proceeding")
parser_new = subparsers.add_parser('new', help="generate CubeMX code, create PlatformIO project")
parser_generate = subparsers.add_parser('generate', help="generate CubeMX code only")
+ parser_status = subparsers.add_parser('status', help="get the description of the current project state")
parser_clean = subparsers.add_parser('clean', help="clean-up the project (WARNING: it deletes ALL content of "
"'path' except the .ioc file)")
# Common subparsers options
- for p in [parser_init, parser_new, parser_generate, parser_clean]:
+ for p in [parser_init, parser_new, parser_generate, parser_status, parser_clean]:
p.add_argument('-d', '--directory', dest='project_path', default=pathlib.Path.cwd(), required=False,
help="path to the project (current directory, if not given)")
for p in [parser_init, parser_new]:
@@ -75,7 +76,10 @@ def main(sys_argv=None) -> int:
if sys_argv is None:
sys_argv = sys.argv[1:]
+ # Import modules after sys.path modification
import stm32pio.settings
+ import stm32pio.lib
+ import stm32pio.util
args = parse_args(sys_argv)
@@ -85,21 +89,20 @@ def main(sys_argv=None) -> int:
# Currently only 2 levels of verbosity through the '-v' option are counted (INFO (default) and DEBUG (-v))
if args is not None and args.subcommand is not None and args.verbose:
logger.setLevel(logging.DEBUG)
- handler.setFormatter(logging.Formatter("%(levelname)-8s "
- f"%(funcName)-{stm32pio.settings.log_function_fieldwidth}s "
- "%(message)s"))
+ handler.setFormatter(stm32pio.util.DispatchingFormatter(
+ f"%(levelname)-8s %(funcName)-{stm32pio.settings.log_fieldwidth_function}s %(message)s",
+ special=stm32pio.util.special_formatters))
logger.debug("debug logging enabled")
elif args is not None and args.subcommand is not None:
logger.setLevel(logging.INFO)
- handler.setFormatter(logging.Formatter("%(levelname)-8s %(message)s"))
+ handler.setFormatter(stm32pio.util.DispatchingFormatter("%(levelname)-8s %(message)s",
+ special=stm32pio.util.special_formatters))
else:
logger.setLevel(logging.INFO)
handler.setFormatter(logging.Formatter("%(message)s"))
logger.info("\nNo arguments were given, exiting...")
return 0
- import stm32pio.lib # import the module after sys.path modification and logger configuration
-
# Main routine
try:
if args.subcommand == 'init':
@@ -107,7 +110,7 @@ def main(sys_argv=None) -> int:
if not args.board:
logger.warning("STM32 PlatformIO board is not specified, it will be needed on PlatformIO project "
"creation")
- logger.info('project has been initialized. You can now edit stm32pio.ini config file')
+ logger.info("project has been initialized. You can now edit stm32pio.ini config file")
if args.editor:
project.start_editor(args.editor)
@@ -131,16 +134,19 @@ def main(sys_argv=None) -> int:
if args.editor:
project.start_editor(args.editor)
+ elif args.subcommand == 'status':
+ project = stm32pio.lib.Stm32pio(args.project_path, save_on_destruction=False)
+ print(project.state)
+
elif args.subcommand == 'clean':
project = stm32pio.lib.Stm32pio(args.project_path, save_on_destruction=False)
project.clean()
- # library is designed to throw the exception in bad cases so we catch here globally
+ # Library is designed to throw the exception in bad cases so we catch here globally
except Exception as e:
- logger.exception(e, exc_info=logger.getEffectiveLevel() <= logging.DEBUG)
+ logger.exception(e, exc_info=logger.isEnabledFor(logging.DEBUG))
return -1
- logger.info("exiting...")
return 0
diff --git a/stm32pio/lib.py b/stm32pio/lib.py
index 35c81a5..417430e 100644
--- a/stm32pio/lib.py
+++ b/stm32pio/lib.py
@@ -1,7 +1,9 @@
"""
-Main library
+Core library
"""
+from __future__ import annotations
+
import collections
import configparser
import enum
@@ -14,62 +16,107 @@
import weakref
import stm32pio.settings
-
-# Child logger, inherits parameters of the parent that has been set in more high-level code
-logger = logging.getLogger('stm32pio.util')
+import stm32pio.util
@enum.unique
-class ProjectState(enum.IntEnum):
+class ProjectStage(enum.IntEnum):
"""
Codes indicating a project state at the moment. Should be the sequence of incrementing integers to be suited for
- state determining algorithm. Starting from 1
+ state determining algorithm. Starts from 1
Hint: Files/folders to be present on every project state:
UNDEFINED: use this state to indicate none of the states below. Also, when we do not have any .ioc file the
- Stm32pio class cannot be instantiated (constructor raises an exception)
+ Stm32pio class instance cannot be created (constructor raises an exception)
+ EMPTY: ['project.ioc']
INITIALIZED: ['project.ioc', 'stm32pio.ini']
GENERATED: ['Inc', 'Src', 'project.ioc', 'stm32pio.ini']
PIO_INITIALIZED (on case-sensitive FS): ['test', 'include', 'Inc', 'platformio.ini', '.gitignore', 'Src', 'lib',
'project.ioc', '.travis.yml', 'src']
- PIO_INI_PATCHED: ['test', 'Inc', 'platformio.ini', '.gitignore', 'Src', 'lib', 'project.ioc', '.travis.yml']
+ PATCHED: ['test', 'Inc', 'platformio.ini', '.gitignore', 'Src', 'lib', 'project.ioc', '.travis.yml']
BUILT: same as above + '.pio' folder with build artifacts (such as .pio/build/nucleo_f031k6/firmware.bin,
.pio/build/nucleo_f031k6/firmware.elf)
"""
UNDEFINED = enum.auto() # note: starts from 1
+ EMPTY = enum.auto()
INITIALIZED = enum.auto()
GENERATED = enum.auto()
PIO_INITIALIZED = enum.auto()
PATCHED = enum.auto()
BUILT = enum.auto()
+ def __str__(self):
+ string_representations = {
+ 'UNDEFINED': 'The project is messed up',
+ 'EMPTY': '.ioc file is present',
+ 'INITIALIZED': 'stm32pio initialized',
+ 'GENERATED': 'CubeMX code generated',
+ 'PIO_INITIALIZED': 'PlatformIO project initialized',
+ 'PATCHED': 'PlatformIO project patched',
+ 'BUILT': 'PlatformIO project built'
+ }
+ return string_representations[self.name]
+
-class Config(configparser.ConfigParser):
+class ProjectState(collections.OrderedDict):
"""
- A simple subclass that has additional save() method for the better logic encapsulation
+ The ordered dictionary subclass suitable for storing the Stm32pio instances state. For example:
+ {
+ ProjectStage.UNDEFINED: True, # doesn't necessarily means that the project is messed up, see below
+ ProjectStage.EMPTY: True,
+ ProjectStage.INITIALIZED: True,
+ ProjectStage.GENERATED: False,
+ ProjectStage.PIO_INITIALIZED: False,
+ ProjectStage.PATCHED: False,
+ ProjectStage.BUILT: False
+ }
+ It is also extended with additional properties providing useful information such as obtaining the project current
+ stage.
+
+ The class has no special constructor so its filling - both stages and their order - is a responsibility of the
+ external code. It also has no protection nor checks for its internal correctness. Anyway, it is intended to be used
+ (i.e. creating) only by the internal code of this library so there should not be any worries.
"""
- def __init__(self, location: pathlib.Path, *args, **kwargs):
+ def __str__(self):
"""
- Args:
- location: project path (where to store the config file)
- *args, **kwargs: passes to the parent's constructor
+ Pretty human-readable complete representation of the project state (not including the service one UNDEFINED to
+ not confuse the end-user)
"""
- super().__init__(*args, **kwargs)
- self._location = location
+ # Need 2 spaces between the icon and the text to look fine
+ return '\n'.join(f"{'[*]' if stage_value else '[ ]'} {str(stage_name)}"
+ for stage_name, stage_value in self.items() if stage_name != ProjectStage.UNDEFINED)
+
+ @property
+ def current_stage(self) -> ProjectStage:
+ last_consistent_stage = ProjectStage.UNDEFINED
+ zero_found = False
+
+ # Search for a consecutive sequence of True's and find the last of them. For example, if the array is
+ # [1,1,1,0,0,0,0]
+ # ^
+ # we should consider 2 as the last index
+ for name, value in self.items():
+ if value:
+ if zero_found:
+ # Fall back to the UNDEFINED stage if we have breaks in conditions results array. E.g., for
+ # [1,1,1,0,1,0,0]
+ # we should return UNDEFINED as it doesn't look like a correct set of files actually
+ last_consistent_stage = ProjectStage.UNDEFINED
+ break
+ else:
+ last_consistent_stage = name
+ else:
+ zero_found = True
+
+ return last_consistent_stage
- def save(self) -> int:
+ @property
+ def is_consistent(self) -> bool:
"""
- Tries to save the config to the file and gently log if any error occurs
+ Whether the state has been went through the stages consequentially or not (the method is currently unused)
"""
- try:
- with self._location.joinpath(stm32pio.settings.config_file_name).open(mode='w') as config_file:
- self.write(config_file)
- logger.debug("stm32pio.ini config file has been saved")
- return 0
- except Exception as e:
- logger.warning(f"cannot save the config: {e}", exc_info=logger.getEffectiveLevel() <= logging.DEBUG)
- return -1
+ return self.current_stage != ProjectStage.UNDEFINED
class Stm32pio:
@@ -93,174 +140,210 @@ class Stm32pio:
dirty_path (str): path to the project
parameters (dict): additional parameters to set on initialization stage
save_on_destruction (bool): register or not the finalizer that saves the config to file
+ logger (logging.Logger): if an external logger is given, it will be used, otherwise the new one will be created
+ (unique for every instance)
"""
- def __init__(self, dirty_path: str, parameters: dict = None, save_on_destruction: bool = True):
+ def __init__(self, dirty_path: str, parameters: dict = None, save_on_destruction: bool = True,
+ logger: logging.Logger = None):
+
if parameters is None:
parameters = {}
+ # The individual loggers for every single project allow to fine-tune the output when multiple projects are
+ # created by the third-party code.
+ if logger is not None:
+ self.logger = logger
+ else:
+ self.logger = logging.getLogger(f"{__name__}.{id(self)}") # use id() as uniqueness guarantee
+
# The path is a unique identifier of the project so it would be great to remake Stm32pio class as a subclass of
- # pathlib.Path and then reference it like self and not self.project_path. It is more consistent also, as now
- # project_path is perceived like any other config parameter that somehow is appeared to exist outside of a
- # config instance but then it will be a core identifier, a truly 'self' value. But currently pathlib.Path is not
- # intended to be subclassable by-design, unfortunately. See https://bugs.python.org/issue24132
- self.project_path = self._resolve_project_path(dirty_path)
+ # pathlib.Path and then reference it like self and not self.path. It is more consistent also, as now path is
+ # perceived like any other config parameter that somehow is appeared to exist outside of a config instance but
+ # then it will be a core identifier, a truly 'self' value. But currently pathlib.Path is not intended to be
+ # subclassable by-design, unfortunately. See https://bugs.python.org/issue24132
+ self.path = self._resolve_project_path(dirty_path)
- self.config = self._load_config_file()
+ self.config = self._load_config()
- ioc_file = self._find_ioc_file()
- self.config.set('project', 'ioc_file', str(ioc_file))
+ self.ioc_file = self._find_ioc_file()
+ self.config.set('project', 'ioc_file', str(self.ioc_file))
cubemx_script_template = string.Template(self.config.get('project', 'cubemx_script_content'))
- cubemx_script_content = cubemx_script_template.substitute(project_path=self.project_path,
- cubemx_ioc_full_filename=self.config.get('project', 'ioc_file'))
+ cubemx_script_content = cubemx_script_template.substitute(project_path=self.path,
+ cubemx_ioc_full_filename=self.ioc_file)
self.config.set('project', 'cubemx_script_content', cubemx_script_content)
+ # General rule: given parameter takes precedence over the saved one
board = ''
if 'board' in parameters and parameters['board'] is not None:
- try:
- board = self._resolve_board(parameters['board'])
- except Exception as e:
- logger.warning(e)
+ if parameters['board'] in stm32pio.util.get_platformio_boards():
+ board = parameters['board']
+ else:
+ self.logger.warning(f"'{parameters['board']}' was not found in PlatformIO. "
+ "Run 'platformio boards' for possible names")
self.config.set('project', 'board', board)
elif self.config.get('project', 'board', fallback=None) is None:
self.config.set('project', 'board', board)
if save_on_destruction:
- self._finalizer = weakref.finalize(self, self.config.save)
+ # Save the config on an instance destruction
+ self._finalizer = weakref.finalize(self, self._save_config, self.config, self.path, self.logger)
+
+
+ def __repr__(self):
+ return f"Stm32pio project: {str(self.path)}"
@property
def state(self) -> ProjectState:
"""
- Property returning the current state of the project. Calculated at every request
-
- Returns:
- enum value representing a project state
+ Constructing and returning the current state of the project (tweaked dict, see ProjectState docs)
"""
- logger.debug("calculating the project state...")
- logger.debug(f"project content: {[item.name for item in self.project_path.iterdir()]}")
+ # self.logger.debug(f"project content: {[item.name for item in self.path.iterdir()]}")
try:
platformio_ini_is_patched = self.platformio_ini_is_patched()
- except:
+ except (FileNotFoundError, ValueError):
platformio_ini_is_patched = False
- states_conditions = collections.OrderedDict()
- # Fill the ordered dictionary with the conditions results
- states_conditions[ProjectState.UNDEFINED] = [True]
- states_conditions[ProjectState.INITIALIZED] = [
- self.project_path.joinpath(stm32pio.settings.config_file_name).is_file()]
- states_conditions[ProjectState.GENERATED] = [self.project_path.joinpath('Inc').is_dir() and
- len(list(self.project_path.joinpath('Inc').iterdir())) > 0,
- self.project_path.joinpath('Src').is_dir() and
- len(list(self.project_path.joinpath('Src').iterdir())) > 0]
- states_conditions[ProjectState.PIO_INITIALIZED] = [
- self.project_path.joinpath('platformio.ini').is_file() and
- self.project_path.joinpath('platformio.ini').stat().st_size > 0]
- states_conditions[ProjectState.PATCHED] = [
- platformio_ini_is_patched, not self.project_path.joinpath('include').is_dir()]
- states_conditions[ProjectState.BUILT] = [
- self.project_path.joinpath('.pio').is_dir() and
- any([item.is_file() for item in self.project_path.joinpath('.pio').rglob('*firmware*')])]
-
- # Use (1,0) instead of (True,False) because on debug printing it looks better
- conditions_results = []
- for state, conditions in states_conditions.items():
- conditions_results.append(1 if all(condition is True for condition in conditions) else 0)
-
- # Put away unnecessary processing as the string still will be formed even if the logging level doesn't allow
- # propagation of this message
- if logger.getEffectiveLevel() <= logging.DEBUG:
- states_info_str = '\n'.join(
- f"{state.name:20}{conditions_results[state.value - 1]}" for state in ProjectState)
- logger.debug(f"determined states:\n{states_info_str}")
-
- # Search for a consecutive sequence of 1's and find the last of them. For example, if the array is
- # [1,1,0,1,0,0]
- # ^
- # we should consider 1 as the last index
- last_true_index = 0 # ProjectState.UNDEFINED is always True, use as a start value
- for index, value in enumerate(conditions_results):
- if value == 1:
- last_true_index = index
- else:
- break
-
- # Fall back to the UNDEFINED state if we have breaks in conditions results array. For example, in [1,1,0,1,0,0]
- # we still return UNDEFINED as it doesn't look like a correct combination of files actually
- if 1 in conditions_results[last_true_index + 1:]:
- project_state = ProjectState.UNDEFINED
- else:
- project_state = ProjectState(last_true_index + 1)
-
- return project_state
+ # Create the temporary ordered dictionary and fill it with the conditions results arrays
+ stages_conditions = collections.OrderedDict()
+ stages_conditions[ProjectStage.UNDEFINED] = [True]
+ stages_conditions[ProjectStage.EMPTY] = [self.ioc_file.is_file()]
+ stages_conditions[ProjectStage.INITIALIZED] = [self.path.joinpath(stm32pio.settings.config_file_name).is_file()]
+ stages_conditions[ProjectStage.GENERATED] = [self.path.joinpath('Inc').is_dir() and
+ len(list(self.path.joinpath('Inc').iterdir())) > 0,
+ self.path.joinpath('Src').is_dir() and
+ len(list(self.path.joinpath('Src').iterdir())) > 0]
+ stages_conditions[ProjectStage.PIO_INITIALIZED] = [
+ self.path.joinpath('platformio.ini').is_file() and
+ self.path.joinpath('platformio.ini').stat().st_size > 0]
+ stages_conditions[ProjectStage.PATCHED] = [
+ platformio_ini_is_patched, not self.path.joinpath('include').is_dir()]
+ # Hidden folder! Can be not visible in your familiar file manager and cause a confusion
+ stages_conditions[ProjectStage.BUILT] = [
+ self.path.joinpath('.pio').is_dir() and
+ any([item.is_file() for item in self.path.joinpath('.pio').rglob('*firmware*')])]
+
+ # Fold arrays and save results in ProjectState instance
+ conditions_results = ProjectState()
+ for state, conditions in stages_conditions.items():
+ conditions_results[state] = all(condition is True for condition in conditions)
+
+ return conditions_results
def _find_ioc_file(self) -> pathlib.Path:
"""
- Find and return an .ioc file. If there are more than one, return first. If no .ioc file is present raise
+ Find and return an .ioc file. If there are more than one return first. If no .ioc file is present raise
FileNotFoundError exception
Returns:
absolute path to the .ioc file
"""
+ error_message = "not found: CubeMX project .ioc file"
+
ioc_file = self.config.get('project', 'ioc_file', fallback=None)
if ioc_file:
- ioc_file = pathlib.Path(ioc_file).resolve()
- logger.debug(f"use {ioc_file.name} file from the INI config")
+ ioc_file = pathlib.Path(ioc_file).expanduser().resolve()
+ self.logger.debug(f"use {ioc_file.name} file from the INI config")
+ if not ioc_file.is_file():
+ raise FileNotFoundError(error_message)
return ioc_file
else:
- logger.debug("searching for any .ioc file...")
- candidates = list(self.project_path.glob('*.ioc'))
- if len(candidates) == 0: # good candidate for the new Python 3.8 assignment expressions feature :)
- raise FileNotFoundError("not found: CubeMX project .ioc file")
+ self.logger.debug("searching for any .ioc file...")
+ candidates = list(self.path.glob('*.ioc'))
+ if len(candidates) == 0: # good candidate for the new Python 3.8 assignment expression feature :)
+ raise FileNotFoundError(error_message)
elif len(candidates) == 1:
- logger.debug(f"{candidates[0].name} is selected")
+ self.logger.debug(f"{candidates[0].name} is selected")
return candidates[0]
else:
- logger.warning(f"there are multiple .ioc files, {candidates[0].name} is selected")
+ self.logger.warning(f"there are multiple .ioc files, {candidates[0].name} is selected")
return candidates[0]
- def _load_config_file(self) -> Config:
+ def _load_config(self) -> configparser.ConfigParser:
"""
- Prepare configparser config for the project. First, read the default config and then mask these values with user
- ones
+ Prepare ConfigParser config for the project. First, read the default config and then mask these values with user
+ ones.
Returns:
- custom configparser.ConfigParser instance
+ new configparser.ConfigParser instance
"""
- logger.debug(f"searching for {stm32pio.settings.config_file_name}...")
- stm32pio_ini = self.project_path.joinpath(stm32pio.settings.config_file_name)
+ self.logger.debug(f"searching for {stm32pio.settings.config_file_name}...")
- config = Config(self.project_path, interpolation=None)
+ config = configparser.ConfigParser(interpolation=None)
# Fill with default values
config.read_dict(stm32pio.settings.config_default)
# Then override by user values (if exist)
- config.read(str(stm32pio_ini))
+ config.read(str(self.path.joinpath(stm32pio.settings.config_file_name)))
# Put away unnecessary processing as the string still will be formed even if the logging level doesn't allow
# propagation of this message
- if logger.getEffectiveLevel() <= logging.DEBUG:
+ if self.logger.isEnabledFor(logging.DEBUG):
debug_str = 'resolved config:'
for section in config.sections():
debug_str += f"\n========== {section} ==========\n"
for value in config.items(section):
debug_str += f"{value}\n"
- logger.debug(debug_str)
+ self.logger.debug(debug_str)
return config
+ @staticmethod
+ def _save_config(config: configparser.ConfigParser, path: pathlib.Path, logger: logging.Logger) -> int:
+ """
+ Writes ConfigParser config to the file path and logs using Logger logger.
+
+ We declare this helper function which can be safely invoked by both internal methods and outer code. The latter
+ case is suitable for using in weakref' finalizer objects as one of its main requirement is to not keep
+ references to the destroyable object in any of the finalizer argument so the ordinary bound class method does
+ not fit well.
+
+ Returns:
+ 0 on success, -1 otherwise
+ """
+ try:
+ with path.joinpath(stm32pio.settings.config_file_name).open(mode='w') as config_file:
+ config.write(config_file)
+ logger.debug("stm32pio.ini config file has been saved")
+ return 0
+ except Exception as e:
+ logger.warning(f"cannot save the config: {e}", exc_info=logger.isEnabledFor(logging.DEBUG))
+ return -1
+
+ def save_config(self, parameters: dict = None) -> int:
+ """
+ Invokes base _save_config function. Preliminarily, updates the config with given parameters dictionary. It
+ should has the following format:
+ {
+ 'section1_name': {
+ 'key1': 'value1',
+ 'key2': 'value2'
+ },
+ ...
+ }
+
+ Returns:
+ passes forward _save_config result
+ """
+ if parameters is not None:
+ for section_name, section_value in parameters.items():
+ for key, value in section_value.items():
+ self.config.set(section_name, key, value)
+ return self._save_config(self.config, self.path, self.logger)
+
@staticmethod
def _resolve_project_path(dirty_path: str) -> pathlib.Path:
"""
- Handle 'path/to/proj' and 'path/to/proj/', '.' (current directory) and other cases
+ Handle 'path/to/proj', 'path/to/proj/', '.', '../proj' and other cases
Args:
dirty_path (str): some directory in the filesystem
@@ -275,112 +358,86 @@ def _resolve_project_path(dirty_path: str) -> pathlib.Path:
return resolved_path
- def _resolve_board(self, board: str) -> str:
- """
- Check if given board is a correct board name in the PlatformIO database
-
- Args:
- board: string representing PlatformIO board name (for example, 'nucleo_f031k6')
-
- Returns:
- same board that has been given if it was found, raise an exception otherwise
- """
-
- logger.debug("searching for PlatformIO board...")
- result = subprocess.run([self.config.get('app', 'platformio_cmd'), 'boards'], encoding='utf-8',
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- # Or, for Python 3.7 and above:
- # result = subprocess.run(['platformio', 'boards'], encoding='utf-8', capture_output=True)
- if result.returncode == 0:
- if board not in result.stdout.split():
- raise Exception("wrong PlatformIO STM32 board. Run 'platformio boards' for possible names")
- else:
- logger.debug(f"PlatformIO board {board} was found")
- return board
- else:
- raise Exception("failed to search for PlatformIO boards")
-
-
def generate_code(self) -> int:
"""
- Call STM32CubeMX app as a 'java -jar' file to generate the code from the .ioc file. Pass commands to the
+ Call STM32CubeMX app as 'java -jar' file to generate the code from the .ioc file. Pass commands to the
STM32CubeMX in a temp file
Returns:
return code on success, raises an exception otherwise
"""
- # Use mkstemp() instead of higher-level API for compatibility with Windows (see tempfile docs for more details)
+ # Use mkstemp() instead of the higher-level API for the compatibility with the Windows (see tempfile docs for
+ # more details)
cubemx_script_file, cubemx_script_name = tempfile.mkstemp()
# We should necessarily remove the temp directory, so do not let any exception break our plans
try:
# buffering=0 leads to the immediate flushing on writing
with open(cubemx_script_file, mode='w+b', buffering=0) as cubemx_script:
- # encode since mode='w+b'
+ # should encode, since mode='w+b'
cubemx_script.write(self.config.get('project', 'cubemx_script_content').encode())
- logger.info("starting to generate a code from the CubeMX .ioc file...")
+ self.logger.info("starting to generate a code from the CubeMX .ioc file...")
command_arr = [self.config.get('app', 'java_cmd'), '-jar', self.config.get('app', 'cubemx_cmd'), '-q',
- cubemx_script_name, '-s'] # -q: read commands from file, -s: silent performance
- if logger.getEffectiveLevel() <= logging.DEBUG:
- result = subprocess.run(command_arr)
- else:
- result = subprocess.run(command_arr, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- # Or, for Python 3.7 and above:
- # result = subprocess.run(command_arr, capture_output=True)
+ cubemx_script_name, '-s'] # -q: read the commands from the file, -s: silent performance
+ # Redirect the output of the subprocess into the logging module (with DEBUG level)
+ with stm32pio.util.LogPipe(self.logger, logging.DEBUG) as log_pipe:
+ result = subprocess.run(command_arr, stdout=log_pipe, stderr=log_pipe)
except Exception as e:
- raise e # re-raise an exception after the final block
+ raise e # re-raise an exception after the 'finally' block
finally:
pathlib.Path(cubemx_script_name).unlink()
if result.returncode == 0:
- logger.info("successful code generation")
+ self.logger.info("successful code generation")
return result.returncode
else:
- logger.error(f"code generation error (CubeMX return code is {result.returncode}).\n"
- "Enable a verbose output or try to generate a code from the CubeMX itself.")
+ self.logger.error(f"code generation error (CubeMX return code is {result.returncode}).\n"
+ "Enable a verbose output or try to generate a code from the CubeMX itself.")
raise Exception("code generation error")
+
def pio_init(self) -> int:
"""
Call PlatformIO CLI to initialize a new project. It uses parameters (path, board) collected before so the
- confirmation of the data presence is on a user
+ confirmation of the data presence is lying on the invoking code
Returns:
return code of the PlatformIO on success, raises an exception otherwise
"""
- logger.info("starting PlatformIO project initialization...")
+ self.logger.info("starting PlatformIO project initialization...")
- platformio_ini_file = self.project_path.joinpath('platformio.ini')
+ platformio_ini_file = self.path.joinpath('platformio.ini')
if platformio_ini_file.is_file() and platformio_ini_file.stat().st_size > 0:
- logger.warning("'platformio.ini' file is already exist")
+ self.logger.warning("'platformio.ini' file is already exist")
- command_arr = [self.config.get('app', 'platformio_cmd'), 'init', '-d', str(self.project_path), '-b',
+ command_arr = [self.config.get('app', 'platformio_cmd'), 'init', '-d', str(self.path), '-b',
self.config.get('project', 'board'), '-O', 'framework=stm32cube']
- if logger.getEffectiveLevel() > logging.DEBUG:
+ if not self.logger.isEnabledFor(logging.DEBUG):
command_arr.append('--silent')
- result = subprocess.run(command_arr, encoding='utf-8', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ result = subprocess.run(command_arr, encoding='utf-8', stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
error_msg = "PlatformIO project initialization error"
if result.returncode == 0:
# PlatformIO returns 0 even on some errors (e.g. no '--board' argument)
- for output in [result.stdout, result.stderr]:
- if 'ERROR' in output.upper():
- logger.error(output)
- raise Exception(error_msg)
- logger.info("successful PlatformIO project initialization")
+ if 'error' in result.stdout.lower():
+ self.logger.error(result.stdout)
+ raise Exception('\n' + error_msg)
+ self.logger.debug(result.stdout, 'from_subprocess')
+ self.logger.info("successful PlatformIO project initialization")
return result.returncode
else:
+ self.logger.error(result.stdout)
raise Exception(error_msg)
def platformio_ini_is_patched(self) -> bool:
"""
Check whether 'platformio.ini' config file is patched or not. It doesn't check for complete project patching
- (e.g. unnecessary folders deletion). Throws an error on non-existing file and on incorrect patch or file
+ (e.g. unnecessary folders deletion). Throws errors on non-existing file and on incorrect patch or file
Returns:
boolean indicating a result
@@ -388,29 +445,30 @@ def platformio_ini_is_patched(self) -> bool:
platformio_ini = configparser.ConfigParser(interpolation=None)
try:
- if len(platformio_ini.read(self.project_path.joinpath('platformio.ini'))) == 0:
+ if len(platformio_ini.read(self.path.joinpath('platformio.ini'))) == 0:
raise FileNotFoundError("not found: 'platformio.ini' file")
except FileNotFoundError as e:
raise e
except Exception as e:
- raise Exception("'platformio.ini' file is incorrect") from e
+ # Re-raise parsing exceptions as ValueError
+ raise ValueError("'platformio.ini' file is incorrect") from e
patch_config = configparser.ConfigParser(interpolation=None)
try:
patch_config.read_string(self.config.get('project', 'platformio_ini_patch_content'))
except Exception as e:
- raise Exception("Desired patch content is invalid (should satisfy INI-format requirements)") from e
+ raise ValueError("Desired patch content is invalid (should satisfy INI-format requirements)") from e
for patch_section in patch_config.sections():
if platformio_ini.has_section(patch_section):
for patch_key, patch_value in patch_config.items(patch_section):
platformio_ini_value = platformio_ini.get(patch_section, patch_key, fallback=None)
if platformio_ini_value != patch_value:
- logger.debug(f"[{patch_section}]{patch_key}: patch value is\n{patch_value}\nbut "
- f"platformio.ini contains\n{platformio_ini_value}")
+ self.logger.debug(f"[{patch_section}]{patch_key}: patch value is\n {patch_value}\nbut "
+ f"platformio.ini contains\n {platformio_ini_value}")
return False
else:
- logger.debug(f"platformio.ini has not {patch_section} section")
+ self.logger.debug(f"platformio.ini has no '{patch_section}' section")
return False
return True
@@ -418,18 +476,18 @@ def platformio_ini_is_patched(self) -> bool:
def patch(self) -> None:
"""
Patch platformio.ini file by a user's patch. By default, it sets the created earlier (by CubeMX 'Src' and 'Inc')
- folders as sources. configparser doesn't preserve any comments unfortunately so keep in mid that all of them
+ folders as sources. configparser doesn't preserve any comments unfortunately so keep in mind that all of them
will be lost at this stage. Also, the order may be violated. In the end, remove old empty folders
"""
- logger.debug("patching 'platformio.ini' file...")
+ self.logger.debug("patching 'platformio.ini' file...")
if self.platformio_ini_is_patched():
- logger.info("'platformio.ini' has been already patched")
+ self.logger.info("'platformio.ini' has been already patched")
else:
# Existing .ini file
platformio_ini_config = configparser.ConfigParser(interpolation=None)
- platformio_ini_config.read(self.project_path.joinpath('platformio.ini'))
+ platformio_ini_config.read(self.path.joinpath('platformio.ini'))
# Our patch has the config format too
patch_config = configparser.ConfigParser(interpolation=None)
@@ -438,28 +496,31 @@ def patch(self) -> None:
# Merge 2 configs
for patch_section in patch_config.sections():
if not platformio_ini_config.has_section(patch_section):
- logger.debug(f"[{patch_section}] section was added")
+ self.logger.debug(f"[{patch_section}] section was added")
platformio_ini_config.add_section(patch_section)
for patch_key, patch_value in patch_config.items(patch_section):
- logger.debug(f"set [{patch_section}]{patch_key} = {patch_value}")
+ self.logger.debug(f"set [{patch_section}]{patch_key} = {patch_value}")
platformio_ini_config.set(patch_section, patch_key, patch_value)
# Save, overwriting the original file (deletes all comments!)
- with self.project_path.joinpath('platformio.ini').open(mode='w') as platformio_ini_file:
+ with self.path.joinpath('platformio.ini').open(mode='w') as platformio_ini_file:
platformio_ini_config.write(platformio_ini_file)
-
- logger.info("'platformio.ini' has been patched")
+ self.logger.debug("'platformio.ini' has been patched")
try:
- shutil.rmtree(self.project_path.joinpath('include'))
+ shutil.rmtree(self.path.joinpath('include'))
+ self.logger.debug("'include' folder has been removed")
except:
- logger.info("cannot delete 'include' folder", exc_info=logger.getEffectiveLevel() <= logging.DEBUG)
+ self.logger.info("cannot delete 'include' folder", exc_info=self.logger.isEnabledFor(logging.DEBUG))
# Remove 'src' directory too but on case-sensitive file systems 'Src' == 'src' == 'SRC' so we need to check
- if not self.project_path.joinpath('SRC').is_dir():
+ if not self.path.joinpath('SRC').is_dir():
try:
- shutil.rmtree(self.project_path.joinpath('src'))
+ shutil.rmtree(self.path.joinpath('src'))
+ self.logger.debug("'src' folder has been removed")
except:
- logger.info("cannot delete 'src' folder", exc_info=logger.getEffectiveLevel() <= logging.DEBUG)
+ self.logger.info("cannot delete 'src' folder", exc_info=self.logger.isEnabledFor(logging.DEBUG))
+
+ self.logger.info("project has been patched")
def start_editor(self, editor_command: str) -> int:
@@ -475,17 +536,18 @@ def start_editor(self, editor_command: str) -> int:
passes a return code of the command
"""
- logger.info(f"starting an editor '{editor_command}'...")
+ self.logger.info(f"starting an editor '{editor_command}'...")
try:
# Works unstable on some Windows 7 systems, but correct on latest Win7 and Win10...
- # result = subprocess.run([editor_command, str(self.project_path)], check=True)
- result = subprocess.run(f"{editor_command} {str(self.project_path)}", shell=True, check=True,
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ # result = subprocess.run([editor_command, str(self.path)], check=True)
+ result = subprocess.run(f"{editor_command} {str(self.path)}", shell=True, check=True,
+ stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ self.logger.debug(result.stdout, 'from_subprocess')
+
return result.returncode
except subprocess.CalledProcessError as e:
- output = e.stdout if e.stderr is None else e.stderr
- logger.error(f"failed to start the editor {editor_command}: {output}")
+ self.logger.error(f"failed to start the editor {editor_command}: {e.stdout}")
return e.returncode
@@ -498,15 +560,18 @@ def build(self) -> int:
passes a return code of the PlatformIO
"""
- command_arr = [self.config.get('app', 'platformio_cmd'), 'run', '-d', str(self.project_path)]
- if logger.getEffectiveLevel() > logging.DEBUG:
+ self.logger.info("starting PlatformIO project build...")
+
+ command_arr = [self.config.get('app', 'platformio_cmd'), 'run', '-d', str(self.path)]
+ if not self.logger.isEnabledFor(logging.DEBUG):
command_arr.append('--silent')
- result = subprocess.run(command_arr)
+ with stm32pio.util.LogPipe(self.logger, logging.DEBUG) as log_pipe:
+ result = subprocess.run(command_arr, stdout=log_pipe, stderr=log_pipe)
if result.returncode == 0:
- logger.info("successful PlatformIO build")
+ self.logger.info("successful PlatformIO build")
else:
- logger.error("PlatformIO build error")
+ self.logger.error("PlatformIO build error")
return result.returncode
@@ -515,13 +580,13 @@ def clean(self) -> None:
Clean-up the project folder preserving only an '.ioc' file
"""
- for child in self.project_path.iterdir():
- if child.name != f"{self.project_path.name}.ioc":
+ for child in self.path.iterdir():
+ if child.name != f"{self.path.name}.ioc":
if child.is_dir():
shutil.rmtree(child, ignore_errors=True)
- logger.debug(f"del {child}")
+ self.logger.debug(f"del {child}/")
elif child.is_file():
child.unlink()
- logger.debug(f"del {child}")
+ self.logger.debug(f"del {child}")
- logger.info("project has been cleaned")
+ self.logger.info("project has been cleaned")
diff --git a/stm32pio/settings.py b/stm32pio/settings.py
index 9e54f0c..1ad0172 100644
--- a/stm32pio/settings.py
+++ b/stm32pio/settings.py
@@ -27,12 +27,18 @@
},
project={
# (default is OK) See CubeMX user manual PDF (UM1718) to get other useful options
- 'cubemx_script_content': "config load $cubemx_ioc_full_filename\ngenerate code $project_path\nexit",
+ 'cubemx_script_content': inspect.cleandoc('''
+ config load $cubemx_ioc_full_filename
+ generate code $project_path
+ exit
+ ''') + '\n',
# Override the defaults to comply with CubeMX project structure. This should meet INI-style requirements. You
# can include existing sections, too (e.g.
+ #
# [env:nucleo_f031k6]
- # key=value
+ # key = value
+ #
# will add a 'key' parameter)
'platformio_ini_patch_content': inspect.cleandoc('''
[platformio]
@@ -44,4 +50,4 @@
config_file_name = 'stm32pio.ini'
-log_function_fieldwidth = 26
+log_fieldwidth_function = 26 # TODO: can be calculated actually (longest name
diff --git a/stm32pio/tests/test.py b/stm32pio/tests/test.py
index 445ec8a..6e1e614 100755
--- a/stm32pio/tests/test.py
+++ b/stm32pio/tests/test.py
@@ -1,14 +1,18 @@
"""
-'pyenv' was used to perform tests with different Python versions under Ubuntu:
+NOTE: make sure the test project tree is clean before running the tests!
+
+'pyenv' was used to perform tests with different Python versions (under Ubuntu):
https://www.tecmint.com/pyenv-install-and-manage-multiple-python-versions-in-linux/
-To get the test coverage install and use 'coverage':
+To get the test coverage install and use 'coverage' package:
$ coverage run -m stm32pio.tests.test -b
$ coverage html
"""
import configparser
+import contextlib
import inspect
+import io
import pathlib
import platform
import re
@@ -22,6 +26,7 @@
import stm32pio.app
import stm32pio.lib
import stm32pio.settings
+import stm32pio.util
STM32PIO_MAIN_SCRIPT: str = inspect.getfile(stm32pio.app) # absolute path to the main stm32pio script
@@ -36,12 +41,15 @@
# proceeding)
TEST_PROJECT_BOARD = 'nucleo_f031k6'
-# Instantiate a temporary folder on every fixture run. It is used across all tests and is deleted on shutdown
+# Instantiate a temporary folder on every test suite run. It is used across all the tests and is deleted on shutdown
+# automatically
temp_dir = tempfile.TemporaryDirectory()
FIXTURE_PATH = pathlib.Path(temp_dir.name).joinpath(TEST_PROJECT_PATH.name)
+
print(f"The file of 'stm32pio.app' module: {STM32PIO_MAIN_SCRIPT}")
print(f"Python executable: {PYTHON_EXEC} {sys.version}")
print(f"Temp test fixture path: {FIXTURE_PATH}")
+print()
class CustomTestCase(unittest.TestCase):
@@ -194,19 +202,19 @@ def test_run_editor(self):
}
}
- for command, name in editors.items():
+ for editor, editor_process_names in editors.items():
# Look for the command presence in the system so we test only installed editors
if platform.system() == 'Windows':
- command_str = f"where {command} /q"
+ command_str = f"where {editor} /q"
else:
- command_str = f"command -v {command}"
+ command_str = f"command -v {editor}"
editor_exists = False
if subprocess.run(command_str, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).returncode == 0:
editor_exists = True
if editor_exists:
- with self.subTest(command=command, name=name[platform.system()]):
- project.start_editor(command)
+ with self.subTest(command=editor, name=editor_process_names[platform.system()]):
+ project.start_editor(editor)
time.sleep(1) # wait a little bit for app to start
@@ -219,7 +227,7 @@ def test_run_editor(self):
encoding='utf-8')
# Or, for Python 3.7 and above:
# result = subprocess.run(command_arr, capture_output=True, encoding='utf-8')
- self.assertIn(name[platform.system()], result.stdout)
+ self.assertIn(editor_process_names[platform.system()], result.stdout)
def test_init_path_not_found_should_raise(self):
"""
@@ -240,7 +248,7 @@ def test_save_config(self):
# 'board' is non-default, 'project'-section parameter
project = stm32pio.lib.Stm32pio(FIXTURE_PATH, parameters={'board': TEST_PROJECT_BOARD},
save_on_destruction=False)
- project.config.save()
+ project.save_config()
self.assertTrue(FIXTURE_PATH.joinpath(stm32pio.settings.config_file_name).is_file(),
msg=f"{stm32pio.settings.config_file_name} file hasn't been created")
@@ -257,6 +265,13 @@ def test_save_config(self):
self.assertEqual(config.get('project', 'board', fallback="Not found"), TEST_PROJECT_BOARD,
msg="'board' has not been set")
+ def test_get_platformio_boards(self):
+ """
+ PlatformIO identifiers of boards are requested using PlatformIO Python API (not sure it can be called public,
+ though...)
+ """
+ self.assertIsInstance(stm32pio.util.get_platformio_boards(), list)
+
class TestIntegration(CustomTestCase):
"""
@@ -344,44 +359,42 @@ def test_regenerate_code(self):
# Re-generate CubeMX project
project.generate_code()
- # Check if added information is preserved
+ # Check if added information has been preserved
for test_content, after_regenerate_content in [(test_content_1, test_file_1.read_text()),
(test_content_2, test_file_2.read_text())]:
with self.subTest(msg=f"User content hasn't been preserved in {after_regenerate_content}"):
self.assertIn(test_content, after_regenerate_content)
- # main_c_after_regenerate_content = test_file_1.read_text()
- # my_header_h_after_regenerate_content = test_file_2.read_text()
- # self.assertIn(test_content_1, main_c_after_regenerate_content,
- # msg=f"User content hasn't been preserved after regeneration in {test_file_1}")
- # self.assertIn(test_content_2, my_header_h_after_regenerate_content,
- # msg=f"User content hasn't been preserved after regeneration in {test_file_2}")
-
- def test_get_state(self):
+ def test_current_stage(self):
"""
Go through the sequence of states emulating the real-life project lifecycle
"""
project = stm32pio.lib.Stm32pio(FIXTURE_PATH, parameters={'board': TEST_PROJECT_BOARD},
save_on_destruction=False)
- self.assertEqual(project.state, stm32pio.lib.ProjectState.UNDEFINED)
+ self.assertEqual(project.state.current_stage, stm32pio.lib.ProjectStage.EMPTY)
- project.config.save()
- self.assertEqual(project.state, stm32pio.lib.ProjectState.INITIALIZED)
+ project.save_config()
+ self.assertEqual(project.state.current_stage, stm32pio.lib.ProjectStage.INITIALIZED)
project.generate_code()
- self.assertEqual(project.state, stm32pio.lib.ProjectState.GENERATED)
+ self.assertEqual(project.state.current_stage, stm32pio.lib.ProjectStage.GENERATED)
project.pio_init()
- self.assertEqual(project.state, stm32pio.lib.ProjectState.PIO_INITIALIZED)
+ self.assertEqual(project.state.current_stage, stm32pio.lib.ProjectStage.PIO_INITIALIZED)
project.patch()
- self.assertEqual(project.state, stm32pio.lib.ProjectState.PATCHED)
+ self.assertEqual(project.state.current_stage, stm32pio.lib.ProjectStage.PATCHED)
project.build()
- self.assertEqual(project.state, stm32pio.lib.ProjectState.BUILT)
+ self.assertEqual(project.state.current_stage, stm32pio.lib.ProjectStage.BUILT)
project.clean()
- self.assertEqual(project.state, stm32pio.lib.ProjectState.UNDEFINED)
+ self.assertEqual(project.state.current_stage, stm32pio.lib.ProjectStage.EMPTY)
+
+ # Should be UNDEFINED when the project is messed up
+ project.pio_init()
+ self.assertEqual(project.state.current_stage, stm32pio.lib.ProjectStage.UNDEFINED)
+ self.assertFalse(project.state.is_consistent)
class TestCLI(CustomTestCase):
@@ -468,24 +481,26 @@ def test_verbose(self):
^(?=(DEBUG|INFO|WARNING|ERROR|CRITICAL) {0,4})(?=.{8} (?=(build|pio_init|...) {0,26})(?=.{26} [^ ]))
"""
project = stm32pio.lib.Stm32pio(FIXTURE_PATH, save_on_destruction=False)
- methods = [method[0] for method in inspect.getmembers(project, predicate=inspect.ismethod)]
- methods.append('main')
+ methods = [member[0] for member in inspect.getmembers(project, predicate=inspect.ismethod)] + ['main']
- result = subprocess.run([PYTHON_EXEC, STM32PIO_MAIN_SCRIPT, '-v', 'generate', '-d', str(FIXTURE_PATH)],
- encoding='utf-8', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ buffer_stdout, buffer_stderr = io.StringIO(), io.StringIO()
+ with contextlib.redirect_stdout(buffer_stdout), contextlib.redirect_stderr(buffer_stderr):
+ return_code = stm32pio.app.main(sys_argv=['-v', 'generate', '-d', str(FIXTURE_PATH)])
+
+ self.assertEqual(return_code, 0, msg="Non-zero return code")
+ # stderr and not stdout contains the actual output (by default for the logging module)
+ self.assertEqual(len(buffer_stdout.getvalue()), 0,
+ msg="Process has printed something directly into STDOUT bypassing logging")
+ self.assertIn('DEBUG', buffer_stderr.getvalue(), msg="Verbose logging output hasn't been enabled on stderr")
- self.assertEqual(result.returncode, 0, msg="Non-zero return code")
- # Somehow stderr and not stdout contains the actual output but we check both
- self.assertTrue('DEBUG' in result.stderr or 'DEBUG' in result.stdout,
- msg="Verbose logging output hasn't been enabled on stderr")
# Inject all methods' names in the regex. Inject the width of field in a log format string
regex = re.compile("^(?=(DEBUG) {0,4})(?=.{8} (?=(" + '|'.join(methods) + ") {0," +
- str(stm32pio.settings.log_function_fieldwidth) + "})(?=.{" +
- str(stm32pio.settings.log_function_fieldwidth) + "} [^ ]))",
- flags=re.MULTILINE)
- self.assertGreaterEqual(len(re.findall(regex, result.stderr)), 1, msg="Logs messages doesn't match the format")
+ str(stm32pio.settings.log_fieldwidth_function) + "})(?=.{" +
+ str(stm32pio.settings.log_fieldwidth_function) + "} [^ ]))", flags=re.MULTILINE)
+ self.assertGreaterEqual(len(re.findall(regex, buffer_stderr.getvalue())), 1,
+ msg="Logs messages doesn't match the format")
- self.assertIn('Starting STM32CubeMX', result.stdout, msg="STM32CubeMX didn't print its logs")
+ self.assertIn('Starting STM32CubeMX', buffer_stderr.getvalue(), msg="STM32CubeMX has not printed its logs")
def test_non_verbose(self):
"""
@@ -498,25 +513,27 @@ def test_non_verbose(self):
methods = [method[0] for method in inspect.getmembers(project, predicate=inspect.ismethod)]
methods.append('main')
- result = subprocess.run([PYTHON_EXEC, STM32PIO_MAIN_SCRIPT, 'generate', '-d', str(FIXTURE_PATH)],
- encoding='utf-8', stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ buffer_stdout, buffer_stderr = io.StringIO(), io.StringIO()
+ with contextlib.redirect_stdout(buffer_stdout), contextlib.redirect_stderr(buffer_stderr):
+ return_code = stm32pio.app.main(sys_argv=['generate', '-d', str(FIXTURE_PATH)])
- self.assertEqual(result.returncode, 0, msg="Non-zero return code")
- self.assertNotIn('DEBUG', result.stderr, msg="Verbose logging output has been enabled on stderr")
- self.assertNotIn('DEBUG', result.stdout, msg="Verbose logging output has been enabled on stdout")
+ self.assertEqual(return_code, 0, msg="Non-zero return code")
+ # stderr and not stdout contains the actual output (by default for the logging module)
+ self.assertNotIn('DEBUG', buffer_stderr.getvalue(), msg="Verbose logging output has been enabled on stderr")
+ self.assertEqual(len(buffer_stdout.getvalue()), 0, msg="All app output should flow through the logging module")
regex = re.compile("^(?=(INFO) {0,4})(?=.{8} ((?!( |" + '|'.join(methods) + "))))", flags=re.MULTILINE)
- self.assertGreaterEqual(len(re.findall(regex, result.stderr)), 1,
+ self.assertGreaterEqual(len(re.findall(regex, buffer_stderr.getvalue())), 1,
msg="Logs messages doesn't match the format")
- self.assertNotIn('Starting STM32CubeMX', result.stdout, msg="STM32CubeMX printed its logs")
+ self.assertNotIn('Starting STM32CubeMX', buffer_stderr.getvalue(), msg="STM32CubeMX has printed its logs")
def test_init(self):
"""
Check for config creation and parameters presence
"""
result = subprocess.run([PYTHON_EXEC, STM32PIO_MAIN_SCRIPT, 'init', '-d', str(FIXTURE_PATH),
- '-b', TEST_PROJECT_BOARD])
+ '-b', TEST_PROJECT_BOARD], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.assertEqual(result.returncode, 0, msg="Non-zero return code")
self.assertTrue(FIXTURE_PATH.joinpath(stm32pio.settings.config_file_name).is_file(),
@@ -531,6 +548,30 @@ def test_init(self):
self.assertEqual(config.get('project', 'board', fallback="Not found"), TEST_PROJECT_BOARD,
msg="'board' has not been set")
+ def test_status(self):
+ """
+ Test the output returning by the app on a request to the 'status' command
+ """
+
+ buffer_stdout = io.StringIO()
+ with contextlib.redirect_stdout(buffer_stdout), contextlib.redirect_stderr(None):
+ return_code = stm32pio.app.main(sys_argv=['status', '-d', str(FIXTURE_PATH)])
+
+ self.assertEqual(return_code, 0, msg="Non-zero return code")
+
+ matches_counter = 0
+ last_stage_pos = -1
+ for stage in stm32pio.lib.ProjectStage:
+ if stage != stm32pio.lib.ProjectStage.UNDEFINED:
+ match = re.search(r"^((\[ \])|(\[\*\])) {2}" + str(stage) + '$', buffer_stdout.getvalue(), re.MULTILINE)
+ self.assertTrue(match, msg="Status information was not found on STDOUT")
+ if match:
+ matches_counter += 1
+ self.assertGreater(match.start(), last_stage_pos, msg="The order of stages is messed up")
+ last_stage_pos = match.start()
+
+ self.assertEqual(matches_counter, len(stm32pio.lib.ProjectStage) - 1)
+
if __name__ == '__main__':
unittest.main()
diff --git a/stm32pio/util.py b/stm32pio/util.py
new file mode 100644
index 0000000..cfe481c
--- /dev/null
+++ b/stm32pio/util.py
@@ -0,0 +1,129 @@
+"""
+Some auxiliary entities not falling into other categories
+"""
+
+import logging
+import os
+import threading
+from typing import List
+
+from platformio.managers.platform import PlatformManager
+
+
+module_logger = logging.getLogger(__name__)
+
+
+
+# Do not add or remove any information from the message and simply pass it "as-is"
+special_formatters = {
+ 'subprocess': logging.Formatter('%(message)s')
+}
+
+default_log_record_factory = logging.getLogRecordFactory()
+
+def log_record_factory(*log_record_args, **log_record_kwargs):
+ """
+ Replace the default factory of logging.LogRecord's instances so we can handle our special logging flags
+ """
+ args_idx = 5 # index of 'args' argument in the positional arguments list
+
+ if 'from_subprocess' in log_record_args[args_idx]:
+ # Remove our custom flag from the tuple (it is inside a tuple that is inside a list)
+ new_log_record_args = log_record_args[:args_idx] + \
+ (tuple(arg for arg in log_record_args[args_idx] if arg != 'from_subprocess'),) + \
+ log_record_args[args_idx + 1:]
+ # Construct an ordinary LogRecord and append our flag as an attribute
+ record = default_log_record_factory(*new_log_record_args, **log_record_kwargs)
+ record.from_subprocess = True
+ else:
+ record = default_log_record_factory(*log_record_args, **log_record_kwargs)
+
+ return record
+
+logging.setLogRecordFactory(log_record_factory)
+
+
+class DispatchingFormatter(logging.Formatter):
+ """
+ The wrapper around the ordinary logging.Formatter allowing to have multiple formatters for different purposes.
+ 'extra' argument of the log() function has a similar intention but different mechanics
+ """
+
+ def __init__(self, *args, special: dict = None, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ # Store all provided formatters in an internal variable
+ if isinstance(special, dict) and all(isinstance(value, logging.Formatter) for value in special.values()):
+ self._formatters = special
+ else:
+ module_logger.warning(f"'special' argument is for providing custom formatters for special logging events "
+ "and should be a dictionary with logging.Formatter values")
+ self._formatters = {}
+
+ self.warn_was_shown = False
+
+ def format(self, record: logging.LogRecord) -> str:
+ """
+ Use suitable formatter based on the LogRecord attributes
+ """
+ if hasattr(record, 'from_subprocess') and record.from_subprocess:
+ if 'subprocess' in self._formatters:
+ return self._formatters['subprocess'].format(record)
+ elif not self.warn_was_shown:
+ module_logger.warning("No formatter found for the 'subprocess' case, use default hereinafter")
+ return super().format(record)
+
+
+class LogPipe(threading.Thread):
+ """
+ The thread combined with a context manager to provide a nice way to temporarily redirect something's stream output
+ into logging module. The most straightforward application is to suppress subprocess STDOUT and/or STDERR streams and
+ wrap them in the logging mechanism as it is for now for any other message in your app.
+ """
+
+ def __init__(self, logger: logging.Logger, level: int, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ self.logger = logger
+ self.level = level
+
+ self.fd_read, self.fd_write = os.pipe() # create 2 ends of the pipe and setup the reading one
+ self.pipe_reader = os.fdopen(self.fd_read)
+
+ def __enter__(self) -> int:
+ """
+ Activate the thread and return the consuming end of the pipe so the invoking code can use it to feed its
+ messages from now on
+ """
+ self.start()
+ return self.fd_write
+
+ def run(self):
+ """
+ Routine of the thread, logging everything
+ """
+ for line in iter(self.pipe_reader.readline, ''):
+ self.logger.log(self.level, line.strip('\n'), 'from_subprocess') # mark the message origin
+
+ self.pipe_reader.close()
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ """
+ The exception will be passed forward, if present, so we don't need to do something with that. The following
+ tear-down process will be done anyway
+ """
+ os.close(self.fd_write)
+
+
+
+def get_platformio_boards() -> List[str]:
+ """
+ Use PlatformIO Python sources to obtain the boards list. As we interested only in STM32 ones, cut off all the
+ others.
+
+ IMPORTANT NOTE: The inner implementation can go to the Internet from time to time when it decides that its cache is
+ out of date. So it can take a long time to execute.
+ """
+
+ pm = PlatformManager()
+ return [board['id'] for board in pm.get_all_boards() if 'stm32cube' in board['frameworks']]