1pub use decl::time;
8
9pub(crate) use decl::module_def;
10
11#[pymodule(name = "time", with(#[cfg(any(unix, windows))] platform))]
12mod decl {
13 #![allow(unreachable_pub)]
14
15 #[cfg(any(unix, windows))]
16 use crate::builtins::PyBaseExceptionRef;
17 use crate::{
18 AsObject, Py, PyObjectRef, PyResult, VirtualMachine,
19 builtins::{PyStr, PyStrRef, PyTypeRef},
20 class::PyClassDef,
21 function::{Either, FuncArgs, OptionalArg, OptionalOption},
22 types::{PyStructSequence, PyStructSequenceData, struct_sequence_new},
23 };
24 #[cfg(target_os = "wasi")]
25 use crate::{
26 PyRef,
27 builtins::{PyNamespace, PyUtf8StrRef},
28 };
29 #[cfg(any(unix, windows))]
30 use crate::{
31 common::wtf8::Wtf8Buf,
32 convert::{ToPyException, ToPyObject},
33 };
34 use core::time::Duration;
35 #[cfg(not(any(unix, windows)))]
36 use jiff::{Timestamp, Zoned, civil::DateTime, tz::TimeZone};
37 #[cfg(target_os = "wasi")]
38 use rustpython_host_env::time::ClockId;
39 #[cfg(any(unix, windows))]
40 use rustpython_host_env::time::asctime_from_tm;
41 use rustpython_host_env::time::{self as host_time};
42
43 #[allow(dead_code)]
44 pub(super) const SEC_TO_MS: i64 = host_time::SEC_TO_MS;
45 #[allow(dead_code)]
46 pub(super) const MS_TO_US: i64 = host_time::MS_TO_US;
47 #[allow(dead_code)]
48 pub(super) const SEC_TO_US: i64 = host_time::SEC_TO_US;
49 #[allow(dead_code)]
50 pub(super) const US_TO_NS: i64 = host_time::US_TO_NS;
51 #[allow(dead_code)]
52 pub(super) const MS_TO_NS: i64 = host_time::MS_TO_NS;
53 #[allow(dead_code)]
54 pub(super) const SEC_TO_NS: i64 = host_time::SEC_TO_NS;
55 #[allow(dead_code)]
56 pub(super) const NS_TO_MS: i64 = host_time::NS_TO_MS;
57 #[allow(dead_code)]
58 pub(super) const NS_TO_US: i64 = host_time::NS_TO_US;
59
60 fn duration_since_system_now(vm: &VirtualMachine) -> PyResult<Duration> {
61 host_time::duration_since_system_now()
62 .map_err(|e| vm.new_value_error(format!("Time error: {e:?}")))
63 }
64
65 #[pyattr]
66 pub const _STRUCT_TM_ITEMS: usize = 11;
67
68 #[cfg(target_os = "wasi")]
69 fn get_clock_time(id: ClockId, vm: &VirtualMachine) -> PyResult<Duration> {
70 host_time::clock_gettime(id).map_err(|err| vm.new_os_error(err.to_string()))
71 }
72
73 #[cfg(target_os = "wasi")]
74 fn get_monotonic_time(vm: &VirtualMachine) -> PyResult<Duration> {
75 get_clock_time(ClockId::CLOCK_MONOTONIC, vm)
76 }
77
78 #[cfg(target_os = "wasi")]
79 fn get_perf_time(vm: &VirtualMachine) -> PyResult<Duration> {
80 get_clock_time(ClockId::CLOCK_MONOTONIC, vm)
81 }
82
83 #[cfg(target_os = "wasi")]
84 fn clock_getres(id: ClockId, vm: &VirtualMachine) -> PyResult<f64> {
85 host_time::clock_getres(id)
86 .map(|d| d.as_secs_f64())
87 .map_err(|err| vm.new_os_error(err.to_string()))
88 }
89
90 #[cfg(target_os = "wasi")]
91 #[pyfunction]
92 fn get_clock_info(name: PyUtf8StrRef, vm: &VirtualMachine) -> PyResult<PyRef<PyNamespace>> {
93 let (adj, imp, mono, res) = match name.as_str() {
94 "monotonic" | "perf_counter" => (
95 false,
96 "time.clock_gettime(CLOCK_MONOTONIC)",
97 true,
98 clock_getres(ClockId::CLOCK_MONOTONIC, vm)?,
99 ),
100 "time" => (
101 true,
102 "time.clock_gettime(CLOCK_REALTIME)",
103 false,
104 clock_getres(ClockId::CLOCK_REALTIME, vm)?,
105 ),
106 _ => return Err(vm.new_value_error("unknown clock")),
107 };
108
109 Ok(py_namespace!(vm, {
110 "implementation" => vm.new_pyobj(imp),
111 "monotonic" => vm.ctx.new_bool(mono),
112 "adjustable" => vm.ctx.new_bool(adj),
113 "resolution" => vm.ctx.new_float(res),
114 }))
115 }
116
117 #[cfg(not(any(unix, windows, target_os = "wasi")))]
118 fn get_monotonic_time(vm: &VirtualMachine) -> PyResult<Duration> {
119 duration_since_system_now(vm)
120 }
121
122 #[cfg(not(any(unix, windows, target_os = "wasi")))]
123 fn get_perf_time(vm: &VirtualMachine) -> PyResult<Duration> {
124 duration_since_system_now(vm)
125 }
126
127 #[pyfunction]
128 fn sleep(object: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
129 vm.audit("time.sleep", || (object.clone(),))?;
130
131 let seconds_type_name = object.class().name().to_owned();
132 let dur = object.try_into_value::<Duration>(vm).map_err(|e| {
133 if e.class().is(vm.ctx.exceptions.value_error)
134 && let Some(s) = e.args().as_slice().first().and_then(|arg| arg.str(vm).ok())
135 && s.as_bytes() == b"negative duration"
136 {
137 return vm.new_value_error("sleep length must be non-negative");
138 }
139 if e.class().is(vm.ctx.exceptions.type_error) {
140 return vm.new_type_error(format!(
141 "'{seconds_type_name}' object cannot be interpreted as an integer or float"
142 ));
143 }
144 e
145 })?;
146
147 #[cfg(unix)]
148 {
149 use std::time::Instant;
153 let deadline = Instant::now() + dur;
154 loop {
155 let remaining = deadline.saturating_duration_since(Instant::now());
156 if remaining.is_zero() {
157 break;
158 }
159 let sleep_result = vm.allow_threads(|| host_time::nanosleep(remaining));
160 match sleep_result {
161 Ok(()) => break,
162 Err(err) if err.raw_os_error() == Some(libc::EINTR) => {}
163 Err(err) => {
164 let errno = err.raw_os_error().unwrap_or(0);
165 return Err(vm.new_os_error(format!(
166 "nanosleep: {}",
167 host_time::nix_errno_display(errno)
168 )));
169 }
170 }
171 vm.check_signals()?;
173 }
174 }
175
176 #[cfg(not(unix))]
177 {
178 vm.allow_threads(|| std::thread::sleep(dur));
179 }
180
181 Ok(())
182 }
183
184 #[pyfunction]
185 fn time_ns(vm: &VirtualMachine) -> PyResult<u64> {
186 Ok(duration_since_system_now(vm)?.as_nanos() as u64)
187 }
188
189 #[pyfunction]
190 pub fn time(vm: &VirtualMachine) -> PyResult<f64> {
191 _time(vm)
192 }
193
194 #[cfg(not(all(
195 target_arch = "wasm32",
196 not(any(target_os = "emscripten", target_os = "wasi")),
197 )))]
198 fn _time(vm: &VirtualMachine) -> PyResult<f64> {
199 Ok(duration_since_system_now(vm)?.as_secs_f64())
200 }
201
202 #[cfg(all(
203 target_arch = "wasm32",
204 feature = "wasmbind",
205 not(any(target_os = "emscripten", target_os = "wasi"))
206 ))]
207 fn _time(_vm: &VirtualMachine) -> PyResult<f64> {
208 use wasm_bindgen::prelude::*;
209 #[wasm_bindgen]
210 extern "C" {
211 type Date;
212 #[wasm_bindgen(static_method_of = Date)]
213 fn now() -> f64;
214 }
215 Ok(Date::now() / 1000.0)
217 }
218
219 #[cfg(all(
220 target_arch = "wasm32",
221 not(feature = "wasmbind"),
222 not(any(target_os = "emscripten", target_os = "wasi"))
223 ))]
224 fn _time(vm: &VirtualMachine) -> PyResult<f64> {
225 Err(vm.new_not_implemented_error("time.time"))
226 }
227
228 #[pyfunction]
229 fn monotonic(vm: &VirtualMachine) -> PyResult<f64> {
230 Ok(get_monotonic_time(vm)?.as_secs_f64())
231 }
232
233 #[pyfunction]
234 fn monotonic_ns(vm: &VirtualMachine) -> PyResult<u128> {
235 Ok(get_monotonic_time(vm)?.as_nanos())
236 }
237
238 #[pyfunction]
239 fn perf_counter(vm: &VirtualMachine) -> PyResult<f64> {
240 Ok(get_perf_time(vm)?.as_secs_f64())
241 }
242
243 #[pyfunction]
244 fn perf_counter_ns(vm: &VirtualMachine) -> PyResult<u128> {
245 Ok(get_perf_time(vm)?.as_nanos())
246 }
247
248 #[cfg(windows)]
249 #[cfg(not(target_arch = "wasm32"))]
250 pub(super) fn get_tz_info() -> host_time::WindowsTimeZoneInfo {
251 host_time::get_tz_info()
252 }
253
254 #[cfg(not(target_env = "msvc"))]
260 #[cfg(not(target_arch = "wasm32"))]
261 #[pyattr]
262 fn altzone(_vm: &VirtualMachine) -> core::ffi::c_long {
263 crate::host_env::time::tz::altzone()
264 }
265
266 #[cfg(target_env = "msvc")]
267 #[cfg(not(target_arch = "wasm32"))]
268 #[pyattr]
269 fn altzone(_vm: &VirtualMachine) -> i32 {
270 let info = get_tz_info();
271 (info.bias + info.standard_bias) * 60 - 3600
273 }
274
275 #[cfg(not(target_env = "msvc"))]
276 #[cfg(not(target_arch = "wasm32"))]
277 #[pyattr]
278 fn timezone(_vm: &VirtualMachine) -> core::ffi::c_long {
279 crate::host_env::time::tz::timezone()
280 }
281
282 #[cfg(target_env = "msvc")]
283 #[cfg(not(target_arch = "wasm32"))]
284 #[pyattr]
285 fn timezone(_vm: &VirtualMachine) -> i32 {
286 let info = get_tz_info();
287 (info.bias + info.standard_bias) * 60
289 }
290
291 #[cfg(not(target_os = "freebsd"))]
292 #[cfg(not(target_env = "msvc"))]
293 #[cfg(not(target_arch = "wasm32"))]
294 #[pyattr]
295 fn daylight(_vm: &VirtualMachine) -> core::ffi::c_int {
296 crate::host_env::time::tz::daylight()
297 }
298
299 #[cfg(target_env = "msvc")]
300 #[cfg(not(target_arch = "wasm32"))]
301 #[pyattr]
302 fn daylight(_vm: &VirtualMachine) -> i32 {
303 let info = get_tz_info();
304 (info.standard_bias != info.daylight_bias) as i32
306 }
307
308 #[cfg(not(target_env = "msvc"))]
309 #[cfg(not(target_arch = "wasm32"))]
310 #[pyattr]
311 fn tzname(vm: &VirtualMachine) -> crate::builtins::PyTupleRef {
312 use crate::builtins::tuple::IntoPyTuple;
313 crate::host_env::time::tz::tzname_strings().into_pytuple(vm)
314 }
315
316 #[cfg(target_env = "msvc")]
317 #[cfg(not(target_arch = "wasm32"))]
318 #[pyattr]
319 fn tzname(vm: &VirtualMachine) -> crate::builtins::PyTupleRef {
320 use crate::builtins::tuple::IntoPyTuple;
321 let info = get_tz_info();
322 let tz_name = (&*info.standard_name, &*info.daylight_name);
323 tz_name.into_pytuple(vm)
324 }
325
326 #[cfg(not(any(unix, windows)))]
327 fn pyobj_to_timestamp(value: Either<f64, i64>, vm: &VirtualMachine) -> PyResult<Timestamp> {
328 let secs = match value {
329 Either::A(float) => {
330 if !float.is_finite() {
331 return Err(vm.new_value_error("Invalid value for timestamp"));
332 }
333 float.floor() as i64
334 }
335 Either::B(int) => int,
336 };
337 Timestamp::from_second(secs)
338 .map_err(|_| vm.new_overflow_error("timestamp out of range for platform time_t"))
339 }
340
341 #[cfg(not(any(unix, windows)))]
342 fn naive_or_local(secs: Option<Either<f64, i64>>, vm: &VirtualMachine) -> PyResult<Zoned> {
343 Ok(match secs {
344 Some(secs) => pyobj_to_timestamp(secs, vm)?.to_zoned(TimeZone::system()),
345 None => Zoned::now(),
346 })
347 }
348
349 #[cfg(any(unix, windows))]
350 fn checked_tm_from_struct_time(
351 t: &StructTimeData,
352 vm: &VirtualMachine,
353 func_name: &'static str,
354 ) -> PyResult<host_time::CheckedTm> {
355 let invalid_tuple =
356 || vm.new_type_error(format!("{func_name}(): illegal time tuple argument"));
357 let classify_err = |e: PyBaseExceptionRef| {
358 if e.class().is(vm.ctx.exceptions.overflow_error) {
359 vm.new_overflow_error(format!("{func_name} argument out of range"))
360 } else {
361 invalid_tuple()
362 }
363 };
364
365 let year: i64 = t.tm_year.clone().try_into_value(vm).map_err(|e| {
366 if e.class().is(vm.ctx.exceptions.overflow_error) {
367 vm.new_overflow_error("year out of range")
368 } else {
369 invalid_tuple()
370 }
371 })?;
372 if year < i64::from(i32::MIN) + 1900 || year > i64::from(i32::MAX) {
373 return Err(vm.new_overflow_error("year out of range"));
374 }
375 let year = year as i32;
376 let tm_mon = t
377 .tm_mon
378 .clone()
379 .try_into_value::<i32>(vm)
380 .map_err(classify_err)?
381 - 1;
382 let tm_mday = t.tm_mday.clone().try_into_value(vm).map_err(classify_err)?;
383 let tm_hour = t.tm_hour.clone().try_into_value(vm).map_err(classify_err)?;
384 let tm_min = t.tm_min.clone().try_into_value(vm).map_err(classify_err)?;
385 let tm_sec = t.tm_sec.clone().try_into_value(vm).map_err(classify_err)?;
386 let tm_wday = (t
387 .tm_wday
388 .clone()
389 .try_into_value::<i32>(vm)
390 .map_err(classify_err)?
391 + 1)
392 % 7;
393 let tm_yday = t
394 .tm_yday
395 .clone()
396 .try_into_value::<i32>(vm)
397 .map_err(classify_err)?
398 - 1;
399 let tm_isdst = t
400 .tm_isdst
401 .clone()
402 .try_into_value(vm)
403 .map_err(classify_err)?;
404
405 #[cfg(unix)]
406 {
407 use crate::builtins::PyUtf8StrRef;
408 let zone = if t.tm_zone.is(&vm.ctx.none) {
409 None
410 } else {
411 let zone: PyUtf8StrRef = t
412 .tm_zone
413 .clone()
414 .try_into_value(vm)
415 .map_err(|_| invalid_tuple())?;
416 Some(zone.as_str().to_owned())
417 };
418 let gmtoff = if t.tm_gmtoff.is(&vm.ctx.none) {
419 None
420 } else {
421 Some(
422 t.tm_gmtoff
423 .clone()
424 .try_into_value::<i64>(vm)
425 .map_err(classify_err)?,
426 )
427 };
428 host_time::checked_tm_from_parts(host_time::CheckedTmParts {
429 year: year.into(),
430 tm_mon,
431 tm_mday,
432 tm_hour,
433 tm_min,
434 tm_sec,
435 tm_wday,
436 tm_yday,
437 tm_isdst,
438 zone,
439 gmtoff,
440 })
441 .map_err(|err| err.to_pyexception(vm))
442 }
443 #[cfg(windows)]
444 {
445 host_time::checked_tm_from_parts(host_time::CheckedTmParts {
446 year: year.into(),
447 tm_mon,
448 tm_mday,
449 tm_hour,
450 tm_min,
451 tm_sec,
452 tm_wday,
453 tm_yday,
454 tm_isdst,
455 })
456 .map_err(|err| err.to_pyexception(vm))
457 }
458 }
459
460 #[cfg(not(any(unix, windows)))]
461 impl OptionalArg<StructTimeData> {
462 fn naive_or_local(self, vm: &VirtualMachine) -> PyResult<DateTime> {
463 Ok(match self {
464 Self::Present(t) => t.to_date_time(vm)?,
465 Self::Missing => Zoned::now().datetime(),
466 })
467 }
468 }
469
470 #[pyfunction]
472 fn gmtime(
473 secs: OptionalOption<Either<f64, i64>>,
474 vm: &VirtualMachine,
475 ) -> PyResult<StructTimeData> {
476 let secs = secs.flatten();
478 cfg_select! {
479 any(unix, windows) => {
480 let ts = match secs {
481 Some(value) => pyobj_to_time_t(value, vm)?,
482 None => current_time_t(),
483 };
484 gmtime_from_timestamp(ts, vm)
485 }
486 _ => {
487 let instant = match secs {
488 Some(secs) => pyobj_to_timestamp(secs, vm)?.to_zoned(TimeZone::UTC),
489 None => Zoned::now().with_time_zone(TimeZone::UTC),
490 };
491 Ok(StructTimeData::new_utc(vm, instant))
492 }
493 }
494 }
495
496 #[pyfunction]
497 fn localtime(
498 secs: OptionalOption<Either<f64, i64>>,
499 vm: &VirtualMachine,
500 ) -> PyResult<StructTimeData> {
501 let secs = secs.flatten();
503 cfg_select! {
504 any(unix, windows) => {
505 let ts = match secs {
506 Some(value) => pyobj_to_time_t(value, vm)?,
507 None => current_time_t(),
508 };
509 localtime_from_timestamp(ts, vm)
510 }
511 _ => {
512 let instant = naive_or_local(secs, vm)?;
513 StructTimeData::new_local(vm, instant.into(), 0)
514 }
515 }
516 }
517
518 #[pyfunction]
519 fn mktime(object: StructTimeData, vm: &VirtualMachine) -> PyResult<f64> {
520 #[cfg(unix)]
521 {
522 unix_mktime(&object, vm)
523 }
524
525 #[cfg(windows)]
526 {
527 win_mktime(&object, vm)
528 }
529
530 #[cfg(not(any(unix, windows)))]
531 {
532 let datetime = object.to_date_time(vm)?;
533 let local_dt = datetime
535 .to_zoned(TimeZone::system())
536 .map_err(|_| vm.new_overflow_error("mktime argument out of range"))?;
537 let seconds_since_epoch = local_dt.timestamp().as_second() as f64;
538 Ok(seconds_since_epoch)
539 }
540 }
541
542 #[cfg(not(any(unix, windows)))]
543 const CFMT: &str = "%a %b %e %H:%M:%S %Y";
544
545 #[pyfunction]
546 fn asctime(t: OptionalArg<StructTimeData>, vm: &VirtualMachine) -> PyResult {
547 #[cfg(any(unix, windows))]
548 {
549 let tm = match t {
550 OptionalArg::Present(value) => {
551 checked_tm_from_struct_time(&value, vm, "asctime")?.tm
552 }
553 OptionalArg::Missing => {
554 let now = current_time_t();
555 let local = localtime_from_timestamp(now, vm)?;
556 checked_tm_from_struct_time(&local, vm, "asctime")?.tm
557 }
558 };
559 Ok(vm.ctx.new_str(asctime_from_tm(&tm)).into())
560 }
561
562 #[cfg(not(any(unix, windows)))]
563 {
564 let instant = t.naive_or_local(vm)?;
565 let formatted_time = instant.strftime(CFMT).to_string();
566 Ok(vm.ctx.new_str(formatted_time).into())
567 }
568 }
569
570 #[pyfunction]
571 fn ctime(secs: OptionalOption<Either<f64, i64>>, vm: &VirtualMachine) -> PyResult<String> {
572 let secs = secs.flatten();
574 #[cfg(any(unix, windows))]
575 {
576 let ts = match secs {
577 Some(value) => pyobj_to_time_t(value, vm)?,
578 None => current_time_t(),
579 };
580 let local = localtime_from_timestamp(ts, vm)?;
581 let tm = checked_tm_from_struct_time(&local, vm, "asctime")?.tm;
582 Ok(asctime_from_tm(&tm))
583 }
584
585 #[cfg(not(any(unix, windows)))]
586 {
587 let instant = naive_or_local(secs, vm)?;
588 Ok(instant.strftime(CFMT).to_string())
589 }
590 }
591
592 #[cfg(any(unix, windows))]
593 fn strftime_crt(
594 format: &Py<PyStr>,
595 checked_tm: host_time::CheckedTm,
596 vm: &VirtualMachine,
597 ) -> PyResult {
598 #[cfg(unix)]
599 let _keep_zone_alive = &checked_tm.zone;
600 let mut tm = checked_tm.tm;
601 tm.tm_isdst = tm.tm_isdst.clamp(-1, 1);
602
603 #[cfg(windows)]
605 {
606 let year = tm.tm_year + 1900;
607 if !(1..=9999).contains(&year) {
608 return Err(vm.new_value_error("strftime() requires year in [1; 9999]"));
609 }
610 }
611
612 let mut out = Wtf8Buf::new();
613 let mut ascii = String::new();
614
615 for codepoint in format.as_wtf8().code_points() {
616 if codepoint.to_u32() == 0 {
617 if !ascii.is_empty() {
618 let part =
619 host_time::strftime_ascii(&ascii, &tm).map_err(|e| e.to_pyexception(vm))?;
620 out.extend(part.chars());
621 ascii.clear();
622 }
623 out.push(codepoint);
624 continue;
625 }
626 if let Some(ch) = codepoint.to_char()
627 && ch.is_ascii()
628 {
629 ascii.push(ch);
630 continue;
631 }
632
633 if !ascii.is_empty() {
634 let part =
635 host_time::strftime_ascii(&ascii, &tm).map_err(|e| e.to_pyexception(vm))?;
636 out.extend(part.chars());
637 ascii.clear();
638 }
639 out.push(codepoint);
640 }
641 if !ascii.is_empty() {
642 let part = host_time::strftime_ascii(&ascii, &tm).map_err(|e| e.to_pyexception(vm))?;
643 out.extend(part.chars());
644 }
645 Ok(out.to_pyobject(vm))
646 }
647
648 #[pyfunction]
649 #[cfg_attr(not(any(unix, windows)), expect(clippy::unnecessary_wraps,))]
650 fn strftime(format: PyStrRef, t: OptionalArg<StructTimeData>, vm: &VirtualMachine) -> PyResult {
651 #[cfg(any(unix, windows))]
652 {
653 let checked_tm = match t {
654 OptionalArg::Present(value) => checked_tm_from_struct_time(&value, vm, "strftime")?,
655 OptionalArg::Missing => {
656 let now = current_time_t();
657 let local = localtime_from_timestamp(now, vm)?;
658 checked_tm_from_struct_time(&local, vm, "strftime")?
659 }
660 };
661 strftime_crt(&format, checked_tm, vm)
662 }
663
664 #[cfg(not(any(unix, windows)))]
665 {
666 use core::fmt::Write;
667
668 let fmt_lossy = format.to_string_lossy();
669
670 let instant = match t.naive_or_local(vm) {
674 Ok(dt) => dt,
675 Err(_) => return Ok(vm.ctx.new_str(fmt_lossy.into_owned()).into()),
676 };
677
678 let mut formatted_time = String::new();
679 write!(&mut formatted_time, "{}", instant.strftime(&*fmt_lossy))
680 .unwrap_or_else(|_| formatted_time = format.to_string());
681 Ok(vm.ctx.new_str(formatted_time).into())
682 }
683 }
684
685 #[pyfunction]
686 fn strptime(string: PyStrRef, format: OptionalArg<PyStrRef>, vm: &VirtualMachine) -> PyResult {
687 let strptime_module = vm.import("_strptime", 0)?;
689 let strptime_func = strptime_module.get_attr("_strptime_time", vm)?;
690
691 match format.into_option() {
693 Some(fmt) => strptime_func.call((string, fmt), vm),
694 None => strptime_func.call((string,), vm),
695 }
696 }
697
698 #[cfg(not(any(
699 windows,
700 target_vendor = "apple",
701 target_os = "android",
702 target_os = "dragonfly",
703 target_os = "freebsd",
704 target_os = "linux",
705 target_os = "fuchsia",
706 target_os = "emscripten",
707 )))]
708 fn get_thread_time(vm: &VirtualMachine) -> PyResult<Duration> {
709 Err(vm.new_not_implemented_error("thread time unsupported in this system"))
710 }
711
712 #[pyfunction]
713 fn thread_time(vm: &VirtualMachine) -> PyResult<f64> {
714 Ok(get_thread_time(vm)?.as_secs_f64())
715 }
716
717 #[pyfunction]
718 fn thread_time_ns(vm: &VirtualMachine) -> PyResult<u64> {
719 Ok(get_thread_time(vm)?.as_nanos() as u64)
720 }
721
722 #[cfg(any(windows, all(target_arch = "wasm32", target_os = "emscripten")))]
723 pub(super) fn time_muldiv(ticks: i64, mul: i64, div: i64) -> u64 {
724 let int_part = ticks / div;
725 let ticks = ticks % div;
726 let remaining = (ticks * mul) / div;
727 (int_part * mul + remaining) as u64
728 }
729
730 #[cfg(all(target_arch = "wasm32", target_os = "emscripten"))]
731 fn get_process_time(vm: &VirtualMachine) -> PyResult<Duration> {
732 let times =
733 host_time::process_times().map_err(|_| vm.new_os_error("Failed to get clock time"))?;
734 Ok(Duration::from_secs_f64(times.user + times.system))
735 }
736
737 #[cfg(not(any(
738 windows,
739 target_os = "macos",
740 target_os = "android",
741 target_os = "dragonfly",
742 target_os = "freebsd",
743 target_os = "linux",
744 target_os = "illumos",
745 target_os = "netbsd",
746 target_os = "solaris",
747 target_os = "openbsd",
748 target_os = "redox",
749 all(target_arch = "wasm32", target_os = "emscripten")
750 )))]
751 fn get_process_time(vm: &VirtualMachine) -> PyResult<Duration> {
752 Err(vm.new_not_implemented_error("process time unsupported in this system"))
753 }
754
755 #[pyfunction]
756 fn process_time(vm: &VirtualMachine) -> PyResult<f64> {
757 Ok(get_process_time(vm)?.as_secs_f64())
758 }
759
760 #[pyfunction]
761 fn process_time_ns(vm: &VirtualMachine) -> PyResult<u64> {
762 Ok(get_process_time(vm)?.as_nanos() as u64)
763 }
764
765 #[pystruct_sequence_data(try_from_object)]
767 pub struct StructTimeData {
768 pub tm_year: PyObjectRef,
769 pub tm_mon: PyObjectRef,
770 pub tm_mday: PyObjectRef,
771 pub tm_hour: PyObjectRef,
772 pub tm_min: PyObjectRef,
773 pub tm_sec: PyObjectRef,
774 pub tm_wday: PyObjectRef,
775 pub tm_yday: PyObjectRef,
776 pub tm_isdst: PyObjectRef,
777 #[pystruct_sequence(skip)]
778 pub tm_zone: PyObjectRef,
779 #[pystruct_sequence(skip)]
780 pub tm_gmtoff: PyObjectRef,
781 }
782
783 impl core::fmt::Debug for StructTimeData {
784 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
785 write!(f, "struct_time()")
786 }
787 }
788
789 impl StructTimeData {
790 #[cfg(not(any(unix, windows)))]
791 fn new_inner(vm: &VirtualMachine, tm: Zoned, isdst: i32) -> Self {
792 Self {
793 tm_year: vm.ctx.new_int(tm.year()).into(),
794 tm_mon: vm.ctx.new_int(tm.month()).into(),
795 tm_mday: vm.ctx.new_int(tm.day()).into(),
796 tm_hour: vm.ctx.new_int(tm.hour()).into(),
797 tm_min: vm.ctx.new_int(tm.minute()).into(),
798 tm_sec: vm.ctx.new_int(tm.second()).into(),
799 tm_wday: vm.ctx.new_int(tm.weekday().to_sunday_zero_offset()).into(),
800 tm_yday: vm.ctx.new_int(tm.day_of_year()).into(),
801 tm_isdst: vm.ctx.new_int(isdst).into(),
802 tm_zone: vm.ctx.new_str(tm.strftime("%Z").to_string()).into(),
803 tm_gmtoff: vm.ctx.new_int(tm.offset().seconds()).into(),
804 }
805 }
806
807 #[cfg(not(any(unix, windows)))]
809 fn new_utc(vm: &VirtualMachine, tm: Zoned) -> Self {
810 Self::new_inner(vm, tm, 0)
811 }
812
813 #[cfg(not(any(unix, windows)))]
815 fn new_local(vm: &VirtualMachine, tm: DateTime, isdst: i32) -> PyResult<Self> {
816 tm.to_zoned(TimeZone::system())
817 .map(|tm| Self::new_inner(vm, tm, isdst))
818 .map_err(|_| {
819 vm.new_overflow_error("timestamp is ambiguous for the system timezone")
820 })
821 }
822
823 #[cfg(not(any(unix, windows)))]
824 fn to_date_time(&self, vm: &VirtualMachine) -> PyResult<DateTime> {
825 macro_rules! field {
826 ($field:ident) => {
827 self.$field.clone().try_into_value(vm)?
828 };
829 }
830 DateTime::new(
831 field!(tm_year),
832 field!(tm_mon),
833 field!(tm_mday),
834 field!(tm_hour),
835 field!(tm_min),
836 field!(tm_sec),
837 0,
838 )
839 .map_err(|_| vm.new_overflow_error("mktime argument out of range"))
840 }
841 }
842
843 #[pyattr]
844 #[pystruct_sequence(name = "struct_time", module = "time", data = "StructTimeData")]
845 pub struct PyStructTime;
846
847 #[pyclass(with(PyStructSequence))]
848 impl PyStructTime {
849 #[pyslot]
850 fn slot_new(cls: PyTypeRef, args: FuncArgs, vm: &VirtualMachine) -> PyResult {
851 struct_sequence_new(
852 cls,
853 args.bind_for(vm, Self::NAME)?,
854 StructTimeData::OPTIONAL_FIELD_NAMES,
855 vm,
856 )
857 }
858 }
859
860 #[cfg(any(unix, windows))]
862 pub(super) fn tm_from_struct_time(
863 t: &StructTimeData,
864 vm: &VirtualMachine,
865 ) -> PyResult<libc::tm> {
866 let invalid_tuple = || vm.new_type_error("mktime(): illegal time tuple argument");
867 let classify_err = |e: PyBaseExceptionRef| {
868 if e.class().is(vm.ctx.exceptions.overflow_error) {
869 vm.new_overflow_error("mktime argument out of range")
870 } else {
871 invalid_tuple()
872 }
873 };
874 let year: i32 = t.tm_year.clone().try_into_value(vm).map_err(classify_err)?;
875 if year < i32::MIN + 1900 {
876 return Err(vm.new_overflow_error("year out of range"));
877 }
878
879 host_time::mktime_tm_from_parts(host_time::MktimeTmParts {
880 year,
881 tm_sec: t.tm_sec.clone().try_into_value(vm).map_err(classify_err)?,
882 tm_min: t.tm_min.clone().try_into_value(vm).map_err(classify_err)?,
883 tm_hour: t.tm_hour.clone().try_into_value(vm).map_err(classify_err)?,
884 tm_mday: t.tm_mday.clone().try_into_value(vm).map_err(classify_err)?,
885 tm_mon: t
886 .tm_mon
887 .clone()
888 .try_into_value::<i32>(vm)
889 .map_err(classify_err)?,
890 tm_yday: t
891 .tm_yday
892 .clone()
893 .try_into_value::<i32>(vm)
894 .map_err(classify_err)?,
895 tm_isdst: t
896 .tm_isdst
897 .clone()
898 .try_into_value(vm)
899 .map_err(classify_err)?,
900 })
901 .map_err(|err| match err {
902 host_time::CheckedTmError::YearOutOfRange => vm.new_overflow_error("year out of range"),
903 _ => vm.new_type_error("mktime(): illegal time tuple argument"),
904 })
905 }
906
907 #[cfg(any(unix, windows))]
908 #[cfg_attr(target_env = "musl", allow(deprecated))]
909 fn pyobj_to_time_t(value: Either<f64, i64>, vm: &VirtualMachine) -> PyResult<libc::time_t> {
910 match value {
911 Either::A(float) => {
912 if !float.is_finite() {
913 return Err(vm.new_value_error("Invalid value for timestamp"));
914 }
915 let secs = float.floor();
916 #[cfg_attr(target_env = "musl", allow(deprecated))]
917 if secs < libc::time_t::MIN as f64 || secs > libc::time_t::MAX as f64 {
918 return Err(vm.new_overflow_error("timestamp out of range for platform time_t"));
919 }
920 #[cfg_attr(target_env = "musl", allow(deprecated))]
921 Ok(secs as libc::time_t)
922 }
923 Either::B(int) => {
924 #[allow(clippy::useless_conversion)]
926 #[cfg_attr(target_env = "musl", allow(deprecated))]
927 let ts: libc::time_t = int.try_into().map_err(|_| {
928 vm.new_overflow_error("timestamp out of range for platform time_t")
929 })?;
930 Ok(ts)
931 }
932 }
933 }
934
935 #[cfg(any(unix, windows))]
936 #[allow(unused_imports)]
937 use super::platform::*;
938
939 #[expect(clippy::unnecessary_wraps, reason = "Needs to comply with a signature")]
940 pub(crate) fn module_exec(
941 vm: &VirtualMachine,
942 module: &Py<crate::builtins::PyModule>,
943 ) -> PyResult<()> {
944 #[cfg(not(target_env = "msvc"))]
945 #[cfg(not(target_arch = "wasm32"))]
946 crate::host_env::time::tz::tzset();
947
948 __module_exec(vm, module);
949 Ok(())
950 }
951}
952
953#[cfg(unix)]
954#[pymodule(sub, name = "time")]
955mod platform {
956 #[allow(unused_imports)]
957 use super::decl::{SEC_TO_NS, StructTimeData, US_TO_NS};
958 #[cfg_attr(target_os = "macos", allow(unused_imports))]
959 use crate::{
960 PyObject, PyRef, PyResult, TryFromBorrowedObject, VirtualMachine,
961 builtins::{PyNamespace, PyUtf8StrRef},
962 convert::IntoPyException,
963 };
964 use core::time::Duration;
965 #[cfg(any(
966 target_os = "illumos",
967 target_os = "netbsd",
968 target_os = "openbsd",
969 target_os = "solaris",
970 ))]
971 use rustpython_host_env::resource as host_resource;
972 use rustpython_host_env::time::{self as host_time, ClockId};
973
974 #[cfg(target_os = "solaris")]
975 #[pyattr]
976 use host_time::CLOCK_HIGHRES;
977 #[cfg(any(target_os = "linux", target_vendor = "apple"))]
978 #[pyattr]
979 use host_time::CLOCK_MONOTONIC_RAW;
980 #[cfg(not(any(
981 target_os = "illumos",
982 target_os = "netbsd",
983 target_os = "solaris",
984 target_os = "openbsd",
985 target_os = "wasi",
986 )))]
987 #[pyattr]
988 use host_time::CLOCK_PROCESS_CPUTIME_ID;
989 #[cfg(not(any(
990 target_os = "illumos",
991 target_os = "netbsd",
992 target_os = "solaris",
993 target_os = "openbsd",
994 target_os = "redox",
995 )))]
996 #[pyattr]
997 use host_time::CLOCK_THREAD_CPUTIME_ID;
998 #[cfg(target_os = "linux")]
999 #[pyattr]
1000 use host_time::{CLOCK_BOOTTIME, CLOCK_TAI};
1001 #[pyattr]
1002 use host_time::{CLOCK_MONOTONIC, CLOCK_REALTIME};
1003 #[cfg(target_vendor = "apple")]
1004 #[pyattr]
1005 use host_time::{CLOCK_MONOTONIC_RAW_APPROX, CLOCK_UPTIME_RAW, CLOCK_UPTIME_RAW_APPROX};
1006 #[cfg(any(target_os = "freebsd", target_os = "openbsd", target_os = "dragonfly"))]
1007 #[pyattr]
1008 use host_time::{CLOCK_PROF, CLOCK_UPTIME};
1009
1010 impl<'a> TryFromBorrowedObject<'a> for ClockId {
1011 fn try_from_borrowed_object(vm: &VirtualMachine, obj: &'a PyObject) -> PyResult<Self> {
1012 obj.try_to_value(vm).map(Self::from_raw)
1013 }
1014 }
1015
1016 fn struct_time_from_tm(vm: &VirtualMachine, tm: libc::tm) -> StructTimeData {
1017 let zone = unsafe {
1018 if tm.tm_zone.is_null() {
1019 String::new()
1020 } else {
1021 core::ffi::CStr::from_ptr(tm.tm_zone)
1022 .to_string_lossy()
1023 .into_owned()
1024 }
1025 };
1026 StructTimeData {
1027 tm_year: vm.ctx.new_int(tm.tm_year + 1900).into(),
1028 tm_mon: vm.ctx.new_int(tm.tm_mon + 1).into(),
1029 tm_mday: vm.ctx.new_int(tm.tm_mday).into(),
1030 tm_hour: vm.ctx.new_int(tm.tm_hour).into(),
1031 tm_min: vm.ctx.new_int(tm.tm_min).into(),
1032 tm_sec: vm.ctx.new_int(tm.tm_sec).into(),
1033 tm_wday: vm.ctx.new_int((tm.tm_wday + 6) % 7).into(),
1034 tm_yday: vm.ctx.new_int(tm.tm_yday + 1).into(),
1035 tm_isdst: vm.ctx.new_int(tm.tm_isdst).into(),
1036 tm_zone: vm.ctx.new_str(zone).into(),
1037 tm_gmtoff: vm.ctx.new_int(tm.tm_gmtoff).into(),
1038 }
1039 }
1040
1041 pub(super) fn current_time_t() -> host_time::TimeT {
1042 host_time::current_time_t()
1043 }
1044
1045 pub(super) fn gmtime_from_timestamp(
1046 when: host_time::TimeT,
1047 vm: &VirtualMachine,
1048 ) -> PyResult<StructTimeData> {
1049 let Some(tm) = host_time::gmtime_from_timestamp(when) else {
1050 return Err(vm.new_overflow_error("timestamp out of range for platform time_t"));
1051 };
1052 Ok(struct_time_from_tm(vm, tm))
1053 }
1054
1055 pub(super) fn localtime_from_timestamp(
1056 when: host_time::TimeT,
1057 vm: &VirtualMachine,
1058 ) -> PyResult<StructTimeData> {
1059 let Some(tm) = host_time::localtime_from_timestamp(when) else {
1060 return Err(vm.new_overflow_error("timestamp out of range for platform time_t"));
1061 };
1062 Ok(struct_time_from_tm(vm, tm))
1063 }
1064
1065 pub(super) fn unix_mktime(t: &StructTimeData, vm: &VirtualMachine) -> PyResult<f64> {
1066 let mut tm = super::decl::tm_from_struct_time(t, vm)?;
1067 let timestamp = host_time::mktime(&mut tm);
1068 if timestamp == -1 && tm.tm_wday == -1 {
1069 return Err(vm.new_overflow_error("mktime argument out of range"));
1070 }
1071 Ok(timestamp as f64)
1072 }
1073
1074 fn get_clock_time(clk_id: ClockId, vm: &VirtualMachine) -> PyResult<Duration> {
1075 rustpython_host_env::time::clock_gettime(clk_id).map_err(|e| e.into_pyexception(vm))
1076 }
1077
1078 #[cfg(target_vendor = "apple")]
1083 fn monotonic_clock_id() -> ClockId {
1084 ClockId::CLOCK_UPTIME_RAW
1085 }
1086
1087 #[cfg(not(target_vendor = "apple"))]
1088 fn monotonic_clock_id() -> ClockId {
1089 ClockId::CLOCK_MONOTONIC
1090 }
1091
1092 #[pyfunction]
1093 fn clock_gettime(clk_id: ClockId, vm: &VirtualMachine) -> PyResult<f64> {
1094 get_clock_time(clk_id, vm).map(|d| d.as_secs_f64())
1095 }
1096
1097 #[pyfunction]
1098 fn clock_gettime_ns(clk_id: ClockId, vm: &VirtualMachine) -> PyResult<u128> {
1099 get_clock_time(clk_id, vm).map(|d| d.as_nanos())
1100 }
1101
1102 #[cfg(not(target_os = "redox"))]
1103 #[pyfunction]
1104 fn clock_getres(clk_id: ClockId, vm: &VirtualMachine) -> PyResult<f64> {
1105 rustpython_host_env::time::clock_getres(clk_id)
1106 .map(|d| d.as_secs_f64())
1107 .map_err(|e| e.into_pyexception(vm))
1108 }
1109
1110 #[cfg(not(target_os = "redox"))]
1111 #[cfg(any(not(target_vendor = "apple"), target_os = "macos"))]
1112 #[pyfunction]
1113 fn clock_settime(clk_id: ClockId, time: Duration, vm: &VirtualMachine) -> PyResult<()> {
1114 rustpython_host_env::time::clock_settime(clk_id, time).map_err(|e| e.into_pyexception(vm))
1115 }
1116
1117 #[cfg(not(target_os = "redox"))]
1118 #[cfg(any(not(target_vendor = "apple"), target_os = "macos"))]
1119 #[cfg_attr(target_env = "musl", allow(deprecated))]
1120 #[pyfunction]
1121 fn clock_settime_ns(clk_id: ClockId, time: libc::time_t, vm: &VirtualMachine) -> PyResult<()> {
1122 rustpython_host_env::time::clock_settime(clk_id, Duration::from_nanos(time as _))
1123 .map_err(|e| e.into_pyexception(vm))
1124 }
1125
1126 #[cfg(any(
1128 target_os = "macos",
1129 target_os = "android",
1130 target_os = "dragonfly",
1131 target_os = "freebsd",
1132 target_os = "fuchsia",
1133 target_os = "emscripten",
1134 target_os = "linux",
1135 ))]
1136 #[pyfunction]
1137 fn get_clock_info(name: PyUtf8StrRef, vm: &VirtualMachine) -> PyResult<PyRef<PyNamespace>> {
1138 let (adj, imp, mono, res) = match name.as_str() {
1139 #[cfg(target_vendor = "apple")]
1140 "monotonic" | "perf_counter" => (
1141 false,
1142 "mach_absolute_time()",
1143 true,
1144 clock_getres(monotonic_clock_id(), vm)?,
1145 ),
1146 #[cfg(not(target_vendor = "apple"))]
1147 "monotonic" | "perf_counter" => (
1148 false,
1149 "time.clock_gettime(CLOCK_MONOTONIC)",
1150 true,
1151 clock_getres(monotonic_clock_id(), vm)?,
1152 ),
1153 "process_time" => (
1154 false,
1155 "time.clock_gettime(CLOCK_PROCESS_CPUTIME_ID)",
1156 true,
1157 clock_getres(ClockId::CLOCK_PROCESS_CPUTIME_ID, vm)?,
1158 ),
1159 "thread_time" => (
1160 false,
1161 "time.clock_gettime(CLOCK_THREAD_CPUTIME_ID)",
1162 true,
1163 clock_getres(ClockId::CLOCK_THREAD_CPUTIME_ID, vm)?,
1164 ),
1165 "time" => (
1166 true,
1167 "time.clock_gettime(CLOCK_REALTIME)",
1168 false,
1169 clock_getres(ClockId::CLOCK_REALTIME, vm)?,
1170 ),
1171 _ => return Err(vm.new_value_error("unknown clock")),
1172 };
1173
1174 Ok(py_namespace!(vm, {
1175 "implementation" => vm.new_pyobj(imp),
1176 "monotonic" => vm.ctx.new_bool(mono),
1177 "adjustable" => vm.ctx.new_bool(adj),
1178 "resolution" => vm.ctx.new_float(res),
1179 }))
1180 }
1181
1182 #[cfg(not(any(
1183 target_os = "macos",
1184 target_os = "android",
1185 target_os = "dragonfly",
1186 target_os = "freebsd",
1187 target_os = "fuchsia",
1188 target_os = "emscripten",
1189 target_os = "linux",
1190 )))]
1191 #[pyfunction]
1192 fn get_clock_info(_name: PyUtf8StrRef, vm: &VirtualMachine) -> PyResult<PyRef<PyNamespace>> {
1193 Err(vm.new_not_implemented_error("get_clock_info unsupported on this system"))
1194 }
1195
1196 pub(super) fn get_monotonic_time(vm: &VirtualMachine) -> PyResult<Duration> {
1197 get_clock_time(monotonic_clock_id(), vm)
1198 }
1199
1200 pub(super) fn get_perf_time(vm: &VirtualMachine) -> PyResult<Duration> {
1201 get_clock_time(monotonic_clock_id(), vm)
1202 }
1203
1204 #[cfg(not(any(
1205 target_os = "illumos",
1206 target_os = "netbsd",
1207 target_os = "openbsd",
1208 target_os = "redox"
1209 )))]
1210 pub(super) fn get_thread_time(vm: &VirtualMachine) -> PyResult<Duration> {
1211 get_clock_time(ClockId::CLOCK_THREAD_CPUTIME_ID, vm)
1212 }
1213
1214 #[cfg(target_os = "solaris")]
1215 pub(super) fn get_thread_time(vm: &VirtualMachine) -> PyResult<Duration> {
1216 let _ = vm;
1217 Ok(host_time::gethrvtime_duration())
1218 }
1219
1220 #[cfg(not(any(
1221 target_os = "illumos",
1222 target_os = "netbsd",
1223 target_os = "solaris",
1224 target_os = "openbsd",
1225 )))]
1226 pub(super) fn get_process_time(vm: &VirtualMachine) -> PyResult<Duration> {
1227 get_clock_time(ClockId::CLOCK_PROCESS_CPUTIME_ID, vm)
1228 }
1229
1230 #[cfg(any(
1231 target_os = "illumos",
1232 target_os = "netbsd",
1233 target_os = "solaris",
1234 target_os = "openbsd",
1235 ))]
1236 pub(super) fn get_process_time(vm: &VirtualMachine) -> PyResult<Duration> {
1237 let ru = host_resource::getrusage(host_resource::RUSAGE_SELF)
1238 .map_err(|e| e.into_pyexception(vm))?;
1239 ru.total_cpu_duration()
1240 .ok_or_else(|| vm.new_overflow_error("timestamp too large to convert to i64"))
1241 }
1242}
1243
1244#[cfg(windows)]
1245#[pymodule(sub, name = "time")]
1246mod platform {
1247 use super::decl::{MS_TO_NS, SEC_TO_NS, StructTimeData, get_tz_info, time_muldiv};
1248 use crate::{
1249 PyRef, PyResult, VirtualMachine,
1250 builtins::{PyNamespace, PyUtf8StrRef},
1251 };
1252 use core::time::Duration;
1253 use rustpython_host_env::time as host_time;
1254
1255 fn struct_time_from_tm(
1256 vm: &VirtualMachine,
1257 tm: libc::tm,
1258 zone: &str,
1259 gmtoff: i32,
1260 ) -> StructTimeData {
1261 StructTimeData {
1262 tm_year: vm.ctx.new_int(tm.tm_year + 1900).into(),
1263 tm_mon: vm.ctx.new_int(tm.tm_mon + 1).into(),
1264 tm_mday: vm.ctx.new_int(tm.tm_mday).into(),
1265 tm_hour: vm.ctx.new_int(tm.tm_hour).into(),
1266 tm_min: vm.ctx.new_int(tm.tm_min).into(),
1267 tm_sec: vm.ctx.new_int(tm.tm_sec).into(),
1268 tm_wday: vm.ctx.new_int((tm.tm_wday + 6) % 7).into(),
1269 tm_yday: vm.ctx.new_int(tm.tm_yday + 1).into(),
1270 tm_isdst: vm.ctx.new_int(tm.tm_isdst).into(),
1271 tm_zone: vm.ctx.new_str(zone).into(),
1272 tm_gmtoff: vm.ctx.new_int(gmtoff).into(),
1273 }
1274 }
1275
1276 pub(super) fn current_time_t() -> host_time::TimeT {
1277 host_time::current_time_t()
1278 }
1279
1280 pub(super) fn gmtime_from_timestamp(
1281 when: host_time::TimeT,
1282 vm: &VirtualMachine,
1283 ) -> PyResult<StructTimeData> {
1284 let tm = host_time::gmtime_from_timestamp(when)
1285 .ok_or_else(|| vm.new_overflow_error("timestamp out of range for platform time_t"))?;
1286 Ok(struct_time_from_tm(vm, tm, "UTC", 0))
1287 }
1288
1289 pub(super) fn localtime_from_timestamp(
1290 when: host_time::TimeT,
1291 vm: &VirtualMachine,
1292 ) -> PyResult<StructTimeData> {
1293 let tm = host_time::localtime_from_timestamp(when)
1294 .ok_or_else(|| vm.new_overflow_error("timestamp out of range for platform time_t"))?;
1295
1296 let info = get_tz_info();
1298 let (bias, name) = if tm.tm_isdst > 0 {
1299 (info.daylight_bias, &info.daylight_name)
1300 } else {
1301 (info.standard_bias, &info.standard_name)
1302 };
1303
1304 let gmtoff = -(info.bias + bias) * 60;
1305
1306 Ok(struct_time_from_tm(vm, tm, name, gmtoff))
1307 }
1308
1309 pub(super) fn win_mktime(t: &StructTimeData, vm: &VirtualMachine) -> PyResult<f64> {
1310 let mut tm = super::decl::tm_from_struct_time(t, vm)?;
1311 let timestamp = host_time::mktime(&mut tm);
1312 if timestamp == -1 && tm.tm_wday == -1 {
1313 return Err(vm.new_overflow_error("mktime argument out of range"));
1314 }
1315 Ok(timestamp as f64)
1316 }
1317
1318 fn win_perf_counter_frequency(vm: &VirtualMachine) -> PyResult<i64> {
1319 let frequency =
1320 host_time::query_performance_frequency().ok_or_else(|| vm.new_last_os_error())?;
1321
1322 if frequency < 1 {
1323 Err(vm.new_runtime_error("invalid QueryPerformanceFrequency"))
1324 } else if frequency > i64::MAX / SEC_TO_NS {
1325 Err(vm.new_overflow_error("QueryPerformanceFrequency is too large"))
1326 } else {
1327 Ok(frequency)
1328 }
1329 }
1330
1331 fn global_frequency(vm: &VirtualMachine) -> PyResult<i64> {
1332 rustpython_common::static_cell! {
1333 static FREQUENCY: PyResult<i64>;
1334 };
1335 FREQUENCY
1336 .get_or_init(|| win_perf_counter_frequency(vm))
1337 .clone()
1338 }
1339
1340 pub(super) fn get_perf_time(vm: &VirtualMachine) -> PyResult<Duration> {
1341 let ticks = host_time::query_performance_counter();
1342
1343 Ok(Duration::from_nanos(time_muldiv(
1344 ticks,
1345 SEC_TO_NS,
1346 global_frequency(vm)?,
1347 )))
1348 }
1349
1350 fn get_system_time_adjustment(vm: &VirtualMachine) -> PyResult<u32> {
1351 host_time::get_system_time_adjustment().ok_or_else(|| vm.new_last_os_error())
1352 }
1353
1354 pub(super) fn get_monotonic_time(vm: &VirtualMachine) -> PyResult<Duration> {
1355 let ticks = host_time::tick_count64();
1356
1357 Ok(Duration::from_nanos(
1358 (ticks as i64)
1359 .checked_mul(MS_TO_NS)
1360 .ok_or_else(|| vm.new_overflow_error("timestamp too large to convert to i64"))?
1361 as u64,
1362 ))
1363 }
1364
1365 #[pyfunction]
1366 fn get_clock_info(name: PyUtf8StrRef, vm: &VirtualMachine) -> PyResult<PyRef<PyNamespace>> {
1367 let (adj, imp, mono, res) = match name.as_str() {
1368 "monotonic" => (
1369 false,
1370 "GetTickCount64()",
1371 true,
1372 get_system_time_adjustment(vm)? as f64 * 1e-7,
1373 ),
1374 "perf_counter" => (
1375 false,
1376 "QueryPerformanceCounter()",
1377 true,
1378 1.0 / (global_frequency(vm)? as f64),
1379 ),
1380 "process_time" => (false, "GetProcessTimes()", true, 1e-7),
1381 "thread_time" => (false, "GetThreadTimes()", true, 1e-7),
1382 "time" => (
1383 true,
1384 "GetSystemTimeAsFileTime()",
1385 false,
1386 get_system_time_adjustment(vm)? as f64 * 1e-7,
1387 ),
1388 _ => return Err(vm.new_value_error("unknown clock")),
1389 };
1390
1391 Ok(py_namespace!(vm, {
1392 "implementation" => vm.new_pyobj(imp),
1393 "monotonic" => vm.ctx.new_bool(mono),
1394 "adjustable" => vm.ctx.new_bool(adj),
1395 "resolution" => vm.ctx.new_float(res),
1396 }))
1397 }
1398
1399 pub(super) fn get_thread_time(vm: &VirtualMachine) -> PyResult<Duration> {
1400 let total = host_time::get_thread_time_100ns()
1401 .ok_or_else(|| vm.new_os_error("Failed to get clock time"))?;
1402 Ok(Duration::from_nanos(total * 100))
1403 }
1404
1405 pub(super) fn get_process_time(vm: &VirtualMachine) -> PyResult<Duration> {
1406 let total = host_time::get_process_time_100ns()
1407 .ok_or_else(|| vm.new_os_error("Failed to get clock time"))?;
1408 Ok(Duration::from_nanos(total * 100))
1409 }
1410}