Skip to content

Commit db79ce8

Browse files
authored
Fix time series preprocess (#4339)
* fix time series preprocess * support http url, file url * minor * remove pickle for safety, add some safe check * remove pandas * should not use time_series_url due to chat template * time_series_url is fine
1 parent 4432b5d commit db79ce8

6 files changed

Lines changed: 252 additions & 13 deletions

File tree

lmdeploy/serve/processors/multimodal.py

Lines changed: 31 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,7 @@ def merge_message_content(msg: Dict) -> Dict:
8787
@staticmethod
8888
async def async_convert_multimodal_data(messages: List[Dict]) -> List[Dict]:
8989
"""Convert user-input multimodal data into GPT4V message format."""
90+
from lmdeploy.vl.time_series_utils import load_time_series
9091
from lmdeploy.vl.utils import load_image
9192

9293
if isinstance(messages, Dict):
@@ -164,7 +165,34 @@ def _inner_call(i, in_messages, out_messages):
164165
message['content'].append(data)
165166
except KeyError:
166167
logger.error(f'invalid format {message}')
167-
elif item['type'] in ['text', 'time_series']:
168+
elif item['type'] == 'time_series_url':
169+
"""
170+
convert the following item:
171+
{
172+
'type': 'time_series_url',
173+
'time_series_url': {
174+
'url': 'time series url or base64-encoded time series data',
175+
'key': 'value' # parameters used in time series processing
176+
...
177+
}
178+
}
179+
to:
180+
{
181+
'type': 'time_series',
182+
'time_series': np.ndarray,
183+
'key': 'value' # parameters used in time series processing
184+
...
185+
}
186+
""" # noqa
187+
data = item['time_series_url'].copy()
188+
try:
189+
url = data.pop('url')
190+
time_series = load_time_series(url)
191+
data.update(type='time_series', time_series=time_series)
192+
message['content'].append(data)
193+
except KeyError:
194+
logger.error(f'invalid format {message}')
195+
elif item['type'] in ['text']:
168196
message['content'].append(item)
169197
else:
170198
logger.error(f'unexpected content type {message}')
@@ -324,10 +352,10 @@ def _re_format_prompt_images_pair(prompt: Tuple) -> Dict:
324352

325353
def _has_multimodal_input(self, messages: List[Dict]) -> bool:
326354
"""Check if messages contain multimodal input (images)."""
355+
multimodal_types = ['image_url', 'image_data', 'time_series_url']
327356
return any(
328357
isinstance(message.get('content'), list) and any(
329-
item.get('type') in ['image_url', 'image_data', 'time_series'] for item in message['content'])
330-
for message in messages)
358+
item.get('type') in multimodal_types for item in message['content']) for message in messages)
331359

332360
async def _get_text_prompt_input(self,
333361
prompt: str | List[Dict],

lmdeploy/vl/__init__.py

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
# Copyright (c) OpenMMLab. All rights reserved.
2+
from .time_series_utils import load_time_series
23
from .utils import load_image
34

4-
__all__ = ['load_image']
5+
__all__ = ['load_image', 'load_time_series']

lmdeploy/vl/model/base.py

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,26 @@ def collect_images(messages):
185185
}) for x in content if x['type'] == 'image'])
186186
return images
187187

188+
@staticmethod
189+
def collect_time_series(messages):
190+
"""Gather all time series data along with their respective parameters
191+
from the messages and compile them into a single list.
192+
193+
Args:
194+
messages (List[Tuple[np.ndarray, Dict]]): a list of time
195+
series data with their corresponding parameters
196+
""" # noqa
197+
time_series = []
198+
for message in messages:
199+
content = message['content']
200+
if not isinstance(content, List):
201+
continue
202+
time_series.extend([(x['time_series'], {
203+
k: v
204+
for k, v in x.items() if k not in {'type', 'time_series'}
205+
}) for x in content if x['type'] == 'time_series'])
206+
return time_series
207+
188208
@staticmethod
189209
def IMAGE_TOKEN_included(messages):
190210
"""Check whether the IMAGE_TOKEN is included in the messages.

lmdeploy/vl/model/interns1_pro.py

Lines changed: 38 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -82,19 +82,49 @@ def check_time_series_input(self, messages):
8282
for message in messages)
8383
self.has_time_series_input = has_time_series_input
8484

85+
def time_series_processor(self, ts_input, sr):
86+
if not isinstance(ts_input, np.ndarray):
87+
ts_input = np.array(ts_input, dtype=np.float32)
88+
89+
mean = ts_input.mean(axis=0, keepdims=True)
90+
std = ts_input.std(axis=0, keepdims=True)
91+
ts_input = (ts_input - mean) / (std + 1e-8)
92+
93+
# truncate to 240k to avoid OOM
94+
max_ts_len = 240000
95+
if len(ts_input) > max_ts_len:
96+
ts_input = ts_input[:max_ts_len]
97+
98+
if ts_input.ndim == 1:
99+
ts_input = ts_input[:, None] # [T,C]
100+
101+
ts_len = ts_input.shape[0]
102+
103+
# set the default value to ts_len / 4 if sr is not provided or invalid
104+
if sr is None or sr <= 0:
105+
sr = max(ts_len / 4, 1.0)
106+
107+
# compute num ts tokens
108+
stride = np.floor(160 / ((1 + np.exp(-sr / 100))**6))
109+
patch_size = stride * 2
110+
embed_length = (np.ceil((ts_len - patch_size) / stride) + 1)
111+
num_ts_tokens = int((embed_length // 2 + 1) // 2)
112+
113+
return dict(ts_values=[ts_input], ts_sr=[sr], ts_lens=[ts_len], num_ts_tokens=[num_ts_tokens])
114+
85115
def preprocess(self, messages: List[Dict], mm_processor_kwargs: Optional[Dict[str, Any]] = None) -> List[Dict]:
86116
"""Refer to `super().preprocess()` for spec."""
87117

88118
self.check_time_series_input(messages)
89119

90120
if self.has_time_series_input:
91-
time_series_inputs = self.processor.time_series_preprocessor(messages)
92-
time_series_inputs = self.processor.time_series_processor(
93-
ts_paths=time_series_inputs['time_series_paths'],
94-
sampling_rates=time_series_inputs['time_series_sampling_rates'])
95-
time_series_inputs.update({'ts_token_id': self.ts_token_id})
96-
outputs = [time_series_inputs]
97-
121+
time_series = self.collect_time_series(messages)
122+
outputs = []
123+
for ts_input, params in time_series:
124+
sr = params.get('sampling_rate') if params is not None else None
125+
time_series_inputs = self.time_series_processor(ts_input, sr)
126+
time_series_inputs.update({'ts_token_id': self.ts_token_id})
127+
outputs.append(time_series_inputs)
98128
else:
99129
min_pixels, max_pixels = self.get_processor_args(mm_processor_kwargs)
100130

@@ -181,7 +211,7 @@ def ts_to_pytorch_aux(self, messages, prompt, TS_TOKEN, tokenizer, sequence_star
181211
ts_tokens = preps[i - 1]['num_ts_tokens']
182212

183213
# NOTE: zhouxinyu, better to be valid type in the processor
184-
ts_tokens = int(ts_tokens[0])
214+
ts_tokens = ts_tokens[0]
185215
ts_array = np.array(preps[i - 1]['ts_values'])
186216

187217
preps[i - 1].update(num_ts_tokens=ts_tokens)

lmdeploy/vl/time_series_utils.py

Lines changed: 160 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,160 @@
1+
# Copyright (c) OpenMMLab. All rights reserved.
2+
import csv
3+
import os
4+
from io import BytesIO, StringIO
5+
6+
import numpy as np
7+
import pybase64
8+
import requests
9+
10+
from lmdeploy.utils import get_logger
11+
12+
logger = get_logger('lmdeploy')
13+
14+
FETCH_TIMEOUT = int(os.environ.get('LMDEPLOY_FETCH_TIMEOUT', 10))
15+
HEADERS = {
16+
'User-Agent':
17+
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
18+
'(KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3'
19+
}
20+
21+
22+
def encode_time_series_base64(data: str | np.ndarray) -> str:
23+
"""Encode time series data to base64.
24+
25+
Supports: HTTP URL, local path, or numpy array.
26+
"""
27+
buffered = BytesIO()
28+
29+
try:
30+
if isinstance(data, str):
31+
if data.startswith('http'):
32+
response = requests.get(data, headers=HEADERS, timeout=FETCH_TIMEOUT)
33+
response.raise_for_status()
34+
ts_array = _load_bytes(response.content, data)
35+
elif data.startswith('file://'):
36+
path = data.removeprefix('file://')
37+
ts_array = _load_path(path)
38+
elif os.path.exists(data):
39+
ts_array = _load_path(data)
40+
else:
41+
raise ValueError(f'Path does not exist: {data}')
42+
elif isinstance(data, np.ndarray):
43+
ts_array = data
44+
else:
45+
raise TypeError(f'Expected str or np.ndarray, got {type(data)}')
46+
47+
np.save(buffered, ts_array)
48+
49+
except Exception as error:
50+
data_info = str(data)[:100] + ' ...' if isinstance(data, str) and len(data) > 100 else str(data)
51+
logger.error(f'{error}, data={data_info}')
52+
np.save(buffered, np.zeros((6000, 3), dtype=np.float32)) # dummy
53+
54+
return pybase64.b64encode(buffered.getvalue()).decode('utf-8')
55+
56+
57+
def load_time_series_from_base64(ts_base64: bytes | str) -> np.ndarray:
58+
"""Load time series from base64 format."""
59+
if isinstance(ts_base64, str):
60+
ts_base64 = ts_base64.encode('utf-8')
61+
return np.load(BytesIO(pybase64.b64decode(ts_base64)), allow_pickle=False)
62+
63+
64+
def load_time_series(data_source: str | np.ndarray) -> np.ndarray:
65+
"""Load time series from URL, local path, base64 data URL, or numpy
66+
array."""
67+
try:
68+
if isinstance(data_source, np.ndarray):
69+
return data_source
70+
71+
if data_source.startswith('http'):
72+
response = requests.get(data_source, headers=HEADERS, timeout=FETCH_TIMEOUT)
73+
response.raise_for_status()
74+
return _load_bytes(response.content, data_source)
75+
76+
if data_source.startswith('data:time_series'):
77+
return load_time_series_from_base64(data_source.split(',')[1])
78+
79+
if data_source.startswith('file://'):
80+
path = data_source.removeprefix('file://')
81+
return _load_path(path)
82+
83+
if os.path.exists(data_source):
84+
return _load_path(data_source)
85+
86+
raise ValueError(f'Invalid data source: {data_source}')
87+
except Exception as error:
88+
data_info = str(data_source)[:100] + ' ...' if isinstance(data_source,
89+
str) and len(data_source) > 100 else str(data_source)
90+
logger.error(f'{error}, data_source={data_info}')
91+
return np.zeros((6000, 3), dtype=np.float32) # dummy
92+
93+
94+
def _load_bytes(content: bytes, hint: str = '') -> np.ndarray:
95+
"""Auto-detect format from bytes.
96+
97+
Try: npy -> csv -> audio.
98+
"""
99+
hint = hint.lower()
100+
101+
# Format hints from URL/path
102+
if '.npy' in hint:
103+
return np.load(BytesIO(content))
104+
if '.csv' in hint:
105+
return _load_csv(content)
106+
if any(ext in hint for ext in ['.wav', '.mp3', '.flac']):
107+
return _load_audio(content)
108+
109+
# Fallback: try all formats
110+
loaders = [lambda: np.load(BytesIO(content)), lambda: _load_csv(content), lambda: _load_audio(content)]
111+
for loader in loaders:
112+
try:
113+
return loader()
114+
except Exception:
115+
continue
116+
raise ValueError(f'Cannot detect format from bytes: {hint[:50]}')
117+
118+
119+
def _load_path(path: str) -> np.ndarray:
120+
"""Load from local file path based on extension."""
121+
ext = os.path.splitext(path)[-1].lower()
122+
123+
if ext == '.npy':
124+
return np.load(path)
125+
if ext == '.csv':
126+
return _load_csv(path)
127+
if ext in ['.wav', '.mp3', '.flac']:
128+
return _load_audio(path)
129+
130+
raise ValueError(f'Unsupported format: {ext}')
131+
132+
133+
def _load_csv(source: bytes | str) -> np.ndarray:
134+
"""Load CSV from bytes or file path."""
135+
# Read content as text
136+
if isinstance(source, bytes):
137+
text = source.decode('utf-8')
138+
else:
139+
with open(source, 'r', newline='') as f:
140+
text = f.read()
141+
142+
# Parse CSV
143+
f = StringIO(text)
144+
reader = csv.reader(f)
145+
rows = list(reader)
146+
147+
return np.array(rows, dtype=np.float32)
148+
149+
150+
def _load_audio(source: bytes | str) -> np.ndarray:
151+
"""Load audio from bytes or file path."""
152+
try:
153+
import soundfile as sf
154+
except ImportError:
155+
raise ImportError('Please install soundfile to process audio files.')
156+
157+
if isinstance(source, bytes):
158+
source = BytesIO(source)
159+
ts, sr = sf.read(source)
160+
return ts

lmdeploy/vl/utils.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@
1313

1414

1515
def encode_image_base64(image: Union[str, Image.Image]) -> str:
16-
"""Encode raw date to base64 format."""
16+
"""Encode raw data to base64 format."""
1717
buffered = BytesIO()
1818
FETCH_TIMEOUT = int(os.environ.get('LMDEPLOY_FETCH_TIMEOUT', 10))
1919
headers = {

0 commit comments

Comments
 (0)