nautilus_common/python/
logging.rs

Help
1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2025 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::collections::HashMap;
17
18use log::LevelFilter;
19use nautilus_core::{UUID4, python::to_pyvalue_err};
20use nautilus_model::identifiers::TraderId;
21use pyo3::prelude::*;
22use ustr::Ustr;
23
24use crate::{
25    enums::{LogColor, LogLevel},
26    logging::{
27        self, headers,
28        logger::{self, LogGuard, LoggerConfig},
29        logging_clock_set_realtime_mode, logging_clock_set_static_mode,
30        logging_clock_set_static_time, logging_set_bypass, map_log_level_to_filter,
31        parse_level_filter_str,
32        writer::FileWriterConfig,
33    },
34};
35
36#[pymethods]
37impl LoggerConfig {
38    #[staticmethod]
39    #[pyo3(name = "from_spec")]
40    pub fn py_from_spec(spec: String) -> PyResult<Self> {
41        Self::from_spec(&spec).map_err(to_pyvalue_err)
42    }
43}
44
45#[pymethods]
46impl FileWriterConfig {
47    #[new]
48    #[pyo3(signature = (directory=None, file_name=None, file_format=None))]
49    #[must_use]
50    pub const fn py_new(
51        directory: Option<String>,
52        file_name: Option<String>,
53        file_format: Option<String>,
54    ) -> Self {
55        Self::new(directory, file_name, file_format)
56    }
57}
58
59/// Initialize tracing.
60///
61/// Tracing is meant to be used to trace/debug async Rust code. It can be
62/// configured to filter modules and write up to a specific level only using
63/// by passing a configuration using the `RUST_LOG` environment variable.
64///
65/// # Safety
66///
67/// Should only be called once during an applications run, ideally at the
68/// beginning of the run.
69///
70/// # Errors
71///
72/// Returns an error if tracing subscriber fails to initialize.
73#[pyfunction()]
74#[pyo3(name = "init_tracing")]
75pub fn py_init_tracing() -> PyResult<()> {
76    logging::init_tracing().map_err(to_pyvalue_err)
77}
78
79/// Initialize logging.
80///
81/// Logging should be used for Python and sync Rust logic which is most of
82/// the components in the [nautilus_trader](https://pypi.org/project/nautilus_trader) package.
83/// Logging can be configured to filter components and write up to a specific level only
84/// by passing a configuration using the `NAUTILUS_LOG` environment variable.
85///
86/// # Safety
87///
88/// Should only be called once during an applications run, ideally at the
89/// beginning of the run.
90#[pyfunction]
91#[pyo3(name = "init_logging")]
92#[allow(clippy::too_many_arguments)]
93#[pyo3(signature = (trader_id, instance_id, level_stdout, level_file=None, component_levels=None, directory=None, file_name=None, file_format=None, is_colored=None, is_bypassed=None, print_config=None))]
94pub fn py_init_logging(
95    trader_id: TraderId,
96    instance_id: UUID4,
97    level_stdout: LogLevel,
98    level_file: Option<LogLevel>,
99    component_levels: Option<HashMap<String, String>>,
100    directory: Option<String>,
101    file_name: Option<String>,
102    file_format: Option<String>,
103    is_colored: Option<bool>,
104    is_bypassed: Option<bool>,
105    print_config: Option<bool>,
106) -> PyResult<LogGuard> {
107    let level_file = level_file.map_or(LevelFilter::Off, map_log_level_to_filter);
108
109    let config = LoggerConfig::new(
110        map_log_level_to_filter(level_stdout),
111        level_file,
112        parse_component_levels(component_levels),
113        is_colored.unwrap_or(true),
114        print_config.unwrap_or(false),
115    );
116
117    let file_config = FileWriterConfig::new(directory, file_name, file_format);
118
119    if is_bypassed.unwrap_or(false) {
120        logging_set_bypass();
121    }
122
123    logging::init_logging(trader_id, instance_id, config, file_config).map_err(to_pyvalue_err)
124}
125
126fn parse_component_levels(
127    original_map: Option<HashMap<String, String>>,
128) -> HashMap<Ustr, LevelFilter> {
129    match original_map {
130        Some(map) => {
131            let mut new_map = HashMap::new();
132            for (key, value) in map {
133                let ustr_key = Ustr::from(&key);
134                let value = parse_level_filter_str(&value);
135                new_map.insert(ustr_key, value);
136            }
137            new_map
138        }
139        None => HashMap::new(),
140    }
141}
142
143/// Create a new log event.
144#[pyfunction]
145#[pyo3(name = "logger_log")]
146pub fn py_logger_log(level: LogLevel, color: LogColor, component: &str, message: &str) {
147    logger::log(level, color, Ustr::from(component), message);
148}
149
150/// Logs the standard Nautilus system header.
151#[pyfunction]
152#[pyo3(name = "log_header")]
153pub fn py_log_header(trader_id: TraderId, machine_id: &str, instance_id: UUID4, component: &str) {
154    headers::log_header(trader_id, machine_id, instance_id, Ustr::from(component));
155}
156
157/// Logs system information.
158#[pyfunction]
159#[pyo3(name = "log_sysinfo")]
160pub fn py_log_sysinfo(component: &str) {
161    headers::log_sysinfo(Ustr::from(component));
162}
163
164#[pyfunction]
165#[pyo3(name = "logging_clock_set_static_mode")]
166pub fn py_logging_clock_set_static_mode() {
167    logging_clock_set_static_mode();
168}
169
170#[pyfunction]
171#[pyo3(name = "logging_clock_set_realtime_mode")]
172pub fn py_logging_clock_set_realtime_mode() {
173    logging_clock_set_realtime_mode();
174}
175
176#[pyfunction]
177#[pyo3(name = "logging_clock_set_static_time")]
178pub fn py_logging_clock_set_static_time(time_ns: u64) {
179    logging_clock_set_static_time(time_ns);
180}
nautilus_common/python/logging.rs

nautilus_common/python/
logging.rs
