From 063cea1fe94a4dc52213df1d1b79e8c76dc8e20a Mon Sep 17 00:00:00 2001
From: "Robin.Mueller" <robin.mueller.m@gmail.com>
Date: Sun, 26 Apr 2020 17:18:37 +0200
Subject: [PATCH] dummy com if almost working.
unit test commented out for now.
sequential mode untested
---
comIF/obsw_dummy_com_if.py | 18 +++-
comIF/obsw_serial_com_if.py | 5 +-
sendreceive/obsw_command_sender_receiver.py | 5 +-
sendreceive/obsw_tm_listener.py | 18 ++--
tc/obsw_pus_tc_base.py | 10 +-
tm/obsw_pus_tm_base.py | 105 +++++++++++---------
tm/obsw_pus_tm_creator.py | 85 ++++++++++++++++
tm/obsw_pus_tm_factory.py | 8 +-
tm/obsw_tm_service_1.py | 18 +++-
utility/obsw_logger.py | 19 +++-
10 files changed, 219 insertions(+), 72 deletions(-)
create mode 100644 tm/obsw_pus_tm_creator.py
diff --git a/comIF/obsw_dummy_com_if.py b/comIF/obsw_dummy_com_if.py
index 995e6f4..ddc36f8 100644
--- a/comIF/obsw_dummy_com_if.py
+++ b/comIF/obsw_dummy_com_if.py
@@ -9,25 +9,39 @@ from typing import Tuple
from comIF.obsw_com_interface import CommunicationInterface
from tc.obsw_pus_tc_base import PusTelecommand, PusTcInfoT, TcDictionaryKeys
-
+from tm.obsw_pus_tm_factory import PusTelemetryFactory
+from tm.obsw_pus_tm_creator import PusTelemetryCreator
+from tm.obsw_tm_service_1 import Service1TmPacked
class DummyComIF(CommunicationInterface):
def __init__(self, tmtc_printer):
super().__init__(tmtc_printer)
self.service_sent = 0
+ self.reply_pending = False
def close(self) -> None:
pass
def data_available(self, parameters):
+ if self.reply_pending:
+ return True
return False
def poll_interface(self, parameters: any = 0) -> Tuple[bool, list]:
pass
def receive_telemetry(self, parameters: any = 0):
- pass
+ if self.service_sent == 17 and self.reply_pending:
+ tm_packer = Service1TmPacked(subservice=1, ssc=0)
+ tm_packet_raw = tm_packer.pack()
+ tm_packet = PusTelemetryFactory.create(tm_packet_raw)
+ self.reply_pending = False
+ return [tm_packet]
+ return []
+
+
def send_telecommand(self, tc_packet: PusTelecommand, tc_packet_info: PusTcInfoT = None) -> None:
if isinstance(tc_packet_info, dict) and tc_packet_info.__len__() > 0:
self.service_sent = tc_packet_info[TcDictionaryKeys.SERVICE]
+ self.reply_pending = True
diff --git a/comIF/obsw_serial_com_if.py b/comIF/obsw_serial_com_if.py
index 892a343..c718b5d 100644
--- a/comIF/obsw_serial_com_if.py
+++ b/comIF/obsw_serial_com_if.py
@@ -56,13 +56,14 @@ class SerialComIF(CommunicationInterface):
return packet_list
return []
- # TODO: DO NOT PRINT HERE! this will block the listener thread!!!
+ # TODO: Serialization is performed here, but I suspect this might slow down the
+ # listener thread.. but at least no printing is done here.
def poll_interface(self, parameters: any = 0) -> Tuple[bool, PusTmListT]:
if self.data_available():
pus_data_list, number_of_packets = self.__poll_pus_packets()
packet_list = []
for counter in range(0, number_of_packets):
- packet = PusTelemetryFactory.create(pus_data_list[counter])
+ packet = PusTelemetryFactory.create(bytearray(pus_data_list[counter]))
packet_list.append(packet)
return True, packet_list
return False, []
diff --git a/sendreceive/obsw_command_sender_receiver.py b/sendreceive/obsw_command_sender_receiver.py
index 07cdb1e..d800fd7 100644
--- a/sendreceive/obsw_command_sender_receiver.py
+++ b/sendreceive/obsw_command_sender_receiver.py
@@ -21,6 +21,7 @@ from sendreceive.obsw_tm_listener import TmListener
from tc.obsw_pus_tc_base import TcQueueEntryT
from tm.obsw_pus_tm_factory import PusTmQueueT
from utility.obsw_logger import get_logger
+from tm.obsw_pus_tm_factory import PusTelemetryFactory
logger = get_logger()
@@ -150,4 +151,6 @@ class CommandSenderReceiver:
def print_tm_queue(self, tm_queue: PusTmQueueT):
for tm_packet in tm_queue:
- self._tmtc_printer.print_telemetry(tm_packet)
+ logger.debug(str(tm_packet))
+ tm_info = tm_packet.pack_tm_information()
+ self._tmtc_printer.print_telemetry(tm_packet, tm_info)
diff --git a/sendreceive/obsw_tm_listener.py b/sendreceive/obsw_tm_listener.py
index 1be1a71..f017770 100644
--- a/sendreceive/obsw_tm_listener.py
+++ b/sendreceive/obsw_tm_listener.py
@@ -14,12 +14,12 @@ import time
import threading
from collections import deque
from typing import TypeVar
-
-from comIF.obsw_com_interface import ComIfT
+from utility.obsw_logger import get_logger
+from comIF.obsw_com_interface import CommunicationInterface
from tm.obsw_pus_tm_factory import PusTmQueueT, PusTmInfoQueueT
import config.obsw_config as g
-
+logger = get_logger()
TmListenerT = TypeVar('TmListenerT', bound='TmListener')
@@ -30,7 +30,8 @@ class TmListener:
and changing the mode to do special mode operations. The mode operation ends as soon
the modeOpFinished Event is set() !
"""
- def __init__(self, com_interface: ComIfT, tm_timeout: float, tc_timeout_factor: float):
+ def __init__(self, com_interface: CommunicationInterface, tm_timeout: float,
+ tc_timeout_factor: float):
self.tm_timeout = tm_timeout
self.tc_timeout_factor = tc_timeout_factor
self.com_interface = com_interface
@@ -79,7 +80,7 @@ class TmListener:
while not self.mode_op_finished.is_set():
self.perform_mode_operation()
self.mode_op_finished.clear()
- print("Transitioning to listener mode.")
+ logger.info("Transitioning to listener mode.")
self.mode_id = g.ModeList.ListenerMode
def perform_mode_operation(self):
@@ -102,10 +103,11 @@ class TmListener:
elif self.mode_id == g.ModeList.ServiceTestMode or \
self.mode_id == g.ModeList.SoftwareTestMode:
if self.check_for_one_telemetry_sequence():
- print("TM Listener: Reply sequence received!")
+ logger.info("TmListener: Reply sequence received!")
self.replyEvent.set()
elif self.mode_id == g.ModeList.UnitTest:
- self.__tm_info_queue = self.com_interface.receive_telemetry_and_store_info(self.__tm_info_queue)
+ # TODO: needs to be reworked.
+ # self.__tm_info_queue = self.com_interface.receive_telemetry_and_store_info(self.__tm_info_queue)
self.replyEvent.set()
def check_for_one_telemetry_sequence(self) -> bool:
@@ -132,7 +134,7 @@ class TmListener:
self.tm_timeout = g.G_TM_TIMEOUT
return True
else:
- print("TM Listener: Configuration error in communication interface!")
+ logger.error("TmListener: Configuration error in communication interface!")
sys.exit()
def retrieve_tm_packet_queue(self):
diff --git a/tc/obsw_pus_tc_base.py b/tc/obsw_pus_tc_base.py
index d8186cf..b4678e5 100644
--- a/tc/obsw_pus_tc_base.py
+++ b/tc/obsw_pus_tc_base.py
@@ -4,6 +4,8 @@ import logging
from enum import Enum
from typing import Dict, Union, Tuple, Deque
+from utility.obsw_logger import get_logger
+logger = get_logger()
class TcDictionaryKeys(Enum):
SERVICE = 1
@@ -21,7 +23,6 @@ TcQueueT = Deque[TcQueueEntryT]
PusTcInfoQueueT = Deque[PusTcInfoT]
-
class PusTelecommand:
HEADER_SIZE = 6
"""
@@ -29,7 +30,7 @@ class PusTelecommand:
input parameters. The structure of a PUS telecommand is specified in ECSS-E-70-41A on p.42
and is also shown below (bottom)
"""
- def __init__(self, service: int, subservice: int, ssc=0, app_data=bytearray([]),
+ def __init__(self, service: int, subservice: int, ssc=0, app_data: bytearray=bytearray([]),
source_id: int = 0, version: int = 0, apid: int = 0x73):
"""
Initiates a telecommand with the given parameters.
@@ -73,9 +74,8 @@ class PusTelecommand:
data_length = 4 + len(self.app_data) + 1
return data_length
except TypeError:
- print("OBSW_TcPacket: Invalid service_type of data")
- logging.exception("Error")
- sys.exit()
+ logger.error("PusTelecommand: Invalid type of application data!")
+ return 0
def get_total_length(self):
"""
diff --git a/tm/obsw_pus_tm_base.py b/tm/obsw_pus_tm_base.py
index a35eb68..6916758 100644
--- a/tm/obsw_pus_tm_base.py
+++ b/tm/obsw_pus_tm_base.py
@@ -1,8 +1,14 @@
+import math
+import time
+
+from utility.obsw_logger import get_logger
+
import datetime
from enum import Enum, auto
from typing import Final, Dict
from crcmod import crcmod
+logger = get_logger()
class TmDictionaryKeys(Enum):
SERVICE = auto()
@@ -34,6 +40,7 @@ class PusTelemetry:
It automatically deserializes the packet, exposing various packet fields via getter functions.
PUS Telemetry structure according to ECSS-E-70-41A p.46. Also see structure below (bottom).
"""
+ PUS_HEADER_SIZE = 6
def __init__(self, byte_array: bytearray = bytearray()):
if byte_array == bytearray():
return
@@ -130,7 +137,7 @@ class PusTelemetry:
:return: Size of the TM wiretapping_packet
"""
# PusHeader Size + _tm_data size
- size = PusPacketHeader.PUS_HEADER_SIZE + self._pus_header.length + 1
+ size = PusTelemetry.PUS_HEADER_SIZE + self._pus_header.length + 1
return size
def get_ssc(self) -> int:
@@ -161,46 +168,15 @@ class PusTelemetry:
print(self.return_data_string())
-class PusTelemetryPacked(PusTelemetry):
- """
- Alternative way to create a PUS Telemetry packet by specifying telemetry parameters,
- similarly to the way telecommands are created. This can be used to create telemetry
- directly in the software.
- """
- def __init__(self, service: int, subservice: int, ssc: int = 0, apid: int = 0x73, version: int = 0,
- packet_type: int = 0, data_field_header_flag: int = 1):
- super().__init__()
- self.packet_id = [0x0, 0x0]
- self.packet_id[0] = ((version << 5) & 0xE0) | ((packet_type & 0x01) << 4) | \
- ((data_field_header_flag & 0x01) << 3) | ((apid & 0x700) >> 8)
- self.ssc = ssc
- self.psc = (ssc & 0x3FFF) | (0b11 << 16)
- self.pus_version_and_ack_byte = 0b00011111
-
- # NOTE: In PUS-C, the PUS Version is 2 and specified for the first 4 bits.
- # The other 4 bits of the first byte are the spacecraft time reference status
- # To change to PUS-C, set 0b00100000
- self.data_field_version = 0b00010000
- self.service = service
- self.subservice = subservice
- self.pack_subcounter = 0
- self.time = 0
-
- def pack(self) -> bytearray:
-
-
-
-
# pylint: disable=too-many-instance-attributes
class PusPacketHeader:
"""
This class unnpacks the PUS packet header, also see PUS structure below or PUS documentation.
"""
- PUS_HEADER_SIZE = 6
def __init__(self, pus_packet_raw: bytes):
data_to_check = pus_packet_raw
- if len(pus_packet_raw) < PusPacketHeader.PUS_HEADER_SIZE:
+ if len(pus_packet_raw) < PusTelemetry.PUS_HEADER_SIZE:
self.version = 0
self.type = 0
self.data_field_header_flag = 0
@@ -219,15 +195,15 @@ class PusPacketHeader:
self.length = pus_packet_raw[4] << 8 | pus_packet_raw[5]
self.valid = False
crc_func = crcmod.mkCrcFun(0x11021, rev=False, initCrc=0xFFFF, xorOut=0x0000)
- if len(data_to_check) < ((self.length + 1) + PusPacketHeader.PUS_HEADER_SIZE):
- print("Invalid wiretapping_packet length")
+ if len(data_to_check) < ((self.length + 1) + PusTelemetry.PUS_HEADER_SIZE):
+ logger.warning("PusPacketHeader: Invalid packet length")
return
- data_to_check = data_to_check[0:(self.length + 1) + PusPacketHeader.PUS_HEADER_SIZE]
+ data_to_check = data_to_check[0:(self.length + 1) + PusTelemetry.PUS_HEADER_SIZE]
crc = crc_func(data_to_check)
if crc == 0:
self.valid = True
else:
- print("Invalid CRC detected !")
+ logger.warning("PusPacketHeader: Invalid CRC detected !")
def append_pus_packet_header(self, array):
array.append(str(chr(self.apid)))
@@ -280,20 +256,53 @@ class PusPacketDataFieldHeader:
class PusTelemetryTimestamp:
"""
- Unpacks the time datafield of the TM packet.
+ Unpacks the time datafield of the TM packet. Right now, CDS Short timeformat is used,
+ and the size of the time stamp is expected to be seven bytes.
"""
- def __init__(self, byte_array):
- # pField = byte_array[0]
- byte_array = byte_array[1:]
- self.days = ((byte_array[0] << 8) | (byte_array[1])) - 4383
- self.seconds = self.days * (24 * 60 * 60)
- s_day = ((byte_array[2] << 24) | (byte_array[3] << 16) |
- (byte_array[4]) << 8 | byte_array[5]) / 1000
- self.seconds += s_day
- self.time = self.seconds
- self.datetime = str(datetime.datetime.
+ CDS_ID = 4
+ SECONDS_PER_DAY = 86400
+ EPOCH = datetime.datetime.utcfromtimestamp(0)
+ DAYS_CCSDS_TO_UNIX = 4383
+ def __init__(self, byte_array: bytearray=bytearray([])):
+ if len(byte_array) > 0:
+ # pField = byte_array[0]
+ self.days = ((byte_array[1] << 8) | (byte_array[2])) - \
+ PusTelemetryTimestamp.DAYS_CCSDS_TO_UNIX
+ self.seconds = self.days * (24 * 60 * 60)
+ s_day = ((byte_array[3] << 24) | (byte_array[4] << 16) |
+ (byte_array[5]) << 8 | byte_array[6]) / 1000
+ self.seconds += s_day
+ self.time = self.seconds
+ self.datetime = str(datetime.datetime.
utcfromtimestamp(self.time).strftime("%Y-%m-%d %H:%M:%S.%f"))
+ def pack_current_time(self) -> bytearray:
+ """
+ Returns a seven byte CDS short timestamp
+ """
+ timestamp = bytearray()
+ p_field = (PusTelemetryTimestamp.CDS_ID << 4) + 0
+ days = (datetime.datetime.utcnow() - PusTelemetryTimestamp.EPOCH).days + \
+ PusTelemetryTimestamp.DAYS_CCSDS_TO_UNIX
+ days_h = (days & 0xFF00) >> 8
+ days_l = days & 0xFF
+ seconds = time.time()
+ fraction_ms = seconds - math.floor(seconds)
+ days_ms = int((seconds % PusTelemetryTimestamp.SECONDS_PER_DAY) * 1000 + fraction_ms)
+ days_ms_hh = (days_ms & 0xFF000000) >> 24
+ days_ms_h = (days_ms & 0xFF0000) >> 16
+ days_ms_l = (days_ms & 0xFF00) >> 8
+ days_ms_ll = (days_ms & 0xFF)
+ timestamp.append(p_field)
+ timestamp.append(days_h)
+ timestamp.append(days_l)
+ timestamp.append(days_ms_hh)
+ timestamp.append(days_ms_h)
+ timestamp.append(days_ms_l)
+ timestamp.append(days_ms_ll)
+ return timestamp
+
+
def print_time(self, array):
array.append(self.time)
array.append(self.datetime)
diff --git a/tm/obsw_pus_tm_creator.py b/tm/obsw_pus_tm_creator.py
new file mode 100644
index 0000000..fad4301
--- /dev/null
+++ b/tm/obsw_pus_tm_creator.py
@@ -0,0 +1,85 @@
+from tm.obsw_pus_tm_base import PusTelemetry, PusTelemetryTimestamp
+from utility.obsw_logger import get_logger
+import crcmod
+
+
+logger = get_logger()
+
+
+class PusTelemetryCreator(PusTelemetry):
+ """
+ Alternative way to create a PUS Telemetry packet by specifying telemetry parameters,
+ similarly to the way telecommands are created. This can be used to create telemetry
+ directly in the software.
+ """
+ def __init__(self, service: int, subservice: int, ssc: int = 0,
+ source_data: bytearray=bytearray([]), apid: int = 0x73, version: int = 0):
+ super().__init__()
+ # packet type for telemetry is 0 as specified in standard
+ packet_type = 0
+ # specified in standard
+ data_field_header_flag = 1
+ self.packet_id = [0x0, 0x0]
+ self.packet_id[0] = ((version << 5) & 0xE0) | ((packet_type & 0x01) << 4) | \
+ ((data_field_header_flag & 0x01) << 3) | ((apid & 0x700) >> 8)
+ self.packet_id[1] = apid & 0xFF
+ self.ssc = ssc
+ self.psc = (ssc & 0x3FFF) | (0b11 << 16)
+ self.pus_version_and_ack_byte = 0b00011111
+
+ # NOTE: In PUS-C, the PUS Version is 2 and specified for the first 4 bits.
+ # The other 4 bits of the first byte are the spacecraft time reference status
+ # To change to PUS-C, set 0b00100000
+ self.data_field_version = 0b00010000
+ self.service = service
+ self.subservice = subservice
+ self.pack_subcounter = 0
+ # it is assumed the time field consts of 8 bytes.
+
+ self.source_data = source_data
+
+ def pack(self) -> bytearray:
+ tm_packet_raw = bytearray()
+ # PUS Header
+ tm_packet_raw.extend(self.packet_id)
+ tm_packet_raw.append((self.psc & 0xFF00) >> 8)
+ tm_packet_raw.append(self.psc & 0xFF)
+ source_length = self.get_source_data_length()
+ tm_packet_raw.append((source_length & 0xFF00) >> 8)
+ tm_packet_raw.append(source_length & 0xFF)
+ logger.debug(len(tm_packet_raw))
+ # PUS Source Data Field
+ tm_packet_raw.append(self.data_field_version)
+ tm_packet_raw.append(self.service)
+ tm_packet_raw.append(self.subservice)
+ tm_packet_raw.append(self.pack_subcounter)
+ logger.debug(len(tm_packet_raw))
+ timestamper = PusTelemetryTimestamp()
+ tm_packet_raw.extend(timestamper.pack_current_time())
+ logger.debug(len(tm_packet_raw))
+
+ # Source Data
+ tm_packet_raw.extend(self.source_data)
+ logger.debug(len(tm_packet_raw))
+
+ # CRC16 checksum
+ crc_func = crcmod.mkCrcFun(0x11021, rev=False, initCrc=0xFFFF, xorOut=0x0000)
+ crc16 = crc_func(tm_packet_raw)
+ tm_packet_raw.append((crc16 & 0xFF00) >> 8)
+ tm_packet_raw.append(crc16 & 0xFF)
+ logger.debug(len(tm_packet_raw))
+ return tm_packet_raw
+
+ def get_source_data_length(self) -> int:
+ """
+ Retrieve size of TC packet in bytes.
+ Formula according to PUS Standard: C = (Number of octets in packet data field) - 1.
+ The size of the TC packet is the size of the packet secondary header with
+ source ID + the length of the application data + length of the CRC16 checksum - 1
+ """
+ try:
+ data_length = 4 + len(self.source_data) + 1
+ return data_length
+ except TypeError:
+ logger.error("PusTelecommand: Invalid type of application data!")
+ return 0
diff --git a/tm/obsw_pus_tm_factory.py b/tm/obsw_pus_tm_factory.py
index 547b2a4..5e99f6a 100644
--- a/tm/obsw_pus_tm_factory.py
+++ b/tm/obsw_pus_tm_factory.py
@@ -1,12 +1,15 @@
# -*- coding: utf-8 -*-
-from typing import Deque, List, Dict, Tuple
+from typing import Deque, List, Tuple
from tm.obsw_pus_tm_base import PusTelemetry, PusTmInfoT
from tm.obsw_tm_service_1 import Service1TM
from tm.obsw_tm_service_3 import Service3TM
from tm.obsw_tm_service_5 import Service5TM
+from utility.obsw_logger import get_logger
import struct
-
+logger = get_logger()
+PusRawTmList = List[bytearray]
+PusRawTmQueue = Deque[bytearray]
PusTmQueueT = Deque[PusTelemetry]
PusTmListT = List[PusTelemetry]
PusTmTupleT = Tuple[PusTmInfoT, PusTelemetry]
@@ -39,6 +42,7 @@ class PusTelemetryFactory(object):
return PusTelemetry(raw_tm_packet)
+
class Service2TM(PusTelemetry):
def __init__(self, byte_array: bytearray):
super().__init__(byte_array)
diff --git a/tm/obsw_tm_service_1.py b/tm/obsw_tm_service_1.py
index 248c4dd..c790828 100644
--- a/tm/obsw_tm_service_1.py
+++ b/tm/obsw_tm_service_1.py
@@ -9,13 +9,13 @@ import struct
from typing import Dict
from tm.obsw_pus_tm_base import PusTelemetry, TmDictionaryKeys
-
+from tm.obsw_pus_tm_creator import PusTelemetryCreator
pusPacketInfoService1T = Dict[TmDictionaryKeys, any]
class Service1TM(PusTelemetry):
- def __init__(self, byte_array: bytes):
+ def __init__(self, byte_array: bytearray):
super().__init__(byte_array)
self.has_tc_error_code = False
self.is_step_reply = False
@@ -84,3 +84,17 @@ class Service1TM(PusTelemetry):
if self.is_step_reply:
tm_information.update({TmDictionaryKeys.STEP_NUMBER: self.step_number})
return tm_information
+
+
+class Service1TmPacked(PusTelemetryCreator):
+ def __init__(self, subservice: int, ssc: int=0, tc_packed_id: int=0, tc_ssc: int=0):
+ source_data = bytearray()
+ source_data.append((tc_packed_id & 0xFF00) >> 8)
+ source_data.append(tc_packed_id & 0xFF)
+ tc_psc = (tc_ssc & 0x3FFF) | (0b11 << 16)
+ source_data.append((tc_psc & 0xFF00) >> 8)
+ source_data.append(tc_psc & 0xFF)
+ super().__init__(service=1, subservice=subservice, ssc=ssc, source_data=source_data)
+
+ def pack(self) -> bytearray:
+ return super().pack()
diff --git a/utility/obsw_logger.py b/utility/obsw_logger.py
index ab34380..d15de57 100644
--- a/utility/obsw_logger.py
+++ b/utility/obsw_logger.py
@@ -9,7 +9,16 @@ class InfoFilter(logging.Filter):
Filter object, which is used so that only INFO and DEBUG messages are printed to stdout.
"""
def filter(self, rec):
- if rec.levelno == logging.INFO or rec.levelno == logging.DEBUG:
+ if rec.levelno == logging.INFO:
+ return rec.levelno
+
+
+class DebugFilter(logging.Filter):
+ """
+ Filter object, which is used so that only INFO and DEBUG messages are printed to stdout.
+ """
+ def filter(self, rec):
+ if rec.levelno == logging.DEBUG:
return rec.levelno
@@ -27,9 +36,13 @@ def set_tmtc_logger() -> logging.Logger:
datefmt='%Y-%m-%d %H:%M:%S')
console_info_handler = logging.StreamHandler(stream=sys.stdout)
- console_info_handler.setLevel(logging.DEBUG)
+ console_info_handler.setLevel(logging.INFO)
console_info_handler.addFilter(InfoFilter())
+ console_debug_handler = logging.StreamHandler(stream=sys.stdout)
+ console_debug_handler.setLevel(logging.DEBUG)
+ console_debug_handler.addFilter(DebugFilter())
+
console_error_handler = logging.StreamHandler(stream=sys.stderr)
console_error_handler.setLevel(logging.WARNING)
try:
@@ -44,9 +57,11 @@ def set_tmtc_logger() -> logging.Logger:
file_handler.setFormatter(generic_format)
console_info_handler.setFormatter(generic_format)
+ console_debug_handler.setFormatter(fault_format)
console_error_handler.setFormatter(fault_format)
logger.addHandler(file_handler)
logger.addHandler(console_info_handler)
+ logger.addHandler(console_debug_handler)
logger.addHandler(console_error_handler)
return logger
--
GitLab