File: datetime_signals.py

package info (click to toggle)
python-sigima 1.1.1-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 25,608 kB
  • sloc: python: 35,251; makefile: 3
file content (240 lines) | stat: -rw-r--r-- 8,479 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
"""
DateTime Signal Processing
==========================

This example demonstrates how to work with signals that have datetime X-axis
data in Sigima. DateTime support is essential for time-series analysis where
you need to preserve human-readable timestamps while performing signal processing.

The example shows:

* Creating signals from datetime objects and strings
* Using different time units (hours, minutes, seconds, milliseconds, etc.)
* Visualizing datetime signals with proper time formatting
* CSV I/O with datetime preservation
* Roundtrip testing across all supported time units

This tutorial uses PlotPy for visualization, providing interactive plots
with properly formatted datetime axes.
"""

# %%
# Importing necessary modules
# ---------------------------
# We'll start by importing all the required modules for datetime signal processing
# and visualization.

import tempfile
from datetime import datetime, timedelta
from pathlib import Path

import numpy as np

from sigima.io.signal.formats import CSVSignalFormat
from sigima.objects import SignalObj, create_signal
from sigima.objects.signal.constants import TIME_UNIT_FACTORS, VALID_TIME_UNITS
from sigima.viz import view_curves

# %%
# Creating datetime signals from datetime objects
# -----------------------------------------------
# The most common way to create datetime signals is from Python datetime objects.
# This is useful when you have timestamps from sensors, logs, or other time-series data.

# Create a temperature monitoring signal with second-resolution timestamps
base_time = datetime(2025, 10, 7, 10, 0, 0)
timestamps = [base_time + timedelta(seconds=i * 10) for i in range(100)]

# Simulate temperature data with daily variation and noise
hours_elapsed = np.arange(100) * 10 / 3600  # Convert to hours
temperature = (
    20 + 5 * np.sin(2 * np.pi * hours_elapsed / 24) + np.random.randn(100) * 0.5
)

# Create the signal with datetime X-axis
temp_signal = create_signal("Temperature Monitor")
temp_signal.set_x_from_datetime(timestamps, unit="s", format_str="%H:%M:%S")
temp_signal.y = temperature
temp_signal.ylabel = "Temperature"
temp_signal.yunit = "°C"

print("✓ Temperature signal created successfully!")
print(f"Is datetime signal: {temp_signal.is_x_datetime()}")
print(f"Time unit: {temp_signal.xunit}")
print(f"First timestamp: {temp_signal.get_x_as_datetime()[0]}")
print(f"Last timestamp: {temp_signal.get_x_as_datetime()[-1]}")

# Visualize the temperature signal
view_curves(
    temp_signal,
    title="Temperature Monitoring Over Time",
    object_name="datetime_temperature",
)

# %%
# Creating datetime signals from string timestamps
# ------------------------------------------------
# Often, datetime data comes as strings (e.g., from CSV files or logs).
# Sigima can automatically parse common datetime string formats.

# Pressure monitoring with minute-resolution string timestamps
date_strings = [
    "2025-10-07 10:00:00",
    "2025-10-07 10:05:00",
    "2025-10-07 10:10:00",
    "2025-10-07 10:15:00",
    "2025-10-07 10:20:00",
    "2025-10-07 10:25:00",
    "2025-10-07 10:30:00",
]

# Simulate atmospheric pressure readings
pressure = np.array([1013.25, 1013.20, 1013.15, 1013.10, 1013.18, 1013.22, 1013.25])

pressure_signal = create_signal("Atmospheric Pressure")
pressure_signal.set_x_from_datetime(date_strings, unit="min", format_str="%H:%M:%S")
pressure_signal.y = pressure
pressure_signal.ylabel = "Pressure"
pressure_signal.yunit = "hPa"

print("\n✓ Pressure signal created from strings!")
print(f"Number of readings: {len(pressure_signal.y)}")
print(f"Pressure range: {pressure.min():.2f} - {pressure.max():.2f} hPa")

# Visualize the pressure signal
view_curves(
    pressure_signal,
    title="Atmospheric Pressure Readings",
    object_name="datetime_pressure",
)

# %%
# Using different time units
# --------------------------
# Sigima supports various time units to match your data's natural resolution:
# nanoseconds (ns), microseconds (us), milliseconds (ms), seconds (s),
# minutes (min), and hours (h).

print(f"\n✓ Supported time units: {', '.join(VALID_TIME_UNITS)}")
print("\nConversion factors to seconds:")
for unit, factor in TIME_UNIT_FACTORS.items():
    print(f"   1 {unit:>3s} = {factor:>10.1e} seconds")

# Create hourly temperature cycle (24 hours)
hourly_base = datetime(2025, 10, 7, 0, 0, 0)
hourly_times = [hourly_base + timedelta(hours=i) for i in range(24)]
daily_temp = 15 + 8 * np.sin(2 * np.pi * (np.arange(24) - 6) / 24)

hourly_signal = SignalObj()
hourly_signal.set_x_from_datetime(hourly_times, unit="h", format_str="%H:%M")
hourly_signal.y = daily_temp
hourly_signal.title = "Daily Temperature Cycle"
hourly_signal.ylabel = "Temperature"
hourly_signal.yunit = "°C"

print("\n✓ Hourly signal created!")
print(f"   Time unit: {hourly_signal.xunit}")
print(f"   X-axis spacing: {np.diff(hourly_signal.x)[0]:.1f} hours")
print(f"   Temperature range: {daily_temp.min():.1f}°C to {daily_temp.max():.1f}°C")

# Visualize the daily cycle
view_curves(
    hourly_signal,
    title="24-Hour Temperature Cycle",
    object_name="datetime_hourly",
)

# %%
# Comparing multiple datetime signals
# -----------------------------------
# You can visualize multiple datetime signals together to compare trends.

# Create three signals at different time scales
minute_base = datetime(2025, 10, 7, 14, 0, 0)
minute_times = [minute_base + timedelta(minutes=i) for i in range(60)]

# Heart rate over 1 hour
heart_rate = 70 + 10 * np.sin(2 * np.pi * np.arange(60) / 60) + 2 * np.random.randn(60)
hr_signal = SignalObj()
hr_signal.set_x_from_datetime(minute_times, unit="min")
hr_signal.y = heart_rate
hr_signal.title = "Heart Rate"
hr_signal.ylabel = "Heart Rate"
hr_signal.yunit = "bpm"

# Steps per minute
steps = 80 + 20 * np.sin(2 * np.pi * np.arange(60) / 30) + 5 * np.random.randn(60)
steps = np.clip(steps, 0, None)  # No negative steps
steps_signal = SignalObj()
steps_signal.set_x_from_datetime(minute_times, unit="min")
steps_signal.y = steps
steps_signal.title = "Steps Per Minute"
steps_signal.ylabel = "Steps"
steps_signal.yunit = "steps/min"

print("\n✓ Created multiple signals for comparison!")
print(f"   Heart rate avg: {heart_rate.mean():.1f} bpm")
print(f"   Steps avg: {steps.mean():.1f} steps/min")

# Visualize multiple signals together
view_curves(
    [hr_signal, steps_signal],
    title="Exercise Monitoring - Multiple Signals",
    object_name="datetime_multi",
)

# %%
# CSV I/O with datetime preservation
# ----------------------------------
# Sigima automatically preserves datetime information when writing to CSV,
# storing timestamps as human-readable strings instead of opaque float values.

# Write temperature signal to CSV (using temporary directory)
temp_dir = tempfile.mkdtemp(prefix="sigima_")
csv_file = Path(temp_dir) / "temp_data.csv"
fmt = CSVSignalFormat()
fmt.write(str(csv_file), temp_signal)

print(f"\n✓ Saved datetime signal to CSV: {csv_file}")
print("   CSV Preview (first 5 lines):")
with open(csv_file, "r", encoding="utf-8") as f:
    for i, line in enumerate(f):
        if i < 5:
            print(f"      {line.rstrip()}")
        else:
            break

# Read it back and verify
loaded_signals = fmt.read(str(csv_file))
loaded_signal = loaded_signals[0]

# Verify datetime preservation
dt_original = temp_signal.get_x_as_datetime()
dt_loaded = loaded_signal.get_x_as_datetime()
data_matches = np.allclose(loaded_signal.y, temp_signal.y)
time_matches = np.all(dt_original == dt_loaded)

print("\n✓ Loaded signal from CSV successfully!")
print(f"   Is datetime: {loaded_signal.is_x_datetime()}")
print(f"   Data preserved: {data_matches}")
print(f"   Timestamps preserved: {time_matches}")


# %%
# Summary
# -------
# This example demonstrated the complete datetime signal workflow in Sigima:
#
# 1. **Creating datetime signals** from objects and strings
# 2. **Multiple time units** (ns, us, ms, s, min, h) stored in `xunit` attribute
# 3. **Visualization** with properly formatted datetime axes
# 4. **CSV I/O** with automatic datetime preservation
#
# The time unit is stored in the signal's `xunit` attribute, making it easy
# to access and consistent with other signal metadata. DateTime support makes
# Sigima ideal for time-series analysis, sensor data processing, and any
# application requiring human-readable temporal information.

print("\n" + "=" * 70)
print("✓ DateTime Signal Processing Example Completed Successfully!")
print("=" * 70)