package core_extended

  1. Overview
  2. Docs
package core_extended
Legend:
Page
Library
Module
Module type
Parameter
Class
Class type
Source

Source file read.ml

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
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
open Core

exception Bad_csv_formatting = Parse_state.Bad_csv_formatting

module On_invalid_row = struct
  type 'a t =
    line_number:int
    -> int String.Map.t
    -> string Append_only_buffer.t
    -> exn
    -> [ `Skip | `Yield of 'a | `Raise of exn | `Fallback of 'a t ]

  let raise ~line_number header_map buffer exn =
    `Raise
      (Exn.create_s
         [%message
           "Exception raised in Delimited.Read"
             (line_number : int)
             (header_map : int String.Map.t)
             ~buffer:(Append_only_buffer.to_list buffer : string list)
             (exn : exn)])
  ;;

  let skip ~line_number:_ _ _ _ = `Skip
  let create = Fn.id
end

module Header = Header

module Builder = struct
  module T = struct
    type _ t =
      | Column : int -> string t
      | Header : string -> string t
      | Header_opt : string -> string option t
      | Return : 'a -> 'a t
      | Apply : ('b -> 'a) t * 'b t -> 'a t
      | Map : ('b -> 'a) * 'b t -> 'a t
      | Map2 : ('b -> 'c -> 'a) * 'b t * 'c t -> 'a t
      | Both : 'a t * 'b t -> ('a * 'b) t
      | Lambda : (int String.Map.t -> string Append_only_buffer.t -> 'a) -> 'a t
      | Label : 'a t * Info.t -> 'a t

    let return x = Return x
    let label t info = Label (t, info)

    let apply f x =
      match f with
      | Return f -> Map (f, x)
      | Map (f, w) -> Map2 (f, w, x)
      | _ -> Apply (f, x)
    ;;

    let map x ~f =
      match x with
      | Return x -> Return (f x)
      | _ -> Map (f, x)
    ;;

    let ( >>| ) t f = map t ~f

    let map2 x y ~f =
      match x, y with
      | Return x, Return y -> Return (f x y)
      | _ -> Map2 (f, x, y)
    ;;

    let map3 x y z ~f =
      match x, y, z with
      | Return x, Return y, Return z -> Return (f x y z)
      | _ -> Apply (Map2 (f, x, y), z)
    ;;

    let all ts = List.fold_right ts ~init:(return []) ~f:(map2 ~f:(fun x xs -> x :: xs))
    let all_unit ts = map ~f:ignore (all ts)
    let both x y = Both (x, y)
    let ( <*> ) = apply
    let ( *> ) u v = return (fun () y -> y) <*> u <*> v
    let ( <* ) u v = return (fun x () -> x) <*> u <*> v

    module Applicative_infix = struct
      let ( <*> ) = ( <*> )
      let ( *> ) = ( *> )
      let ( <* ) = ( <* )
      let ( >>| ) = ( >>| )
    end
  end

  include T

  let at_index i ~f =
    Label (Map (f, Column i), Info.of_lazy (lazy [%string "[at_index %{i#Int}]"]))
  ;;

  let at_header h ~f =
    Label (Map (f, Header h), Info.of_lazy (lazy [%string "[at_header \"%{h}\"]"]))
  ;;

  let at_header_opt h ~f =
    Label (Map (f, Header_opt h), Info.of_lazy (lazy [%string "[at_header_opt \"%{h}\"]"]))
  ;;

  let lambda f = Lambda f

  module Record_builder = Record_builder.Make (T)

  module Fields_O = struct
    let ( !! ) of_string field =
      Record_builder.field (at_header (Field.name field) ~f:of_string) field
    ;;

    let ( !? ) of_string field =
      Record_builder.field (at_header_opt (Field.name field) ~f:of_string) field
    ;;
  end

  module Open_on_rhs_intf = struct
    module type S = Read_intf.Open_on_rhs with type 'a t := 'a t
  end

  module Let_syntax = struct
    include T
    include Applicative_infix

    module Let_syntax = struct
      include T

      module Open_on_rhs = struct
        let at_index = at_index
        let at_header = at_header
        let at_header_opt = at_header_opt
        let label = label
      end
    end
  end

  module Without_headers = struct
    type 'a t =
      | Column : int -> string t
      | Return : 'a -> 'a t
      | Apply : ('b -> 'a) t * 'b t -> 'a t
      | Map : ('b -> 'a) * 'b t -> 'a t
      | Map2 : ('b -> 'c -> 'a) * 'b t * 'c t -> 'a t
      | Both : ('a t * 'b t) -> ('a * 'b) t
      | Lambda :
          (int String.Map.t -> string Append_only_buffer.t -> 'a) * int String.Map.t
          -> 'a t
      | Label : 'a t * Info.t -> 'a t

    let get_fields_used t =
      let open Option.Let_syntax in
      let rec fields : type a. a t -> Int.Set.t option = function
        | Return _ -> Some Int.Set.empty
        | Column i -> Some (Int.Set.singleton i)
        | Apply (f, x) ->
          let%bind f = fields f in
          let%map x = fields x in
          Set.union f x
        | Map (_, x) -> fields x
        | Map2 (_, x, y) ->
          let%bind x = fields x in
          let%map y = fields y in
          Set.union x y
        | Both (x, y) ->
          let%bind x = fields x in
          let%map y = fields y in
          Set.union x y
        | Lambda _ -> None
        | Label (t, _) -> fields t
      in
      fields t
    ;;

    let build t =
      let rec build' : type a. a t -> string Append_only_buffer.t -> a =
        fun t row ->
        match t with
        | Return x -> x
        | Column i -> Append_only_buffer.nth_exn row i
        | Apply (f, x) -> (build' f row) (build' x row)
        | Map (f, x) -> f (build' x row)
        | Map2 (f, x, y) -> f (build' x row) (build' y row)
        | Both (x, y) -> build' x row, build' y row
        | Lambda (f, header_map) -> f header_map row
        | Label (t, info) ->
          (try build' t row with
           | exn -> Exn.reraisef exn "Raised in %s" (Info.to_string_hum info) ())
      in
      let fields_used = get_fields_used t in
      match fields_used with
      | None -> build' t, None
      | Some fields_used ->
        let fields_used = Set.to_list fields_used in
        let mapping =
          List.mapi fields_used ~f:(fun i field_index -> field_index, i)
          |> Int.Map.of_alist_exn
        in
        let rec remap : type a. a t -> a t =
          fun t ->
          match t with
          | Column i -> Column (Map.find_exn mapping i)
          | Return x -> Return x
          | Apply (f, x) -> Apply (remap f, remap x)
          | Map (f, x) -> Map (f, remap x)
          | Map2 (f, x, y) -> Map2 (f, remap x, remap y)
          | Both (x, y) -> Both (remap x, remap y)
          | Lambda _ -> t
          | Label (t, info) -> Label (remap t, info)
        in
        build' (remap t), Some (Array.of_list fields_used)
    ;;
  end

  let build ~header_map t =
    let rec transform : type a. a t -> a Without_headers.t = function
      | Return x -> Without_headers.Return x
      | Column i -> Without_headers.Column i
      | Header column ->
        (match Map.find_exn (header_map : _ String.Map.t) column with
         | index -> Without_headers.Column index
         | exception (Not_found_s _ | Stdlib.Not_found) ->
           raise_s
             [%message
               "Missing column in header"
                 (column : string)
                 (header_map : int String.Map.t)])
      | Header_opt h ->
        (match Map.find_exn (header_map : _ String.Map.t) h with
         | index -> Without_headers.Map (Option.some, Column index)
         | exception (Not_found_s _ | Stdlib.Not_found) -> Without_headers.Return None)
      | Apply (f, x) -> Without_headers.Apply (transform f, transform x)
      | Map (f, x) -> Without_headers.Map (f, transform x)
      | Map2 (f, x, y) -> Without_headers.Map2 (f, transform x, transform y)
      | Both (x, y) -> Without_headers.Both (transform x, transform y)
      | Lambda f -> Without_headers.Lambda (f, header_map)
      | Label (t, info) -> Without_headers.Label (transform t, info)
    in
    let transformed = transform t in
    Without_headers.build transformed
  ;;
end

include Builder

module Parse_header = struct
  module Partial = struct
    type t =
      { state : string array option Parse_state.t
      ; transform : string array -> int String.Map.t
      }
  end

  module Success = struct
    type t =
      { header_map : int String.Map.t
      ; next_line_number : int
      ; consumed : int
      }
    [@@deriving sexp_of]
  end

  let header_map ~foldi header_row =
    foldi header_row ~init:String.Map.empty ~f:(fun i map header ->
      Map.set map ~key:header ~data:i)
  ;;

  let require_header required_headers' csv_headers' =
    let required_headers = String.Set.of_list required_headers' in
    let csv_headers = String.Set.of_array csv_headers' in
    let missing = Set.diff required_headers csv_headers in
    if not (Set.is_empty missing)
    then
      raise_s
        [%message
          "Header specified in `Require not present in csv document"
            (required_headers : String.Set.t)
            (csv_headers : String.Set.t)
            (missing : String.Set.t)];
    header_map ~foldi:Array.foldi csv_headers'
  ;;

  let create' ?strip ?sep ?quote ?start_line_number transform : Partial.t =
    let f ~line_number:_ r row =
      match r with
      | Some _ -> raise_s [%message "Header already parsed, cannot feed more input"]
      | None -> Some (Append_only_buffer.to_array row)
    in
    { state =
        Parse_state.create
          ?strip
          ?sep
          ?quote
          ?start_line_number
          ~fields_used:None
          ~init:None
          ~f
          ()
    ; transform
    }
  ;;

  let create ?strip ?sep ?quote ?(start_line_number = 1) ?(header = `No) () =
    match header with
    | `No ->
      Second
        { Success.header_map = String.Map.empty
        ; next_line_number = start_line_number
        ; consumed = 0
        }
    | `Add headers ->
      Second
        { header_map = header_map ~foldi:List.foldi headers
        ; next_line_number = start_line_number
        ; consumed = 0
        }
    | `Yes ->
      let f headers = header_map ~foldi:Array.foldi headers in
      First (create' ?strip ?sep ?quote ~start_line_number f)
    | `Require headers ->
      let f csv_headers = require_header headers csv_headers in
      First (create' ?strip ?sep ?quote ~start_line_number f)
    | `Replace headers ->
      let f _ = header_map ~foldi:List.foldi headers in
      First (create' ?strip ?sep ?quote ~start_line_number f)
    | `Transform f ->
      let f headers = header_map ~foldi:List.foldi (f (Array.to_list headers)) in
      First (create' ?strip ?sep ?quote ~start_line_number f)
    | `Filter_map f ->
      let foldi l ~init ~f =
        List.foldi l ~init ~f:(fun i acc x ->
          match x with
          | None -> acc
          | Some x -> f i acc x) [@nontail]
      in
      let f headers = header_map ~foldi (f (Array.to_list headers)) in
      First (create' ?strip ?sep ?quote ~start_line_number f)
  ;;

  let is_set (t : Partial.t) =
    match Parse_state.acc t.state with
    | Some row -> Some (t, row)
    | None -> None
  ;;

  (* We only want to parse the header here and nothing more.  To achieve that, we feed the
     parser with chunks of input that does not contain newlines but at the end.  After
     each chunk, we check whether a header was successfully parsed or not.  *)
  let input_string (t : Partial.t) ?(pos = 0) ?len input : _ Either.t =
    match is_set t with
    | Some _ -> raise_s [%message "Header already parsed, cannot feed more input"]
    | None ->
      let len =
        match len with
        | None -> String.length input - pos
        | Some len -> len
      in
      let start_pos = pos in
      let max_pos = start_pos + len in
      if start_pos < 0 || len < 0 || max_pos > String.length input
      then
        invalid_arg
          "Delimited_kernel.Read.Expert.Parse_header.input_string: index out of bound";
      let rec loop (t : Partial.t) ~pos ~len =
        if pos >= max_pos
        then First t
        else (
          let end_pos =
            match String.index_from input pos '\n' with
            | None -> max_pos
            | Some end_pos -> min max_pos (end_pos + 1)
          in
          let line_len = end_pos - pos in
          let t =
            { t with state = Parse_state.input_string t.state ~pos ~len:line_len input }
          in
          match is_set t with
          | None -> loop t ~pos:end_pos ~len:(len - line_len)
          | Some (t, row) ->
            Second
              { Success.header_map = t.transform row
              ; consumed = end_pos - start_pos
              ; next_line_number = Parse_state.current_line_number t.state
              })
      in
      loop t ~pos ~len
  ;;

  let finish_exn (t : Partial.t) =
    let t = { t with state = Parse_state.finish t.state } in
    match is_set t with
    | Some (t, row) ->
      { Success.header_map = t.transform row
      ; consumed = 0
      ; next_line_number = Parse_state.current_line_number t.state
      }
    | None ->
      if Parse_state.at_beginning_of_row t.state
      then
        (* An empty header is parsed as a single column with an empty name *)
        { Success.header_map = t.transform [| "" |]
        ; consumed = 0
        ; next_line_number = Parse_state.current_line_number t.state
        }
      else (* [Parse_state.finish] would have raised already in the cases  *)
        assert false
  ;;
end

module Streaming = struct
  type 'a builder_t = 'a t

  type 'a t =
    | Parsing_header of
        { partial : Parse_header.Partial.t
        ; init : 'a
        ; mk_state : 'a -> ?start_line_number:int -> int String.Map.t -> 'a Parse_state.t
        }
    | Parsing_rows of
        { header_map : int String.Map.t
        ; state : 'a Parse_state.t
        }

  let create_indexed
    ?strip
    ?sep
    ?quote
    ?start_line_number
    ?(on_invalid_row = (On_invalid_row.raise : _ On_invalid_row.t))
    ?(header : Header.t option)
    builder
    ~init
    ~f
    =
    let mk_state init ?start_line_number header_map =
      let row_to_'a, fields_used = Builder.build ~header_map builder in
      let f ~line_number init row =
        try f line_number init (row_to_'a row) with
        | exn ->
          let rec loop on_invalid_row =
            match on_invalid_row ~line_number header_map row exn with
            | `Yield x -> f line_number init x
            | `Skip -> init
            | `Fallback on_invalid_row -> loop on_invalid_row
            | `Raise exn -> raise exn
          in
          loop on_invalid_row
      in
      Parse_state.create ?strip ?sep ?quote ?start_line_number ~fields_used ~init ~f ()
    in
    match Parse_header.create ?strip ?sep ?quote ?start_line_number ?header () with
    | First partial -> Parsing_header { partial; mk_state; init }
    | Second { header_map; consumed = _; next_line_number } ->
      let state = mk_state init ~start_line_number:next_line_number header_map in
      Parsing_rows { header_map; state }
  ;;

  let create
    ?strip
    ?sep
    ?quote
    ?start_line_number
    ?on_invalid_row
    ?header
    builder
    ~init
    ~f
    =
    create_indexed
      ?strip
      ?sep
      ?quote
      ?start_line_number
      ?on_invalid_row
      ?header
      builder
      ~init
      ~f:(fun (_ : int) -> f)
  ;;

  let input_string t ?(pos = 0) ?len input =
    let len =
      match len with
      | None -> String.length input - pos
      | Some len -> len
    in
    match t with
    | Parsing_header { partial; mk_state; init } ->
      (match Parse_header.input_string partial ~pos ~len input with
       | First partial -> Parsing_header { partial; mk_state; init }
       | Second { consumed; header_map; next_line_number } ->
         let state = mk_state init header_map ~start_line_number:next_line_number in
         let state =
           Parse_state.input_string
             state
             ~pos:(pos + consumed)
             ~len:(len - consumed)
             input
         in
         Parsing_rows { header_map; state })
    | Parsing_rows { header_map; state } ->
      Parsing_rows { header_map; state = Parse_state.input_string state ~pos ~len input }
  ;;

  let input t ?pos ?len input =
    input_string
      t
      ?pos
      ?len
      (Bytes.unsafe_to_string ~no_mutation_while_string_reachable:input)
  ;;

  let finish t =
    match t with
    | Parsing_header { partial; mk_state; init } ->
      let { Parse_header.Success.consumed = _; header_map; next_line_number } =
        Parse_header.finish_exn partial
      in
      let state =
        Parse_state.finish (mk_state init header_map ~start_line_number:next_line_number)
      in
      Parsing_rows { state; header_map }
    | Parsing_rows { state; header_map } ->
      Parsing_rows { state = Parse_state.finish state; header_map }
  ;;

  let acc t =
    match t with
    | Parsing_header { init; _ } -> init
    | Parsing_rows { state; _ } -> Parse_state.acc state
  ;;

  let state t =
    match t with
    | Parsing_header _ -> `Parsing_header
    | Parsing_rows _ -> `Parsing_rows
  ;;

  let headers t =
    match t with
    | Parsing_header _ -> None
    | Parsing_rows { header_map; _ } -> Some (Map.key_set header_map)
  ;;

  let list_of_headers t =
    match t with
    | Parsing_header _ -> None
    | Parsing_rows { header_map; _ } ->
      Map.to_alist header_map
      |> List.sort ~compare:(fun a b -> Int.compare (snd a) (snd b))
      |> List.map ~f:fst
      |> Option.some
  ;;
end

module Expert = struct
  module Append_only_buffer = Append_only_buffer
  module Parse_state = Parse_state
  module Builder = Builder
end

let fold_string ?strip ?sep ?quote ?header ?on_invalid_row builder ~init ~f csv_string =
  let state =
    Streaming.create ?strip ?sep ?quote ?header ?on_invalid_row builder ~init ~f
  in
  Streaming.input_string state csv_string |> Streaming.finish |> Streaming.acc
;;

let list_of_string ?strip ?sep ?quote ?header ?on_invalid_row builder csv_string =
  fold_string
    ?strip
    ?sep
    ?quote
    ?header
    ?on_invalid_row
    builder
    csv_string
    ~init:(Append_only_buffer.create ())
    ~f:(fun queue row ->
    Append_only_buffer.append queue row;
    queue)
  |> Append_only_buffer.to_list
;;

let read_lines ?strip ?sep ?quote ?header ?on_invalid_row builder in_channel =
  let contents = In_channel.input_all in_channel in
  list_of_string ?strip ?sep ?quote ?header ?on_invalid_row builder contents
;;

let fold_lines
  ?(buffer_size = 8192)
  ?strip
  ?sep
  ?quote
  ?header
  ?on_invalid_row
  builder
  ~init
  ~f
  in_channel
  =
  let buf = Bytes.create buffer_size in
  let rec loop state =
    match In_channel.input in_channel ~buf ~pos:0 ~len:buffer_size with
    | 0 -> state
    | len ->
      Streaming.input_string
        state
        ~len
        (Bytes.unsafe_to_string ~no_mutation_while_string_reachable:buf)
      |> loop
  in
  Streaming.create ?strip ?sep ?quote ?header ?on_invalid_row builder ~init ~f
  |> loop
  |> Streaming.finish
  |> Streaming.acc
;;

module Row = struct
  type 'a builder_t = 'a t

  include Row

  let builder = lambda Row.Expert.of_buffer
end
OCaml

Innovation. Community. Security.

GitHub Discord Twitter Peertube RSS
About
Changelog Releases Industrial Users Academic Users Why OCaml
Ecosystem
Packages Community Events OCaml Planet Jobs
Resources
Install OCaml Get Started Platform Tools Language Manual Standard Library API Books Exercises Papers OCaml Playground Logo
Policies
Carbon Footprint Governance Privacy Code of Conduct