Sourcery refactored devel branch

src/fastentrypoints.py

@@ -56,7 +56,7 @@
 
 
 @classmethod
-def get_args(cls, dist, header=None):  # noqa: D205,D400
+def get_args(cls, dist, header=None):    # noqa: D205,D400
     """
     Yield write_script() argument tuples for a distribution's
     console_scripts and gui_scripts entry points.
@@ -74,9 +74,7 @@ def get_args(cls, dist, header=None):  # noqa: D205,D400
             script_text = TEMPLATE.format(
                 ep.module_name, ep.attrs[0], ".".join(ep.attrs), spec, group, name
             )
-            # pylint: disable=E1101
-            args = cls._get_script_args(type_, name, header, script_text)
-            yield from args
+            yield from cls._get_script_args(type_, name, header, script_text)
 
 
 # pylint: disable=E1101

src/fprime_gds/common/data_types/ch_data.py

@@ -63,13 +63,15 @@ def _compute_display_text(self, val_obj, template):
         # in which case we just display the description
         if val_obj is None:
             return template.ch_desc
-        temp_val = val_obj.val if not isinstance(val_obj, (SerializableType, ArrayType)) else val_obj.formatted_val
+        temp_val = (
+            val_obj.formatted_val
+            if isinstance(val_obj, (SerializableType, ArrayType))
+            else val_obj.val
+        )
         fmt_str = template.get_format_str()
         if temp_val is None:
             return ""
-        if fmt_str:
-            return format_string_template(fmt_str, (temp_val,))
-        return temp_val
+        return format_string_template(fmt_str, (temp_val,)) if fmt_str else temp_val
 
     def set_pkt(self, pkt):
         """
@@ -166,13 +168,13 @@ def get_str(self, time_zone=None, verbose=False, csv=False):
         ch_name = self.template.get_full_name()
         display_text = self.get_display_text()
 
-        if verbose and csv:
-            return f"{time_str_nice},{raw_time_str},{ch_name},{self.id},{display_text}"
-        if verbose and not csv:
+        if verbose:
+            if csv:
+                return f"{time_str_nice},{raw_time_str},{ch_name},{self.id},{display_text}"
             return (
                 f"{time_str_nice}: {ch_name} ({self.id}) {raw_time_str} {display_text}"
             )
-        if not verbose and csv:
+        if csv:
             return f"{time_str_nice},{ch_name},{display_text}"
         return f"{time_str_nice}: {ch_name} = {display_text}"
 

src/fprime_gds/common/data_types/cmd_data.py

@@ -133,11 +133,11 @@ def get_str(self, time_zone=None, verbose=False, csv=False):
 
             arg_str = " ".join(str(arg_val_list))
 
-        if verbose and csv:
-            return f"{time_str},{raw_time_str},{name},{self.id},{arg_str}"
-        if verbose and not csv:
+        if verbose:
+            if csv:
+                return f"{time_str},{raw_time_str},{name},{self.id},{arg_str}"
             return f"{time_str}: {name} ({self.id}) {raw_time_str} : {arg_str}"
-        if not verbose and csv:
+        if csv:
             return f"{time_str},{name},{arg_str}"
         return f"{time_str}: {name} : {arg_str}"
 

src/fprime_gds/common/data_types/event_data.py

@@ -39,8 +39,12 @@ def __init__(self, event_args, event_time, event_temp):
         self.args = event_args
         self.time = event_time
         self.template = event_temp
-        self.display_text = event_temp.description if event_args is None else format_string_template(
-            event_temp.format_str, tuple([arg.val for arg in event_args])
+        self.display_text = (
+            event_temp.description
+            if event_args is None
+            else format_string_template(
+                event_temp.format_str, tuple(arg.val for arg in event_args)
+            )
         )
 
     def get_args(self):
@@ -105,13 +109,13 @@ def get_str(self, time_zone=None, verbose=False, csv=False):
         severity = self.template.get_severity()
         display_text = self.display_text
 
-        if verbose and csv:
-            return (
-                f"{time_str},{raw_time_str},{name},{self.id},{severity},{display_text}"
-            )
-        if verbose and not csv:
+        if verbose:
+            if csv:
+                return (
+                    f"{time_str},{raw_time_str},{name},{self.id},{severity},{display_text}"
+                )
             return f"{time_str}: {name} ({self.id}) {raw_time_str} {severity} : {display_text}"
-        if not verbose and csv:
+        if csv:
             return f"{time_str},{name},{severity},{display_text}"
         return f"{time_str}: {name} {severity} : {display_text}"
 

src/fprime_gds/common/data_types/pkt_data.py

@@ -87,7 +87,7 @@ def get_str(self, time_zone=None, verbose=False, csv=False):
                 self.template.get_id(),
                 str(self.time),
             )
-        elif not csv and not verbose:
+        elif not csv:
             pkt_str += f"{self.time.to_readable(time_zone)}: {self.template.get_name()} {{\n"
 
         for i, ch in enumerate(self.chs):

src/fprime_gds/common/decoders/file_decoder.py

@@ -57,8 +57,4 @@ def decode_api(self, data):
         if packetType == "END":  # Packet Type is END
             hashValue = struct.unpack_from(">I", data, 5)
             return [file_data.EndPacketData(seqID, hashValue),]
-        if packetType == "CANCEL":  # Packet Type is CANCEL
-            # CANCEL Packets have no data
-            return [file_data.CancelPacketData(seqID),]
-        # The data was not determined to be any of the packet types so return none
-        return None
+        return [file_data.CancelPacketData(seqID),] if packetType == "CANCEL" else None

src/fprime_gds/common/encoders/ch_encoder.py

@@ -86,6 +86,4 @@ def encode_api(self, data):
         self.len_obj.val = len_val
         len_bin = self.len_obj.serialize()
 
-        binary_data = len_bin + desc_bin + id_bin + time_bin + val_bin
-
-        return binary_data
+        return len_bin + desc_bin + id_bin + time_bin + val_bin

src/fprime_gds/common/encoders/cmd_encoder.py

@@ -102,6 +102,4 @@ def encode_api(self, data):
         self.len_obj.val = length_val
         length = self.len_obj.serialize()
 
-        binary_data = desc + length + descriptor + op_code + arg_data
-
-        return binary_data
+        return desc + length + descriptor + op_code + arg_data

src/fprime_gds/common/encoders/event_encoder.py

@@ -92,6 +92,4 @@ def encode_api(self, data):
         self.len_obj.val = len_val
         len_bin = self.len_obj.serialize()
 
-        binary_data = len_bin + desc_bin + id_bin + time_bin + arg_bin
-
-        return binary_data
+        return len_bin + desc_bin + id_bin + time_bin + arg_bin

src/fprime_gds/common/encoders/pkt_encoder.py

@@ -92,6 +92,4 @@ def encode_api(self, data):
         self.len_obj.val = len_val
         len_bin = self.len_obj.serialize()
 
-        binary_data = len_bin + desc_bin + id_bin + time_bin + ch_bin
-
-        return binary_data
+        return len_bin + desc_bin + id_bin + time_bin + ch_bin

src/fprime_gds/common/files/downlinker.py

@@ -97,8 +97,7 @@ def handle_start(self, data):
         )
         self.active.open(TransmitFileState.WRITE)
         LOGGER.addHandler(self.active.log_handler)
-        message = "Received START packet with metadata:\n"
-        message += "\tSize: %d\n"
+        message = "Received START packet with metadata:\n" + "\tSize: %d\n"
         message += "\tSource: %s\n"
         message += "\tDestination: %s"
         LOGGER.info(message, size, source_path, dest_path)

src/fprime_gds/common/files/uplinker.py

@@ -87,9 +87,7 @@ def remove(self, source):
         try:
             first = None
             found = self.queue.get_nowait()
-            while found != first:
-                if found.source == source:
-                    break
+            while found != first and found.source != source:
                 if first is None:
                     first = found
             self.queue.put_nowait(found)

src/fprime_gds/common/gds_cli/base_commands.py

@@ -232,9 +232,7 @@ def _get_item_string(
         """
         if not item:
             return ""
-        if as_json:
-            return json.dumps(item.get_dict())
-        return item.get_str(verbose=True)
+        return json.dumps(item.get_dict()) if as_json else item.get_str(verbose=True)
 
     @classmethod
     def _execute_command(cls, args, api: IntegrationTestAPI):

src/fprime_gds/common/gds_cli/channels.py

@@ -38,12 +38,11 @@ def _get_item_list(
             channels
         :return: List of channels that passed the filter
         """
-        channel_list = test_api_utils.get_item_list(
+        return test_api_utils.get_item_list(
             item_dictionary=project_dictionary.channel_id,
             search_filter=filter_predicate,
             template_to_data=ChData.get_empty_obj,
         )
-        return channel_list
 
     @classmethod
     def _get_upcoming_item(

src/fprime_gds/common/gds_cli/command_send.py

@@ -37,8 +37,7 @@ def _get_closest_commands(
         :return: A list of the closest matching commands (potentially empty)
         """
         known_commands = project_dictionary.command_name.keys()
-        closest_matches = difflib.get_close_matches(command_name, known_commands, n=num)
-        return closest_matches
+        return difflib.get_close_matches(command_name, known_commands, n=num)
 
     @staticmethod
     def _get_command_template(
@@ -130,8 +129,7 @@ def _get_item_string(
             len(item.get_args()),
         )
 
-        cmd_description = item.get_description()
-        if cmd_description:
+        if cmd_description := item.get_description():
             cmd_string += f"Description: {(cmd_description)}\n" 
 
         for arg in item.get_args():
@@ -159,10 +157,9 @@ def _execute_command(cls, args, api: IntegrationTestAPI):
             api.send_command(command, arguments)
         except KeyError:
             cls._log(f"{command} is not a known command")
-            close_matches = cls._get_closest_commands(
+            if close_matches := cls._get_closest_commands(
                 api.pipeline.dictionaries, command
-            )
-            if close_matches:
+            ):
                 cls._log(f"Similar known commands: {close_matches}")
         except NotInitializedException:
             temp = cls._get_command_template(api.pipeline.dictionaries, command)

src/fprime_gds/common/gds_cli/events.py

@@ -37,12 +37,11 @@ def _get_item_list(
         :param filter_predicate: Test API predicate used to filter shown events
         :return: List of EventData items that passed the filter
         """
-        event_list = test_api_utils.get_item_list(
+        return test_api_utils.get_item_list(
             item_dictionary=project_dictionary.event_id,
             search_filter=filter_predicate,
             template_to_data=EventData.get_empty_obj,
         )
-        return event_list
 
     @classmethod
     def _get_upcoming_item(

src/fprime_gds/common/gds_cli/filtering_utils.py

@@ -109,7 +109,7 @@ def __init__(self, search_string: str, to_string_func: Callable[[Any], str] = st
         :param to_string_func: An optional method for converting the given
             object to a string
         """
-        self.search_string = str(search_string)
+        self.search_string = search_string
         self.to_str = to_string_func
 
     def __call__(self, item):

src/fprime_gds/common/gds_cli/test_api_utils.py

@@ -123,8 +123,7 @@ def repeat_until_interrupt(func: Callable, *args):
     """
     try:
         while True:
-            new_args = func(*args)  # lgtm [py/call/wrong-arguments]
-            if new_args:
+            if new_args := func(*args):
                 args = new_args
     except KeyboardInterrupt:
         pass

src/fprime_gds/common/models/common/channel_telemetry.py

@@ -169,6 +169,5 @@ def changed(self):
 
     @changed.setter
     def changed(self, ch):
-        if not ch == False or not ch == True:
-            ch = True
+        ch = True
         self.__changed = ch

src/fprime_gds/common/models/common/event.py

@@ -92,8 +92,7 @@ def deserialize(self, ser_data):
                 vals.append("ERR")
             offset = offset + arg_obj.getSize()
 
-        vals = [0] + vals
-        return vals
+        return [0] + vals
 
     def stringify(self, event_args_list):
         """

src/fprime_gds/common/parsers/seq_file_parser.py

@@ -99,7 +99,7 @@ def parseArg(arg):
                         # Otherwise it is an enum type:
                         return str(arg)
 
-            return [item for item in map(parseArg, args)]
+            return list(map(parseArg, args))
 
         def parseTime(lineNumber, time):
             """

src/fprime_gds/common/templates/event_template.py

@@ -126,5 +126,3 @@ def get_args(self):
         return self.args
 
 
-if __name__ == "__main__":
-    pass

src/fprime_gds/common/testing_fw/api.py

@@ -1264,8 +1264,6 @@ def __assert_pred(self, name, predicate, value, msg="", expect=False):
         if predicate(value):
             ast_msg = f"{name} succeeded: {msg}\nassert {pred_msg}"
             self.__log(ast_msg, TestLogger.GREEN)
-            if not expect:
-                assert True, pred_msg
             return True
         ast_msg = f"{name} failed: {msg}\nassert {pred_msg}"
         if not expect:

src/fprime_gds/common/tools/seqgen.py

@@ -48,25 +48,24 @@ def generateSequence(inputFile, outputFile, dictionary, timebase, cont=False):
 
     # Check for files
     if not os.path.isfile(inputFile):
-        raise SeqGenException("Can't open file '" + inputFile + "'. ")
+        raise SeqGenException(f"Can't open file '{inputFile}'. ")
 
     if not os.path.isfile(dictionary):
-        raise SeqGenException("Can't open file '" + dictionary + "'. ")
+        raise SeqGenException(f"Can't open file '{dictionary}'. ")
 
     # Check the user environment:
     cmd_xml_dict = CmdXmlLoader()
     try:
         (cmd_id_dict, cmd_name_dict, versions) = cmd_xml_dict.construct_dicts(dictionary)
     except gseExceptions.GseControllerUndefinedFileException:
-        raise SeqGenException("Can't open file '" + dictionary + "'. ")
+        raise SeqGenException(f"Can't open file '{dictionary}'. ")
 
-    # Parse the input file:
-    command_list = []
     file_parser = SeqFileParser()
 
     parsed_seq = file_parser.parse(inputFile, cont=cont)
 
     messages = []
+    command_list = []
     try:
         for i, descriptor, seconds, useconds, mnemonic, args in parsed_seq:
             try:
@@ -75,9 +74,7 @@ def generateSequence(inputFile, outputFile, dictionary, timebase, cont=False):
                         "Line %d: %s"
                         % (
                             i + 1,
-                            "'"
-                            + mnemonic
-                            + "' does not match any command in the command dictionary.",
+                            f"'{mnemonic}' does not match any command in the command dictionary.",
                         )
                     )
                 # Set the command arguments:
@@ -104,7 +101,7 @@ def generateSequence(inputFile, outputFile, dictionary, timebase, cont=False):
         writer.open(outputFile)
     except:
         raise SeqGenException(
-            "Encountered problem opening output file '" + outputFile + "'."
+            f"Encountered problem opening output file '{outputFile}'."
         )
 
     writer.write(command_list)