import json import os from dataclasses import dataclass, field from typing import List, Optional, Dict from PIL import Image import streamlit as st from markdown import markdown st.set_page_config(layout="wide", page_title="Therapy Session Annotation", initial_sidebar_state="collapsed") import re, textwrap, html as py_html from pathlib import Path from streamlit.components.v1 import html as st_html import numpy as np import pandas as pd from fsspec.implementations.local import LocalFileSystem from huggingface_hub import HfFileSystem import streamlit.components.v1 as components @dataclass class Field: type: str title: str name: str = None mandatory: bool = True # if value of field is in the list of those values, makes following siblings mandatory following_mandatory_values: list = False skip_mandatory: bool = False help: Optional[str] = None children: Optional[List['Field']] = None other_params: Optional[Dict[str, object]] = field(default_factory=lambda: {}) # Function to get user ID from URL def get_param_from_url(param): user_id = st.query_params.get(param, "") return user_id ######################################################################################## # 'local' or 'hf'. hf is for Hugging Face file system but has limits on the number of access per hour filesystem = 'hf' input_repo_path = 'datasets/emvecchi/therapy_annotation' output_repo_path = 'datasets/emvecchi/therapy_annotation/pilot' to_annotate_file_name = 'to_annotate.csv' # CSV file to annotate COLS_TO_SAVE = ['dialogue_id','dialogue_name','generatedPatient'] agreement_labels = ['strongly disagree', 'disagree', 'neither agree no disagree', 'agree', 'strongly agree'] mentioned_y_n = ['yes, explicitly', 'yes, implicitly', 'no', 'not sure', 'other'] degree_labels = ['Not at all', 'Slightly', 'Moderately', 'Very much', 'Extremely'] quality_labels = ['very poor', 'poor', 'acceptable', 'good', 'very good'] priority_labels = ['not a priority', 'low priority', 'neutral', 'moderate priority', 'high priority'] yes_no_labels = ['no','yes'] yes_no_other_labels = ['no','yes','other'] default_labels = agreement_labels MAX_RUPTURE_MARKERS = 8 # 1 initial + up to 7 extra RUPTURE_LINE_INPUT_TYPE = "number" # or "text" if your line numbers aren't pure ints rupture_choices = ["", "Denial", "Minimal response", "Abstract communication", "Avoidant storytelling and/or topic-shifting", "Deferential / appeasing behavior", "Content/affect split (i.e. talking content vs emotional tone mismatch)", "Self-criticism and/or hopelessness", "Complaints or concerns about the therapist", "Patient rejects therapist intervention", "Complaints/concerns about the activities of therapy", "Complaints/concerns about the parameters of therapy (e.g. scheduling, rules)", "Complaints/concerns about progress in therapy", "Patient defends self against therapist", "Efforts to control or pressure the therapist", "Other"] default_choices = rupture_choices consent_text = ''' ## Consent Form You will be asked to take part in a research study. Before you decide to take part in this study, it is important that you understand why the study is being done and what it involves. Please read the following information carefully. ________________________________________________________________________________________ Project title: [...]\\ Researchers: [...]\\ Institute: Institute for Machine Speech Processing (IMS)\\ University: University of Stuttgart\\ Contact: eva-maria.vecchi@ims.uni-stuttgart.de _________________________________________________________________________________________ ### Description of the research study In this study, we investigate [...] The intended use of the results of this study includes [...] **Time required:** Your participation will take up to an estimated [...] hours. The time required may vary on an individual basis. **Risks and benefits:** The risks to your participation in this online survey are those associated with basic computer tasks, including boredom, fatigue, mild stress, or breach of confidentiality. Some of the topics discussed in the online posts to be annotated may include violence, suicide or rape. The only benefit to you is the learning experience from participating in a research study. The benefit to society is the contribution to scientific knowledge **Compensation:** You will be compensated for participating in this study. If you are interested, we will also be more than happy to share more information about our research with you. **Voluntary participation:** Your participation in this study is voluntary. It is your decision whether or not to participate in this study. If you decide to participate in this study, you will be asked to confirm this consent form ("I agree."). Even after signing the consent form, you can withdraw from participation at any time and without giving any reason. Partial data will not be analysed. **Confidentiality:** Your responses to this experiment will be anonymous. Please do not share any Information that can be used to identify you. The researcher(s) will make every effort to maintain your confidentiality. **Contact:** If at any time you have questions about this study or would like to report any adverse effects due to this study, please contact the researcher(s). **Trigger Warning:** [...] ### Consent: Please indicate, in the box below, that you are at least 18 years old, have read and understood this consent form, are comfortable using the English and German languages to complete the survey, and you agree to participate in this online research survey. - *I am age 18 or older.* - *I have read this consent form or had it read to me.* - *I am comfortable using the English language to participate in this survey.* - *I agree to participate in this research and I want to continue with the survey.* ''' fields: List[Field] = [ Field(name="patient", type="input_col", title=" "), Field(type="expander", title="**Session Transcription:** *(expand)*", children=[ Field(name="dialogue_name", type="input_col", title=""), ]), Field(type="container", title="##### Session-/Patient-Specific Properties", children=[ Field(name="diagnosis", type="likert_radio", title="Is there mention of a specific [possible] clinical diagnosis?", other_params={'labels': mentioned_y_n}, mandatory=True), Field(name="stressor", type="likert_radio", title="Does the patient mention sources of psychological pressure that contribute to the development or worsening of the patient's state?", other_params={'labels': mentioned_y_n}, mandatory=True), ]), Field(type="container", title="#### Rupture Markers", children=[ Field(type="expander", title="**Overview of rupture markers** *(expand)*", children=[ Field(type="markdown", title="", other_params={"path": "rupture_marker_overview.md"}), ]), Field(name="rupture_marker", type="rupture_markers", title="Select rupture markers noted in the session, include line numbers where rupture is found.", mandatory=False), ]), Field(type="container", title="#### True-To-Patient-Prompt Features", children=[ Field(type="expander", title="**Patient Role Description:** *(expand)*", children=[ Field(name="role_name", type="input_col", title=""), ]), Field(name="emotion", type="likert_radio", title="To what extent did the simulated patient’s emotional expressions (e.g., frustration, sadness, guardedness) align with those expected for this role?", other_params={"labels": degree_labels}, mandatory=True), Field(name="conversationalStyle", type="likert_radio", title="How well did the patient consistently exhibit the conversational style they were assigned, and did this feel realistic compared to actual patients with this style?", other_params={"labels": degree_labels}, mandatory=True), Field(name="maladaptive", type="likert_radio", title="How accurately did the patient’s statements reflect distorted thoughts consistent with the underlying cognitive model?", other_params={"labels": degree_labels}, mandatory=True), Field(name="fidelity", type="likert_radio", title="Overall, how closely did the patient reflect real patients in terms of realism, variability, and challenge during the session?", other_params={"labels": degree_labels}, mandatory=True), ]), Field(type="container", title="#### Interpersonal therapeutic competence", children=[ Field(name="v_expressiveness", type="likert_radio", title="**Verbal expressiveness.** How fluent, clear, and confident is the person’s verbal communication (regardless of content)?", other_params={'labels': ['Very unclear','Unclear','Average','Clear','Very Clear']}, mandatory=True, following_mandatory_values=[1]), Field(name="hope", type="likert_radio", title="**Hope and positive expectation.** To what extent does the person convey optimism, confidence, and positive expectations for the patient’s future or therapy?", other_params={'labels': ['Hopeless','Discouraging','Neutral','Hopeful','Very hopeful']}, mandatory=True, following_mandatory_values=[1]), Field(name="credible", type="likert_radio", title="**Ability to be credible and convincing.** How credible and convincing is the person when presenting a new perspective or interpretation to the patient?", other_params={'labels': ['Unconvincing','Weak','Neutral','Convincing','Very convincing']}, mandatory=True, following_mandatory_values=[1]), Field(name="e_expressiveness", type="likert_radio", title="**Emotional expressiveness.** How well does the person’s voice and manner convey emotional engagement and intensity (without being hostile)?", other_params={'labels': ['Flat','Low','Moderate','Expressive','Very expressive']}, mandatory=True, following_mandatory_values=[1]), Field(name="acceptance", type="likert_radio", title="**Acceptance, understanding, and warmth.** To what extent does the person show care, acceptance, and warmth toward the patient (without judgment)?", other_params={'labels': ['Cold','Limited','Neutral','Warm','Very warm']}, mandatory=True, following_mandatory_values=[1]), Field(name="empathy", type="likert_radio", title="**Empathy.** How well does the person demonstrate an accurate, deeper understanding of the patient’s subjective experience?", other_params={'labels': ['Distorted','Superficial','Partial','Accurate','Deep']}, mandatory=True, following_mandatory_values=[1]), Field(name="alliance", type="likert_radio", title="**Ability to form alliances.** How effectively does the person create a collaborative “working together” atmosphere with the patient?", other_params={'labels': ['Blocking','Weak','Moderate','Cooperative','Strong']}, mandatory=True, following_mandatory_values=[1]), Field(name="tensions", type="likert_radio", title="**Addressing tensions and crises.** How effectively does the person acknowledge and work through interpersonal tensions or crises in the interaction?", other_params={'labels': ['Avoiding','Poor','Partial','Productive','Resolving']}, mandatory=True, following_mandatory_values=[1]), ]), Field(type="container", title="##### Additional Comments *(optional)*", children=[ Field(name="other_comments", type="text", title="Please provide any additional details or information:", mandatory=False), ]), ] url_conditional_fields = [ Field(name="skip", type="skip_checkbox", title="*I am uncomfortable annotating this text and voluntarily skip this instance*", mandatory=False) ] INPUT_FIELD_DEFAULT_VALUES = {'slider': 0, 'text': '', 'textarea': '', 'checkbox': False, 'radio': None, 'select_slider': 0, 'multiselect': [], 'likert_radio': None, 'y_n_radio': None} SHOW_HELP_ICON = False SHOW_VALIDATION_ERROR_MESSAGE = True ######################################################################################## if filesystem == 'hf': HF_TOKEN = os.environ.get("HF_TOKEN_WRITE") print("is none?", HF_TOKEN is None) hf_fs = HfFileSystem(token=HF_TOKEN) else: hf_fs = LocalFileSystem() def get_start_index(): if hf_fs.exists(output_repo_path + '/' + get_base_path()): files = hf_fs.ls(output_repo_path + '/' + get_base_path()) return len(files) #-1 else: return -2 def read_data(): with hf_fs.open(input_repo_path + '/' + to_annotate_file_name) as f: return pd.read_csv(f) def read_saved_data(): _path = get_path() if hf_fs.exists(output_repo_path + '/' + _path): with hf_fs.open(output_repo_path + '/' + _path) as f: try: return json.load(f) except json.JSONDecodeError as e: print(e) return None # Write a remote file def save_data(data): if not hf_fs.exists(f"{output_repo_path}/{get_base_path()}"): hf_fs.mkdir(f"{output_repo_path}/{get_base_path()}") with hf_fs.open(f"{output_repo_path}/{get_path()}", "w") as f: f.write(json.dumps(data)) def get_base_path(): return f"{st.session_state.user_id}" def get_path(): return f"{get_base_path()}/{st.session_state.current_index}.json" def display_dialogue(hf_path: str): with hf_fs.open(hf_path, "rb") as f: txt = f.read().decode("utf-8") txt = load_text(hf_path) st.markdown(txt, unsafe_allow_html=True) def display_image(image_path): with hf_fs.open(image_path) as f: img = Image.open(f) st.image(img, caption='', use_column_width=True) def _ensure_key(key: str, default): if key not in st.session_state: st.session_state[key] = default return st.session_state[key] SPEAKER_RE = re.compile(r'^\s*\*\*(T|P):\*\*\s*(.*)$') def read_md(path: str) -> str: """Read .md file from HF dataset or local disk.""" if filesystem == 'hf': with hf_fs.open(path, "rb") as f: return f.read().decode("utf-8") else: # local path return Path(path).read_text(encoding="utf-8") def wrap_para(text: str, width: int) -> list[str]: if not text.strip(): return [""] return textwrap.wrap( text.strip().replace("\u00A0", " "), width=width, break_long_words=False, break_on_hyphens=False, drop_whitespace=True, replace_whitespace=True, ) or [""] def md_dialogue_to_visual_lines(md_text: str, width: int) -> list[str]: """Return a list of wrapped visual lines with T:/P: on first line.""" md_text = md_text.replace("\r\n", "\n").replace("\r", "\n").strip("\n") paragraphs = re.split(r"\n\s*\n", md_text) out = [] for p in paragraphs: p = p.strip() if not p: out.append("") continue m = SPEAKER_RE.match(p) if m: speaker, content = m.group(1), m.group(2) wrapped = wrap_para(content, width) out.append(f"{speaker}: {py_html.escape(wrapped[0])}".rstrip()) for w in wrapped[1:]: out.append(py_html.escape(w)) else: for w in wrap_para(p, width): out.append(py_html.escape(w)) out.append("") # blank between paragraphs if out and out[-1] == "": out.pop() return out def render_markdown_simple(md_path, height_px=520): md = read_md(md_path) st.markdown( f"
{markdown(md, extensions=['extra','codehilite'])}
", unsafe_allow_html=True ) def render_dialogue(md_path: str, width_chars: int = 100, height_px: int = 520, font_family: str = "system-ui, -apple-system, 'Segoe UI', Roboto, Helvetica, Arial, 'Noto Sans', sans-serif", font_size: str = "1.05rem", show_border: bool = False): """Render the .md dialogue with line numbers inside an iframe to avoid Streamlit CSS quirks.""" text = read_md(md_path) lines = md_dialogue_to_visual_lines(text, width_chars) border_css = "1px solid #e6e6e6" if show_border else "none" radius_css = ".6rem" if show_border else "0" padding_css = ".8rem 1rem" if show_border else "0" # Build stable HTML (table keeps gutter aligned) rows = "\n".join( f"{i}{ln or ' '}" for i, ln in enumerate(lines, 1) ) html_doc = f"""
{rows}
""" # iframe height = inner height; no external scrollbars flashing st_html(html_doc, height=height_px + (16 if show_border else 0), scrolling=False) def render_rupture_markers_widget(index: int, choices: list[str]): """Dynamic list of (rupture_marker_i, line_a_i, line_b_i) with Add/Remove. Works inside a st.form by using form_submit_button instead of callbacks.""" count_key = f"rupture_count_{index}" if count_key not in st.session_state: st.session_state[count_key] = 1 count = st.session_state[count_key] st.markdown("**What rupture markers are found?** *(one per row; max 8)*") for i in range(1, count + 1): col1, col2, col3 = st.columns([3, 1, 1]) marker_key = f"rupture_marker_{i}_{index}" a_key = f"rupture_marker_{i}_line_a_{index}" b_key = f"rupture_marker_{i}_line_b_{index}" with col1: current = st.session_state.get(marker_key, "") if marker_key in st.session_state: st.selectbox(f"Rupture marker {i}", options=choices, key=marker_key) else: default_idx = choices.index(current) if current in choices else 0 st.selectbox(f"Rupture marker {i}", options=choices, key=marker_key, index=default_idx) # st.selectbox( # f"Rupture marker {i}", options=choices, key=marker_key, # index=(choices.index(st.session_state.get(marker_key)) # if st.session_state.get(marker_key) in choices else 0) # if st.session_state.get(marker_key) is not None else 0 # ) with col2: if RUPTURE_LINE_INPUT_TYPE == "number": st.number_input("Lines: from", min_value=0, step=1, key=a_key) else: st.text_input("Lines: from", key=a_key, value=st.session_state.get(a_key, "")) with col3: if RUPTURE_LINE_INPUT_TYPE == "number": st.number_input("Lines: to", min_value=0, step=1, key=b_key) else: st.text_input("Lines: to", key=b_key, value=st.session_state.get(b_key, "")) st.markdown("---") a_col, b_col = st.columns([1, 1]) with a_col: if st.button("➕ Add another rupture marker", disabled=(count >= MAX_RUPTURE_MARKERS)): st.session_state[count_key] = min(MAX_RUPTURE_MARKERS, count + 1) st.rerun() with b_col: if st.button("➖ Remove last", disabled=(count <= 1)): st.session_state[count_key] = max(1, count - 1) st.rerun() st.caption(f"{st.session_state[count_key]}/{MAX_RUPTURE_MARKERS} markers") def _get_value_for_field(f: Field, index: int): """Return the session value for input fields or None.""" if f.type in INPUT_FIELD_DEFAULT_VALUES: key = f.name + str(index) return st.session_state.get(key, INPUT_FIELD_DEFAULT_VALUES[f.type]) return None def _is_default_value(f: Field, val): """Check if current value equals the default for this widget type.""" return val == INPUT_FIELD_DEFAULT_VALUES.get(f.type) def prime_rupture_markers_from_saved(index: int, data_collected: dict | None): """ Copy saved JSON values (rupture_marker_1, *_line_a, *_line_b) into st.session_state keys used by the widget (…_{index}), and set the count. Runs once per index to avoid clobbering user edits on reruns. """ if not data_collected: return primed_flag = f"rupture_primed_{index}" if st.session_state.get(primed_flag): return # already restored for this page slots = [] for i in range(1, MAX_RUPTURE_MARKERS + 1): m = data_collected.get(f"rupture_marker_{i}") a = data_collected.get(f"rupture_marker_{i}_line_a") b = data_collected.get(f"rupture_marker_{i}_line_b") if any(v is not None for v in (m, a, b)): # widget's selectbox uses "" for "no selection" slots.append(((m or ""), a, b)) if not slots: return # set widget count (at least 1, capped to MAX) count_key = f"rupture_count_{index}" st.session_state[count_key] = min(MAX_RUPTURE_MARKERS, max(1, len(slots))) for i, (m, a, b) in enumerate(slots, start=1): st.session_state[f"rupture_marker_{i}_{index}"] = m if m in rupture_choices else (m or "") st.session_state[f"rupture_marker_{i}_line_a_{index}"] = a st.session_state[f"rupture_marker_{i}_line_b_{index}"] = b st.session_state[primed_flag] = True def validate_current_page(fields: List[Field], index: int) -> bool: """ Walk the field tree and verify mandatory inputs are filled. Honors `following_mandatory_values`: if a field's value is in that list, subsequent siblings become mandatory. """ ok = True def walk(nodes: List[Field], following_required: bool = False): nonlocal ok # local flag that can be turned on by a node for its following siblings req_for_following = following_required for f in nodes: if f.children: # containers/expanders: recurse; pass the current rule downward walk(f.children, req_for_following) continue if f.type in INPUT_FIELD_DEFAULT_VALUES: val = _get_value_for_field(f, index) # decide if THIS field is required required = bool(f.mandatory or req_for_following) # if required and default → error if required and _is_default_value(f, val): ok = False # update the "following required" rule for siblings if f.following_mandatory_values: try: if val in f.following_mandatory_values: req_for_following = True except Exception: pass # structural types (markdown/input_col/etc.) don’t affect validation walk(fields, False) if not ok: st.error("Please fill in all mandatory fields.") return ok #################################### Streamlit App #################################### # Function to navigate rows def navigate(index_change): st.session_state.current_index += index_change # only works consistently if done before rerun js = ''' ''' st.components.v1.html(js, height=0) # https://discuss.streamlit.io/t/click-twice-on-button-for-changing-state/45633/2 # disable text input enter to submit # https://discuss.streamlit.io/t/text-input-how-to-disable-press-enter-to-apply/14457/6 components.html( """ """, height=0 ) st.rerun() def show_field(f: Field, index: int, data_collected): if f.type not in INPUT_FIELD_DEFAULT_VALUES.keys(): st.session_state.following_mandatory = False match f.type: case 'input_col': value = ( st.session_state.data.iloc[index][f.name] if f.name and f.name in st.session_state.data.columns else None ) if f.name == 'image_name' and value: display_image(os.path.join(input_repo_path, 'images', value)) elif f.name == 'dialogue_name' and value: render_dialogue( os.path.join(input_repo_path, 'dialogues', value), width_chars=115, height_px=520, show_border=False ) elif f.name == 'role_name' and value: render_markdown_simple(os.path.join(input_repo_path, 'role_descriptions', value), height_px=520) #, #width_chars=115, height_px=520, show_border=False #) elif f.name == 'patient' and value: st.markdown(f"#### Patient:  {value}") elif value not in (None, np.nan, ""): if f.title: st.write(f.title) st.write(value) case 'markdown': path = f.other_params.get("path") if f.other_params else None if path: content = load_text(os.path.join(input_repo_path, path)) st.markdown(content, unsafe_allow_html=True) else: st.markdown(f.title) case 'expander': with (st.expander(f.title) if f.type == 'expander' else st.container(border=True)): for child in f.children: show_field(child, index, data_collected) case 'container': with st.container(border=True): st.markdown(f.title) for child in (f.children or []): show_field(child, index, data_collected) case 'skip_checkbox': st.checkbox(f.title, key=f.name, value=False) case 'rupture_markers': prime_rupture_markers_from_saved(index, data_collected) render_rupture_markers_widget(index, rupture_choices) return else: key = f.name + str(index) # track the logical field name once per page; prep_and_save_data relies on this st.session_state.data_inputs_keys.append(f.name) default_val = ( data_collected.get(f.name, INPUT_FIELD_DEFAULT_VALUES[f.type]) if data_collected else INPUT_FIELD_DEFAULT_VALUES[f.type] ) value = _ensure_key(key, default_val) #value = st.session_state[key] if key in st.session_state else \ # (data_collected[f.name] if data_collected else INPUT_FIELD_DEFAULT_VALUES[f.type]) if not SHOW_HELP_ICON: f.title = f'**{f.title}**\n\n{f.help}' if f.help else f.title validation_error = False if st.session_state.form_displayed == st.session_state.current_index: if st.session_state.following_mandatory and f.skip_mandatory: st.session_state.following_mandatory = False if f.following_mandatory_values and st.session_state[key] in f.following_mandatory_values: st.session_state.following_mandatory = True if f.mandatory or st.session_state.following_mandatory: if value == INPUT_FIELD_DEFAULT_VALUES[f.type]: #if st.session_state[key] == INPUT_FIELD_DEFAULT_VALUES[f.type]: st.session_state.valid = False validation_error = True elif f.following_mandatory_values and st.session_state[key] in f.following_mandatory_values: st.session_state.following_mandatory = True if f.mandatory or st.session_state.following_mandatory: f.title += " :red[* required!]" if (validation_error and not SHOW_VALIDATION_ERROR_MESSAGE) else' :red[*]' f.help = None value = st.session_state[key] if key in st.session_state else \ (data_collected[f.name] if data_collected else INPUT_FIELD_DEFAULT_VALUES[f.type]) match f.type: case 'checkbox': st.checkbox(f.title, key=key, value=value, help=f.help) case 'radio': labels = (f.other_params.get('labels') if f.other_params and f.other_params.get('labels') else default_labels) st.radio( f.title, options=range(len(labels)), format_func=lambda x: labels[x], key=key, index=value, help=f.help, horizontal=False ) case 'slider': st.slider(f.title, min_value=0, max_value=6, step=1, key=key, value=value, help=f.help) case 'select_slider': labels = (f.other_params.get('labels') if f.other_params and f.other_params.get('labels') else default_labels) st.select_slider( f.title, options=[0, 20, 40, 60, 80, 100], format_func=lambda x: labels[x // 20], key=key, value=value, help=f.help ) case 'multiselect': choices = (f.other_params.get('choices') if f.other_params and f.other_params.get('choices') else default_choices) st.multiselect( f.title, options=choices, format_func=lambda x: x, key=key, default=value, max_selections=3, help=f.help ) case 'likert_radio': labels = (f.other_params.get('labels') if f.other_params and f.other_params.get('labels') else default_labels) st.radio( f.title, options=[0, 1, 2, 3, 4], format_func=lambda x: labels[x], key=key, index=value, help=f.help, horizontal=True ) case 'y_n_radio': labels = (f.other_params.get('labels') if f.other_params and f.other_params.get('labels') else yes_no_labels) st.radio( f.title, options=[0, 1], format_func=lambda x: labels[x], key=key, index=value, help=f.help, horizontal=True ) case 'text': st.text_input(f.title, key=key, value=(value if value is not None else ""), max_chars=None, help=f.help) case 'textarea': st.text_area(f.title, key=key, value=(value if value is not None else ""), max_chars=None, help=f.help) def show_fields(fields: List[Field]): index = st.session_state.current_index data_collected = read_saved_data() st.session_state.data_inputs_keys = [] # rebuilt each render st.session_state.following_mandatory = False for field in fields: show_field(field, index, data_collected) # mark that the page has been rendered at least once (if you still use this elsewhere) st.session_state.form_displayed = st.session_state.current_index def prep_and_save_data(index, skip_sample): payload = { 'user_id': st.session_state.user_id, 'index': st.session_state.current_index, **(st.session_state.data.iloc[index][COLS_TO_SAVE].to_dict() if 0 <= index < len(st.session_state.data) else {}), **{k: st.session_state[k + str(index)] for k in st.session_state.data_inputs_keys}, 'skip': skip_sample } # normalize rupture markers -> always write 8 slots count_key = f"rupture_count_{index}" count = st.session_state.get(count_key, 1) for i in range(1, MAX_RUPTURE_MARKERS + 1): m_key = f"rupture_marker_{i}_{index}" a_key = f"rupture_marker_{i}_line_a_{index}" b_key = f"rupture_marker_{i}_line_b_{index}" # export names WITHOUT the trailing _{index}, so your JSON uses a stable schema: out_m = f"rupture_marker_{i}" out_a = f"rupture_marker_{i}_line_a" out_b = f"rupture_marker_{i}_line_b" if i <= count: marker_val = st.session_state.get(m_key) line_a_val = st.session_state.get(a_key) line_b_val = st.session_state.get(b_key) # Basic sanity: if numbers, coerce order (optional) if RUPTURE_LINE_INPUT_TYPE == "number": try: if line_a_val is not None and line_b_val is not None and line_a_val > line_b_val: line_a_val, line_b_val = line_b_val, line_a_val except Exception: pass payload[out_m] = marker_val if marker_val not in ("", None) else None payload[out_a] = line_a_val if line_a_val not in ("", None) else None payload[out_b] = line_b_val if line_b_val not in ("", None) else None else: payload[out_m] = None payload[out_a] = None payload[out_b] = None save_data(payload) #st.set_page_config(layout='wide') # Title of the app st.title("Therapy Session Annotation Task") st.markdown( """ """, unsafe_allow_html=True) def add_annotation_guidelines(): guidelines_text = load_text("texts/guidelines.md") #st.write(f"username is {st.session_state.user_id}") st.markdown( f"
Annotation Guidelines
{guidelines_text}

", unsafe_allow_html=True) if 'unacceptable_response' in st.session_state and st.session_state.unacceptable_response: #error_message = "You are not eligible for this study.

" error_message = "
" if st.session_state.current_index >= -5: error_message += ( "Thank you for your time! You will receive a small compensation for your contribution up to now.

" f'Please return to the study and copy/paste this code: {failed_sanity_check_code}, or ' f'Click Here' ) # Display the error message using custom HTML st.markdown( f"""

Error: You are not eligible for this study.

{error_message}
""", unsafe_allow_html=True ) st.stop() # Load the data to annotate if 'data' not in st.session_state: st.session_state.data = read_data() # user id user_id_from_url = get_param_from_url("user_id") if user_id_from_url and not st.session_state.get("user_id"): st.session_state.user_id = user_id_from_url # current index if 'current_index' not in st.session_state: start_index = get_start_index() if start_index < len(st.session_state.data)-1: st.session_state.current_index = start_index else: st.session_state.current_index = start_index+1 st.session_state.form_displayed = -2 if get_param_from_url('show_extra_fields'): fields += url_conditional_fields else: fields += url_conditional_fields def add_validated_submit(fields, message): st.session_state.form_displayed = st.session_state.current_index if st.form_submit_button("Submit"): if all(not x for x in fields): st.error(message) else: navigate(1) def add_checked_submit(): check = st.checkbox('I agree', key='consent') add_validated_submit([check], "Please agree to give your consent to proceed") def load_text(path: str) -> str: if filesystem == 'hf': with hf_fs.open(path, "rb") as f: return f.read().decode("utf-8") else: with open(path, "r", encoding="utf-8") as f: return f.read() if st.session_state.current_index == -2: with st.form("data_form"): st.markdown(consent_text) add_checked_submit() elif st.session_state.current_index == -1: if st.session_state.get('user_id'): navigate(1) else: with st.form("user_id_form"): st.session_state.user_id = st.text_input("User ID", value="") add_validated_submit([st.session_state.user_id], "Please enter a valid user ID") elif st.session_state.current_index < len(st.session_state.data): show_fields(fields) # Action buttons c1, c2, c3 = st.columns([1,1,6]) with c1: if st.button("Submit"): if validate_current_page(fields, st.session_state.current_index): with st.spinner("saving"): prep_and_save_data(st.session_state.current_index, ('skip' in st.session_state and st.session_state['skip'])) st.success("Saved!") navigate(1) elif st.session_state.current_index == len(st.session_state.data): st.write(f"**Thank you for taking part in this study!** \n ") if st.button("Previous"): navigate(-2) # Navigation buttons if 0 < st.session_state.current_index < len(st.session_state.data): if st.button("Previous"): navigate(-1) if 0 <= st.session_state.current_index < len(st.session_state.data): st.write(f"Page {st.session_state.current_index + 1} out of {len(st.session_state.data)}") st.markdown( """""", unsafe_allow_html=True )