U
vii7 @ sp d Z ddlZddlZddlZddlmZ ddlmZmZm Z m
Z
e dZ dd Z
G dd d Ze
fd
d ZdS ) z
Functions for serialize Pandas DataFrame.
Much of the code here is inspired by that in the aioinflux packet found here: https://github.com/gusutabopb/aioinflux
N)WritePrecision) _ESCAPE_KEY_ESCAPE_STRING_ESCAPE_MEASUREMENTDEFAULT_WRITE_PRECISIONz1influxdb_client.client.write.dataframe_serializerc s, fddt t jD }t jf| S )Nc s g | ]} j d d |f qS N)iloc).0k
data_frame b/opt/alt/python38/lib/python3.8/site-packages/influxdb_client/client/write/dataframe_serializer.py
s z _itertuples..)rangelencolumnszipindex)r colsr
r r _itertuples s r c @ s> e Zd ZdZedfeddddZd edddZd d
ZdS ) DataframeSerializerz'Serialize DataFrame into LineProtocols.N)
chunk_sizereturnc # s ddl m m} t| js.tdt||d}|dkrHtd|dd}|dd} |j d d
}|dkrz|j
n|| }
t|
j r|
}
n j
|
|d }
| rt|
jr|
| }
n |
j| }
t|
d r|
jdkr|
d
}
|dkr|
|_
n|
||< |d} t| pg } g } g }
g }g }|jrh|j D ](\}}||jkr>|||< | | q>tt|j dd d}| }d}|D ]\}\}}t |}dt | d}| | t! |d }d| d}|| kr8|j"| rd| d| d| d| d }nd | d| d }|
| qn |dk rT||krT|}qt |dkrfd nd }t#|j|j$st#|j|j%st#|j|j&rt#|j|j$rd nd }|j"| rd!| d"| | d#| d$| d%
} n| | d#| d$| } n>|j"| r&d&| d'| | d(| d)} n| | d(| d*} | | | | qt | t'} d (|
}
d (|}d+| } |t)j*krd,| } n*|t)j+krd-| } n|t)j,krd.| } t-d/|
d0| d0| d1| t!t.| d2} t/|j D ]&\} }!| | kr|0| d i|j1}q fd3d4}"|| _2| | _3|| _4|j"|d d | _5|"| _6|dk rt7t89t |t:| | _;|| _`_ - ``DataFrame``
:key data_frame_timestamp_timezone: name of the timezone which is used for timestamp column - ``DataFrame``
)pdnpz%Must be DataFrame, but type was: {0}.data_frame_measurement_nameNz4"data_frame_measurement_name" is a Required ArgumentZdata_frame_timestamp_columnZ data_frame_timestamp_timezoneF)deep)unittzinfoUTCdata_frame_tag_columnsc S s | d d S )N r r
)colr
r
r z.DataframeSerializer.__init__..)keyr z{keys[z]}r# zp[]z${
'' if z == '' or pd.isna(z&) else
f',z={str(z7).translate(_ESCAPE_STRING)}'
},z).translate(_ESCAPE_KEY)} iz{"" if pd.isna(z ) else f"z={}z"}z,{
'' if pd.isna(z%) else
f'z="{str(z8).translate(_ESCAPE_STRING)}"'
}z ).translate(_ESCAPE_STRING)}"z
{p[%s].value}z{int(p[%s].value / 1e3)}z{int(p[%s].value / 1e6)}z{int(p[%s].value / 1e9)}z lambda p: f"""{measurement_name} z""")measurement_namer r keysr c s t t fdd|S )Nc s | S r )Zisnax)pr r
r r% r& zDDataframeSerializer.__init__.._any_not_nan..)anymap)r2 Zindexesr )r2 r _any_not_nan s z2DataframeSerializer.__init__.._any_not_nan)=extrasr r
isinstanceZ DataFrame TypeErrorformattypegetcopyr Z PeriodIndexZ to_timestamp to_datetimeZ
DatetimeIndexZ tz_localizedthasattrr setZ defaultTagsitemsr addsorted enumerateZdtypesZisnullr3 strr append translater r
issubclassintegerZfloatingZbool_r joinr ZUSZMSSevalr dictreplacenanr f
field_indexesfirst_field_maybe_nullr6 intmathceilfloatnumber_of_chunksr )#selfr point_settings precisionr kwargsr r Ztimestamp_columnZtimestamp_timezoneZdata_frame_timestampr" r/ tagsfieldsrR r' valuer Z null_columnsZtimestamp_indexr Z
key_formatZ field_indexZ
val_format key_valuesepsuffixZ field_valuer. timestamprQ r
vr6 r
r5 r __init__ s 9
0 $
z DataframeSerializer.__init__) chunk_idxc s |dkr j }n2td|d j j | j |d j } jrt fddt fddt|D }t|S tt j
t|S dS )z
Serialize chunk into LineProtocols.
:param chunk_idx: The index of chunk to serialize. If `None` then serialize whole dataframe.
NzSerialize chunk %s/%s ...r# c 3 s" | ]}t d d |V qdS )z ^(( |[^ ])* ),([a-zA-Z0-9])(.*)z\1\3\4N)resubrQ )r r2 rY r
r s z0DataframeSerializer.serialize..c s | jS r )r6 rR r0 ri r
r r% r& z/DataframeSerializer.serialize..) r loggerdebugrX r rS filterr listr4 rQ )rY rf chunkZlpr
ri r serialize s
z DataframeSerializer.serializec C s | j S )zy
Return the number of chunks.
:return: number of chunks or None if chunk_size is not specified.
)rX ri r
r
r rX s z$DataframeSerializer.number_of_chunks)N) __name__
__module__ __qualname____doc__r rT re rp rX r
r
r
r r s ^r c K s t | ||f| S )a
Serialize DataFrame into LineProtocols.
:param data_frame: Pandas DataFrame to serialize
:param point_settings: Default Tags
:param precision: The precision for the unix timestamps within the body line-protocol.
:key data_frame_measurement_name: name of measurement for writing Pandas DataFrame
:key data_frame_tag_columns: list of DataFrame columns which are tags, rest columns will be fields
:key data_frame_timestamp_column: name of DataFrame column which contains a timestamp. The column can be defined as a :class:`~str` value
formatted as `2018-10-26`, `2018-10-26 12:00`, `2018-10-26 12:00:00-05:00`
or other formats and types supported by `pandas.to_datetime `_ - ``DataFrame``
:key data_frame_timestamp_timezone: name of the timezone which is used for timestamp column - ``DataFrame``
)r rp )r rZ r[ r\ r
r
r data_frame_to_list_of_points s ru )rt loggingrU rg Zinfluxdb_clientr Z"influxdb_client.client.write.pointr r r r getLoggerrk r r ru r
r
r
r s