package Ġorg . radar base . output . format Ċ Ċ import Ġcom . openc sv . CSV Reader Ċ import Ġcom . openc sv . CSV Writer Ċ import Ġorg . apache . avro . generic . Generic Record Ċ import Ġorg . radar base . output . compression . Compression Ċ import Ġorg . radar base . output . util . Resource Context . Companion . resource Context Ċ import Ġorg . radar base . output . util . Sus pended Closeable . Companion . use Sus pended Ċ import Ġorg . radar base . output . util . Time Util . parse Date Ċ import Ġorg . radar base . output . util . Time Util . parse DateTime Ċ import Ġorg . radar base . output . util . Time Util . parse Time Ċ import Ġorg . radar base . output . util . Time Util . toDouble Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . io . IOException Ċ import Ġjava . io . InputStream Ċ import Ġjava . io . Reader Ċ import Ġjava . io . Writer Ċ import Ġjava . nio . file . Path Ċ import Ġkotlin . io . path . inputStream Ċ import Ġkotlin . io . path . output Stream Ċ Ċ class ĠCsv Av ro Converter Factory Ġ: ĠRecord Converter Factory Ġ{ ĊĠĠĠ Ġoverride Ġval Ġextension : ĠString Ġ= Ġ". csv " ĊĊĠĠĠ Ġoverride Ġval Ġformats : ĠCollection < String > Ġ= Ġset Of (" csv ") ĊĊĠĠĠ Ġ@ Throws ( IOException :: class ) ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġde duplicate ( ĊĠĠĠĠĠĠĠ ĠfileName : ĠString , ĊĠĠĠĠĠĠĠ Ġsource : ĠPath , ĊĠĠĠĠĠĠĠ Ġtarget : ĠPath , ĊĠĠĠĠĠĠĠ Ġcompression : ĠCompression , ĊĠĠĠĠĠĠĠ Ġdistinct Fields : ĠSet < String >, ĊĠĠĠĠĠĠĠ Ġignore Fields : ĠSet < String >, ĊĠĠĠ Ġ): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġ( header , Ġline Indexes ) Ġ= Ġsource . inputStream (). use Sus pended Ġ{ Ġinput Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Lines ( input , Ġcompression ) Ġ{ Ġheader , Ġlines Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( header Ġ== Ġnull ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfields Ġ= Ġfield Indexes ( header , Ġdistinct Fields , Ġignore Fields ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġcount Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġline Map Ġ= Ġlines ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Each Ġ{ Ġcount Ġ+= Ġ 1 Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Indexed Ġ{ Ġidx , Ġline Ġ-> ĠPair ( Array Wrapper ( line . by Index ( fields )), Ġidx ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. toMap ( HashMap ()) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( line Map . size Ġ== Ġcount ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" File Ġ{} Ġis Ġalready Ġde d uplicated . ĠSk ipping .", ĠfileName ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġline Map . values ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. to IntArray () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply Ġ{ Ġsort () Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġsource . inputStream (). use Sus pended Ġ{ Ġinput Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess Lines ( input , Ġcompression ) Ġ{ Ġ_, Ġlines Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġindex Index Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwrite Lines ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtarget , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠfileName , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcompression , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlines Ġ= Ġsequence Of ( header ) Ġ+ Ġlines . filter Indexed Ġ{ Ġi , Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( index Index Ġ< Ġline Indexes . size Ġ&& Ġline Indexes [ index Index ] Ġ== Ġi ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġindex Index Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġwrite Lines ( ĊĠĠĠĠĠĠĠ Ġtarget : ĠPath , ĊĠĠĠĠĠĠĠ ĠfileName : ĠString , ĊĠĠĠĠĠĠĠ Ġcompression : ĠCompression , ĊĠĠĠĠĠĠĠ Ġlines : ĠSequence < Array < String >>, ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġresource Context Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcsv Writer Ġ= Ġresource Chain Ġ{ Ġtarget . output Stream () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. chain Ġ{ Ġit . buffered () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. chain Ġ{ Ġcompression . compress ( fileName , Ġit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. chain Ġ{ Ġit . writer () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. con clude Ġ{ ĠCSV Writer ( it ) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġlines . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcsv Writer . write Next ( it , Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġfield Time Parsers Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠ Ġfield Time Parser (" value . time ") Ġ{ Ġit . parse Time () Ġ}, ĊĠĠĠĠĠĠĠ Ġfield Time Parser (" key . time Start ") Ġ{ Ġit . parse Time () Ġ}, ĊĠĠĠĠĠĠĠ Ġfield Time Parser (" key . start ") Ġ{ Ġit . parse Time () Ġ}, ĊĠĠĠĠĠĠĠ Ġfield Time Parser (" value . dateTime ") Ġ{ Ġit . parse DateTime ()?. toDouble () Ġ}, ĊĠĠĠĠĠĠĠ Ġfield Time Parser (" value . date ") Ġ{ Ġit . parse Date ()?. toDouble () Ġ}, ĊĠĠĠĠĠĠĠ Ġfield Time Parser (" value . time Received ") Ġ{ Ġit . parse Time () Ġ}, ĊĠĠĠĠĠĠĠ Ġfield Time Parser (" value . time Completed ") Ġ{ Ġit . parse Time () Ġ}, ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġread Time Seconds ( ĊĠĠĠĠĠĠĠ Ġsource : ĠInputStream , ĊĠĠĠĠĠĠĠ Ġcompression : ĠCompression , ĊĠĠĠ Ġ): ĠPair < Array < String >, ĠList < Double >> ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġprocess Lines ( source , Ġcompression ) Ġ{ Ġheader , Ġlines Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġheader Ġ?: Ġreturn @ process Lines Ġnull ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġparsers Ġ= Ġfield Time Parsers . map NotNull Ġ{ Ġ( key , Ġparser ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader . indexOf ( key ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. take If Ġ{ Ġit Ġ>= Ġ 0 Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ? . let Ġ{ ĠPair ( it , Ġparser ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠPair ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġheader , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( parsers . isEmpty ()) Ġempty List () Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlines . map NotNull Ġ{ Ġline Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfor Ġ(( index , Ġparser ) Ġin Ġparsers ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġparser ( line [ index ]) ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ map NotNull Ġit ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. toList () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcontains ( ĊĠĠĠĠĠĠĠ Ġsource : ĠPath , ĊĠĠĠĠĠĠĠ Ġrecord : ĠGeneric Record , ĊĠĠĠĠĠĠĠ Ġcompression : ĠCompression , ĊĠĠĠĠĠĠĠ Ġusing Fields : ĠSet < String >, ĊĠĠĠĠĠĠĠ Ġignore Fields : ĠSet < String >, ĊĠĠĠ Ġ): ĠBoolean Ġ= Ġsource . inputStream (). use Ġ{ Ġinput Ġ-> ĊĠĠĠĠĠĠĠ Ġprocess Lines ( input , Ġcompression ) Ġ{ Ġheader , Ġlines Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ ĠcheckNotNull ( header ) Ġ{ Ġ" Empty Ġfile Ġfound " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconverter Ġ= ĠCsv Av ro Data Converter ( header ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Values Ġ= Ġconverter . convert Record Values ( record ). to Typed Array () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġindexes Ġ= Ġfield Indexes ( header , Ġusing Fields , Ġignore Fields ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( indexes Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlines . any Ġ{ Ġline Ġ-> Ġrecord Values . content Equals ( line ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlines . any Ġ{ Ġline Ġ-> Ġindexes . all Ġ{ Ġi Ġ-> Ġrecord Values [ i ] Ġ== Ġline [ i ] Ġ} Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Throws ( IOException :: class ) ĊĠĠĠ Ġoverride Ġfun Ġconverter For ( ĊĠĠĠĠĠĠĠ Ġwriter : ĠWriter , ĊĠĠĠĠĠĠĠ Ġrecord : ĠGeneric Record , ĊĠĠĠĠĠĠĠ Ġwrite Header : ĠBoolean , ĊĠĠĠĠĠĠĠ Ġreader : ĠReader , ĊĠĠĠ Ġ): ĠCsv Av ro Converter Ġ= ĠCsv Av ro Converter ( writer , Ġwrite Header , Ġreader , Ġheader For ( record )) ĊĊĠĠĠ Ġoverride Ġval Ġhas Header : ĠBoolean Ġ= Ġtrue ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( Csv Av ro Converter Factory :: class . java ) ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġfield Time Parser ( name : ĠString , Ġmethod : Ġ( String ) Ġ-> ĠDouble ?) Ġ= ĠPair ( name , Ġmethod ) ĊĊĠĠĠĠĠĠĠ Ġprivate Ġsuspend Ġinline Ġfun Ġ< T > Ġprocess Lines ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinput : ĠInputStream , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcompression : ĠCompression , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess : Ġ( header : ĠArray < String > ?, Ġlines : ĠSequence < Array < String >>) Ġ-> ĠT , ĊĠĠĠĠĠĠĠ Ġ): ĠT Ġ= Ġresource Context Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcsv Reader Ġ= Ġresource Chain Ġ{ Ġcompression . de compress ( input ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. chain Ġ{ Ġit . buffered Reader () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. con clude Ġ{ ĠCSV Reader ( it ) Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġheader Ġ= Ġcsv Reader . read Next () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlines Ġ= Ġif Ġ( header Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġgenerate Sequence Ġ{ Ġcsv Reader . read Next () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġempty Sequence () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocess ( header , Ġlines ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġfield Indexes ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġheader : ĠArray < String >, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġusing Fields : ĠSet < String >, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġignore Fields : ĠSet < String >, ĊĠĠĠĠĠĠĠ Ġ): ĠInt Array ? Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( using Fields . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġindexes Ġ= Ġusing Fields . map Ġ{ Ġf Ġ-> Ġheader . indexOf ( f ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( indexes . none Ġ{ Ġidx Ġ-> Ġidx Ġ== Ġ- 1 Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġindexes . to IntArray () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( ignore Fields . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġignore Indexes Ġ= Ġignore Fields . map To ( HashSet ()) Ġ{ Ġf Ġ-> Ġheader . indexOf ( f ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( ignore Indexes . any Ġ{ Ġidx Ġ-> Ġidx Ġ!= Ġ- 1 Ġ}) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġ( header . indices Ġ- Ġignore Indexes ). to IntArray () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġprivate Ġclass ĠArray Wrapper < T >( val Ġvalues : ĠArray < T >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval ĠhashCode Ġ= Ġvalues . content HashCode () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġequals ( other : ĠAny ?): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( this Ġ=== Ġother ) Ġreturn Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( java Class Ġ!= Ġother ?. java Class ) Ġreturn Ġfalse ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġother Ġas ĠArray Wrapper < *> ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġvalues . content Equals ( other . values ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠhashCode (): ĠInt Ġ= ĠhashCode ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ@ Throws ( Index OutOfBoundsException :: class ) ĊĠĠĠĠĠĠĠ Ġinline Ġfun Ġ< re ified ĠT > ĠArray < T >. by Index ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġindexes : ĠInt Array ?, ĊĠĠĠĠĠĠĠ Ġ): ĠArray < T > Ġ= Ġif Ġ( indexes Ġ== Ġnull ) Ġthis Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠArray ( indexes . size ) Ġ{ Ġi Ġ-> Ġthis [ indexes [ i ]] Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ