Page Menu
Home
c4science
Search
Configure Global Search
Log In
Files
F77984278
timeseries.py
No One
Temporary
Actions
Download File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Subscribers
None
File Metadata
Details
File Info
Storage
Attached
Created
Sat, Aug 17, 16:18
Size
6 KB
Mime Type
text/x-python
Expires
Mon, Aug 19, 16:18 (2 d)
Engine
blob
Format
Raw Data
Handle
19958934
Attached To
R4670 PySONIC (old)
timeseries.py
View Options
# -*- coding: utf-8 -*-
# @Author: Theo Lemaire
# @Email: theo.lemaire@epfl.ch
# @Date: 2021-05-15 11:01:04
# @Last Modified by: Theo Lemaire
# @Last Modified time: 2021-05-15 11:40:26
import
pandas
as
pd
import
numpy
as
np
from
scipy.interpolate
import
interp1d
from
..utils
import
cycleAvg
class
TimeSeries
(
pd
.
DataFrame
):
''' Wrapper around pandas DataFrame to store timeseries data. '''
time_key
=
't'
stim_key
=
'stimstate'
def
__init__
(
self
,
t
,
stim
,
dout
):
super
()
.
__init__
(
data
=
{
self
.
time_key
:
t
,
self
.
stim_key
:
stim
,
**
dout
})
@property
def
time
(
self
):
return
self
[
self
.
time_key
]
.
values
@property
def
tbounds
(
self
):
return
self
.
time
.
min
(),
self
.
time
.
max
()
@property
def
stim
(
self
):
return
self
[
self
.
stim_key
]
.
values
@property
def
inputs
(
self
):
return
[
self
.
time_key
,
self
.
stim_key
]
@property
def
outputs
(
self
):
return
list
(
set
(
self
.
columns
.
values
)
-
set
(
self
.
inputs
))
def
addColumn
(
self
,
key
,
arr
,
preceding_key
=
None
):
''' Add a new column to the timeseries dataframe, right after a specific column. '''
self
[
key
]
=
arr
if
preceding_key
is
not
None
:
cols
=
self
.
columns
.
tolist
()[:
-
1
]
preceding_index
=
cols
.
index
(
preceding_key
)
new_cols
=
cols
[:
preceding_index
+
1
]
+
[
key
]
+
cols
[
preceding_index
+
1
:]
self
.
reindex
(
columns
=
new_cols
)
# self = self[cols[:preceding_index + 1] + [key] + cols[preceding_index + 1:]]
def
interpCol
(
self
,
t
,
k
,
kind
):
''' Interpolate a column according to a new time vector. '''
kind
=
'nearest'
if
k
==
self
.
stim_key
else
'linear'
self
[
k
]
=
interp1d
(
self
.
time
,
self
[
k
]
.
values
,
kind
=
kind
)(
t
)
def
interp1d
(
self
,
t
):
''' Interpolate the entire dataframe according to a new time vector. '''
for
k
in
self
.
outputs
:
self
.
interpCol
(
t
,
k
,
'linear'
)
self
.
interpCol
(
t
,
self
.
stim_key
,
'nearest'
)
self
[
self
.
time_key
]
=
t
def
resample
(
self
,
dt
):
''' Resample dataframe at regular time step. '''
tmin
,
tmax
=
self
.
tbounds
n
=
int
((
tmax
-
tmin
)
/
dt
)
+
1
self
.
interp1d
(
np
.
linspace
(
tmin
,
tmax
,
n
))
def
cycleAveraged
(
self
,
T
):
''' Cycle-average a periodic solution. '''
t
=
np
.
arange
(
self
.
time
[
0
],
self
.
time
[
-
1
],
T
)
stim
=
interp1d
(
self
.
time
,
self
.
stim
,
kind
=
'nearest'
)(
t
)
outputs
=
{
k
:
cycleAvg
(
self
.
time
,
self
[
k
]
.
values
,
T
)
for
k
in
self
.
outputs
}
return
self
.
__class__
(
t
,
stim
,
outputs
)
def
prepend
(
self
,
t0
=
0
):
''' Repeat first row outputs for a preceding time. '''
if
t0
>
self
.
time
.
min
():
raise
ValueError
(
't0 greater than minimal time value'
)
self
.
loc
[
-
1
]
=
self
.
iloc
[
0
]
# repeat first row
self
.
index
=
self
.
index
+
1
# shift index
self
.
sort_index
(
inplace
=
True
)
self
[
self
.
time_key
][
0
]
=
t0
self
[
self
.
stim_key
][
0
]
=
0
def
bound
(
self
,
tbounds
):
''' Restrict all columns of dataframe to indexes corresponding to time values
within specific bounds. '''
tmin
,
tmax
=
tbounds
return
self
[
np
.
logical_and
(
self
.
time
>=
tmin
,
self
.
time
<=
tmax
)]
.
reset_index
(
drop
=
True
)
def
checkAgainst
(
self
,
other
):
assert
isinstance
(
other
,
self
.
__class__
),
'classes do not match'
assert
all
(
self
.
keys
()
==
other
.
keys
()),
'differing keys'
for
k
in
self
.
inputs
:
assert
all
(
self
[
k
]
.
values
==
other
[
k
]
.
values
),
f
'{k} vectors do not match'
def
operate
(
self
,
other
,
op
):
''' Generic arithmetic operator. '''
self
.
checkAgainst
(
other
)
return
self
.
__class__
(
self
.
time
,
self
.
stim
,
{
k
:
getattr
(
self
[
k
]
.
values
,
op
)(
other
[
k
]
.
values
)
for
k
in
self
.
outputs
}
)
def
__add__
(
self
,
other
):
''' Addition operator. '''
return
self
.
operate
(
other
,
'__add__'
)
def
__sub__
(
self
,
other
):
''' Subtraction operator. '''
return
self
.
operate
(
other
,
'__sub__'
)
def
__mul__
(
self
,
other
):
''' Multiplication operator. '''
return
self
.
operate
(
other
,
'__mul__'
)
def
__truediv__
(
self
,
other
):
''' Division operator. '''
return
self
.
operate
(
other
,
'__truediv__'
)
class
SpatiallyExtendedTimeSeries
:
def
__init__
(
self
,
data
):
self
.
data
=
data
def
__iter__
(
self
):
raise
ValueError
(
f
'{self.__class__.__name__} is not iterable'
)
def
keys
(
self
):
return
self
.
data
.
keys
()
def
values
(
self
):
return
self
.
data
.
values
()
def
items
(
self
):
return
self
.
data
.
items
()
def
__getitem__
(
self
,
key
):
return
self
.
data
[
key
]
def
__delitem__
(
self
,
key
):
del
self
.
data
[
key
]
def
__setitem__
(
self
,
key
,
value
):
self
.
data
[
key
]
=
value
def
checkAgainst
(
self
,
other
):
assert
isinstance
(
other
,
self
.
__class__
),
'differing classes'
assert
self
.
keys
()
==
other
.
keys
(),
'differing keys'
for
k
in
self
.
keys
():
self
.
data
[
k
]
.
checkAgainst
(
other
.
data
[
k
])
def
operate
(
self
,
other
,
op
):
self
.
checkAgainst
(
other
)
return
self
.
__class__
({
k
:
getattr
(
self
.
data
[
k
],
op
)(
other
.
data
[
k
])
for
k
in
self
.
keys
()})
def
__add__
(
self
,
other
):
''' Addition operator. '''
return
self
.
operate
(
other
,
'__add__'
)
def
__sub__
(
self
,
other
):
''' Subtraction operator. '''
return
self
.
operate
(
other
,
'__sub__'
)
def
__mul__
(
self
,
other
):
''' Multiplication operator. '''
return
self
.
operate
(
other
,
'__mul__'
)
def
__truediv__
(
self
,
other
):
''' Division operator. '''
return
self
.
operate
(
other
,
'__truediv__'
)
def
cycleAveraged
(
self
,
*
args
,
**
kwargs
):
return
self
.
__class__
({
k
:
v
.
cycleAveraged
(
*
args
,
**
kwargs
)
for
k
,
v
in
self
.
items
()})
def
prepend
(
self
,
*
args
,
**
kwargs
):
for
k
in
self
.
keys
():
self
.
data
[
k
]
.
prepend
(
*
args
,
**
kwargs
)
def
getArray
(
self
,
varkey
,
prefix
=
None
):
section_keys
=
list
(
self
.
keys
())
if
prefix
is
not
None
:
section_keys
=
list
(
filter
(
lambda
x
:
x
.
startswith
(
prefix
),
section_keys
))
return
np
.
array
([
self
[
k
][
varkey
]
.
values
for
k
in
section_keys
])
@property
def
refkey
(
self
):
return
list
(
self
.
keys
())[
0
]
@property
def
time
(
self
):
return
self
.
data
[
self
.
refkey
]
.
time
@property
def
stim
(
self
):
return
self
.
data
[
self
.
refkey
]
.
stim
Event Timeline
Log In to Comment