feat: Use LiveView streams; extract ML work; show in-progress

- Move ML processing into new MedicalTranscription.Transcriber module.
- Add updates to a LiveView stream.
- Show a message when the processing is in progress.

lib/medical_transcription/transcriber.ex

@@ -0,0 +1,78 @@
+defmodule MedicalTranscription.Transcriber do
+  def test_stream(live_view_pid) do
+    # Stream test
+    1..5
+    |> Stream.with_index()
+    |> Stream.each(fn {element, index} ->
+      Process.sleep(1_000)
+
+      send(
+        live_view_pid,
+        {:transcription_row,
+         %{id: index, start_mark: 1, end_mark: 2, text: "Hello, world", tags: "ABCD"}}
+      )
+    end)
+    |> Stream.run()
+  end
+
+  def stream_transcription(live_view_pid, audio_file_path) do
+    # Audio transcription only
+    for chunk <- Nx.Serving.batched_run(TranscriptionServing, {:file, audio_file_path}) do
+      chunk_result = %{
+        # id: index,
+        start_mark: chunk.start_timestamp_seconds,
+        end_mark: chunk.end_timestamp_seconds,
+        text: chunk.text,
+        tags: ""
+      }
+      |> dbg()
+
+      send(live_view_pid, {:transcription_row, chunk_result})
+    end
+  end
+
+  def stream_transcription_and_search(live_view_pid, audio_file_path) do
+    {model_info, tokenizer} = AudioTagger.Classifier.SemanticSearch.prepare_model()
+
+    labels_df = AudioTagger.SampleData.icd10_codes()
+    label_embeddings_path = Path.join(__DIR__, "../../icd10_vector_tensors.bin")
+
+    label_embeddings =
+      AudioTagger.Classifier.SemanticSearch.load_label_vectors(label_embeddings_path)
+
+    # Audio transcription + semantic search
+    for {chunk, index} <-
+          TranscriptionServing
+          |> Nx.Serving.batched_run({:file, audio_file_path})
+          |> Stream.with_index() do
+      # TODO: A potential improvement would be to not code each chunk of transcribed audio, but to instead gather
+      # complete sentences based on punctuation.
+      tags =
+        AudioTagger.Classifier.SemanticSearch.tag_one(
+          {model_info, tokenizer},
+          labels_df,
+          label_embeddings,
+          chunk.text
+        )
+
+      [start_mark, end_mark] =
+        for seconds <- [chunk.start_timestamp_seconds, chunk.end_timestamp_seconds] do
+          seconds |> round() |> Time.from_seconds_after_midnight() |> Time.to_string()
+        end
+
+      chunk_result = %{
+        id: index,
+        start_mark: start_mark,
+        end_mark: end_mark,
+        text: chunk.text,
+        tags: tags
+      }
+
+      send(live_view_pid, {:transcription_row, chunk_result})
+    end
+
+    # |> Stream.run()
+
+    # {:ok, %{tagged_audio: result}}
+  end
+end

lib/medical_transcription_web/live/home_live/index.ex

@@ -6,13 +6,20 @@ defmodule MedicalTranscriptionWeb.HomeLive.Index do
     {:ok,
      socket
      |> assign(:uploaded_files, [])
-     |> assign(:transcription_rows, [])
-     |> assign_async(:tagged_audio, fn -> {:ok, %{tagged_audio: []}} end)
+     |> stream(:transcription_rows, [])
+     |> assign(:transcription_in_progress, false)
      |> allow_upload(:audio, accept: ~w(.mp3), max_entries: 1)}
   end
 
   @impl Phoenix.LiveView
   def render(assigns) do
+    # TODO: Show a loading state while returning results from the audio transcription & tagging.
+    # TODO: Stream audio recording instead of uploaded audio.
+    # TODO: Allow editing the transcription inline to correct mistakes. Then, retag based on the updated transcription.
+    # TODO: Show multiple codes.
+    # TODO: Allow users to accept/decline suggested codes.
+    # TODO: Train model based on user feedback for suggested codes.
+
     ~H"""
     <form id="audio-form" phx-submit="save" phx-change="validate">
       <div class="flex flex-col space-y-4">
@@ -28,15 +35,18 @@ defmodule MedicalTranscriptionWeb.HomeLive.Index do
           Transcribe and Tag Audio <span aria-hidden="true">&rarr;</span>
         </button>
 
-        <%= for entry <- @uploads.audio.entries do %>
-          <p>Submitted: <%= entry.client_name %></p>
+        <%= if @transcription_in_progress do %>
+          <div class="flex gap-2 items-center p-2 rounded-md text-slate-800 text-sm bg-slate-200 border border-slate-300">
+            <.icon name="hero-arrow-path" class="w-4 h-4 animate-spin" />
+            <p>Transcribing and tagging audio file...</p>
+          </div>
         <% end %>
 
-        <.table id="streamed_result" rows={@transcription_rows}>
-          <:col :let={row} label="Start"><%= row.start_mark %></:col>
-          <:col :let={row} label="End"><%= row.end_mark %></:col>
-          <:col :let={row} label="Text"><%= row.text %></:col>
-          <:col :let={row} label="Codes"><%= row.tags %></:col>
+        <.table id="streamed_result" rows={@streams.transcription_rows}>
+          <:col :let={row} label="Start"><%= elem(row, 1).start_mark %></:col>
+          <:col :let={row} label="End"><%= elem(row, 1).end_mark %></:col>
+          <:col :let={row} label="Text"><%= elem(row, 1).text %></:col>
+          <:col :let={row} label="Codes"><%= elem(row, 1).tags %></:col>
         </.table>
       </div>
     </form>
@@ -59,66 +69,37 @@ defmodule MedicalTranscriptionWeb.HomeLive.Index do
 
     # Task async -> Audio Tagger -> get transcribed audio
 
-    # Ideas:
-    # 1 - Stream output
-    # 2 - More intelligent code matching (e.g. return more than one if close, filter out low thresholds)
-
     uploaded_file = Enum.at(uploaded_files, 0)
-
     live_view_pid = self()
 
+    Task.async(fn -> transcribe_and_tag_audio(live_view_pid, uploaded_file) end)
+
     socket =
       socket
-      |> assign_async(:tagged_audio, fn ->
-        transcribe_and_tag_audio(live_view_pid, uploaded_file)
-      end)
+      |> assign(:transcription_in_progress, true)
+      |> assign(:transcription_rows, [])
       |> update(:uploaded_files, &(&1 ++ uploaded_files))
 
     {:noreply, socket}
   end
 
+  @impl true
   def handle_info({:transcription_row, chunk_result}, socket) do
-    socket.assigns.transcription_rows |> IO.inspect(label: "transcription_rows")
-    chunk_result |> IO.inspect(label: "chunk_result")
+    # The processing sends a message as each chunk of text is coded. See here for some background and potential
+    # inspiration for this: https://elixirforum.com/t/liveview-asynchronous-task-patterns/44695
+    {:noreply, stream_insert(socket, :transcription_rows, chunk_result)}
+  end
 
-    {:noreply, update(socket, :transcription_rows, &(&1 ++ [chunk_result]))}
+  @impl true
+  def handle_info({ref, result}, socket) do
+    # See this Fly article for the usage of Task.async to start `transcribe_and_tag_audio/2` and handle the end of the
+    # task here: https://fly.io/phoenix-files/liveview-async-task/
+    Process.demonitor(ref, [:flush])
+    {:noreply, assign(socket, :transcription_in_progress, false)}
   end
 
   def transcribe_and_tag_audio(live_view_pid, audio_file_path) do
-    {model_info, tokenizer} = AudioTagger.Classifier.SemanticSearch.prepare_model()
-
-    labels_df = AudioTagger.SampleData.icd10_codes()
-    label_embeddings_path = Path.join(__DIR__, "../../../../icd10_vector_tensors.bin")
-
-    label_embeddings =
-      AudioTagger.Classifier.SemanticSearch.load_label_vectors(label_embeddings_path)
-
-    result =
-      TranscriptionServing
-      |> Nx.Serving.batched_run({:file, audio_file_path})
-      |> Enum.map(fn chunk ->
-        tags =
-          AudioTagger.Classifier.SemanticSearch.tag_one(
-            {model_info, tokenizer},
-            labels_df,
-            label_embeddings,
-            chunk.text
-          )
-
-        [start_mark, end_mark] =
-          for seconds <- [chunk.start_timestamp_seconds, chunk.end_timestamp_seconds] do
-            seconds |> round() |> Time.from_seconds_after_midnight() |> Time.to_string()
-          end
-
-        chunk_result = %{start_mark: start_mark, end_mark: end_mark, text: chunk.text, tags: tags}
-        send(live_view_pid, {:transcription_row, chunk_result})
-
-        chunk_result
-      end)
-      # |> Enum.map(&Function.identity/1)
-      # |> dbg()
-
-    {:ok, %{tagged_audio: result}}
+    MedicalTranscription.Transcriber.stream_transcription_and_search(live_view_pid, audio_file_path)
   end
 
   def error_to_string(:too_large), do: "Too large"