Create app.py

app.py

@@ -0,0 +1,277 @@
+import json
+import os
+from dataclasses import dataclass
+from typing import List, Optional
+from PIL import Image
+
+import pandas as pd
+import streamlit as st
+from huggingface_hub import HfFileSystem
+
+
+@dataclass
+class Field:
+    type: str
+    title: str
+    name: str = None
+    help: Optional[str] = None
+    children: Optional[List['Field']] = None
+
+
+# Function to get user ID from URL
+def get_user_id_from_url():
+    user_id = st.query_params.get("user_id", "")
+    return user_id
+
+
+HF_TOKEN = os.environ.get("HF_TOKEN_WRITE")
+print("is none?", HF_TOKEN is None)
+hf_fs = HfFileSystem(token=HF_TOKEN)
+input_repo_path = 'datasets/emvecchi/annotate-pilot'
+output_repo_path = 'datasets/emvecchi/annotate-pilot'
+to_annotate_file_name = 'to_annotate.csv'  # CSV file to annotate
+COLS_TO_SAVE = ['comment_id']
+
+fields: List[Field] = [
+    Field(name="topic", type="input_col", title="**Topic:**"),
+    Field(name="parent_comment", type="input_col", title="**Preceeding Comment:**"),
+    Field(name="comment", type="input_col", title="**Comment:**"),
+    Field(name="image_name", type="input_col", title="**Visualization of high contributing properties:**"),
+
+    #Field(type="container", title="", children=[
+    #    Field(name="reply_fitting", type="slider",
+    #               title="Is the reply comment **fitting and does it make sense** as a reply to the preceding comment?"),
+    #    Field(name="actions_clear", type="slider",
+    #               title="**Actions clarity**: Does the reply make it clear what actions need to be done?"),
+    #]),
+    #Field(type="expander",
+    #           title="Expand and fill-out this section if you see **issues in the original comment**",
+    #           children=[
+    #                   Field(name="issues_in_comment", type="slider",
+    #                              title="Do **you see some issues** in the original comment?"),
+    #                   Field(name="moderator_spotted", type="slider",
+    #                              title="Based on the reply, has the **moderator spotted the issues** in the original comment?"),
+    #                   Field(name="reply_addresses_issues", type="slider",
+    #                              title="How well does the reply **address those issues**?")
+    #               ]),
+
+    #Field(type="container", title="**Score the following properties of the moderator comment?**", children=[
+    #    Field(name="neutrality", type="slider", title="Neutrality",
+    #               help='Remain Neutral on the topic and on the Comment Substance and Commenter’s Viewpoint. The reply shouldn’t give away the opinion of the moderator on the topic or comment. '),
+    #    # FieldDict(name="attitude", type="slider", title="Attitude", help=''),
+    #    Field(name="clarity", type="slider", title="Clarity",
+    #               help="Plain language, simple, clear, avoid overwhelming the user e.g. too many questions"),
+    #    Field(name="curiosity", type="slider", title="Curiosity",
+    #               help="Moderators should model a spirit of inquiry and a desire to learn from and understand commenter’s experience and views. Try to be interested in the bases upon which each commenter stakes his or her claims and the lines of reasoning that has led each commenter to those particular conclusions."),
+    #    # TODO
+    #    Field(name="bias", type="slider", title="Bias",
+    #               help="Does the reply show some biases towards the commenter? Are there stereotypes or prejudices?"),
+    #    Field(name="encouraging", type="slider", title="Encouraging",
+    #               help="Welcoming, encouraging and acknowledging. Avoid Evaluative and/or Condescending Responses"),
+    #]),
+    #
+    Field(name="other_comments", type="text", title="Further comments: free text"),
+]
+INPUT_FIELD_DEFAULT_VALUES = {'slider': 0,
+                              'text': None,
+                              'textarea': None,
+                              'checkbox': False}
+SHOW_HELP_ICON = False
+
+def read_data(_path):
+    with hf_fs.open(input_repo_path + '/' + _path) 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):
+    hf_fs.mkdir(f"{output_repo_path}/{data['user_id']}")
+    with hf_fs.open(f"{output_repo_path}/{get_path()}", "w") as f:
+        f.write(json.dumps(data))
+
+
+def get_path():
+    return f"{st.session_state.user_id}/{st.session_state.current_index}.json"
+
+
+#def display_image(image_path):
+#    st.image(image_path, caption='Visualization of high contributing properties', use_column_width=True)
+
+def display_image(image_path):
+    with hf_fs.open(image_path) as f:
+        img = Image.open(f)
+        st.image(img, caption='Visualization of high contributing properties', use_column_width=True)
+        
+#################################### Streamlit App ####################################
+
+# Function to navigate rows
+def navigate(index_change):
+    st.session_state.current_index += index_change
+    print(st.session_state.current_index)
+    # https://discuss.streamlit.io/t/click-twice-on-button-for-changing-state/45633/2
+    st.rerun()
+
+
+def show_field(f: Field, index: int):
+    if f.type not in INPUT_FIELD_DEFAULT_VALUES.keys():
+        match f.type:
+            case 'input_col':
+                st.write(f.title)
+                if f.name == 'image_name':
+                    st.write(f.title)
+                    image_name = st.session_state.data.iloc[index][f.name]
+                    if image_name:  # Ensure the image name is not empty
+                        image_path = os.path.join(input_repo_path, 'images', image_name)
+                        display_image(image_path)
+                else:
+                    st.write(st.session_state.data.iloc[index][f.name])
+            case 'markdown':
+                st.markdown(f.title)
+            case 'expander' | 'container':
+                with (st.expander(f.title) if f.type == 'expander' else st.container(border=True)):
+                    if f.type == 'container':
+                        st.markdown(f.title)
+                    for child in f.children:
+                        show_field(child, index)
+    else:
+        key = f.name + str(index)
+        value = st.session_state.default_values[f.name] = 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
+            f.help = None
+        match f.type:
+            case 'checkbox':
+                st.session_state.data_inputs[f.name] = st.checkbox(f.title,
+                                                                   key=key,
+                                                                   value=value, help=f.help)
+            case 'slider':
+                st.session_state.data_inputs[f.name] = st.slider(f.title,
+                                                                 min_value=0, max_value=100, step=25,
+                                                                 key=key,
+                                                                 value=value, help=f.help)
+            case 'text':
+                st.session_state.data_inputs[f.name] = st.text_input(f.title, key=key, value=value)
+            case 'textarea':
+                st.session_state.data_inputs[f.name] = st.text_area(f.title, key=key, value=value)
+
+
+
+
+# st.set_page_config(layout='wide')
+# Title of the app
+st.title("Moderation Prediction")
+
+st.markdown(
+    """<style>
+div[data-testid="stMarkdownContainer"] > p {
+    font-size: 1rem;
+}
+    </style>
+    """, unsafe_allow_html=True)
+
+st.markdown(
+    """<details open>
+  <summary>Annotation Guidelines</summary>
+  
+  some guidelines here
+</details>
+    """, unsafe_allow_html=True)
+
+# Load the data to annotate
+if 'data' not in st.session_state:
+    st.session_state.data = read_data(to_annotate_file_name)
+
+# Initialize the current index
+if 'current_index' not in st.session_state:
+    st.session_state.current_index = -1
+
+if st.session_state.current_index == -1:
+    user_id_from_url = get_user_id_from_url()
+    if user_id_from_url:
+        st.session_state.user_id = user_id_from_url
+        navigate(1)
+    else:
+        st.session_state.user_id = st.text_input('Please enter your user ID to proceed', value=user_id_from_url)
+        if st.button("Next"):
+            navigate(1)
+
+elif st.session_state.current_index < len(st.session_state.data):
+    st.write(f"username is {st.session_state.user_id}")
+
+    # Creating the form
+    with st.form("feedback_form"):
+        index = st.session_state.current_index
+        data_collected = read_saved_data()
+        st.session_state.default_values = {}
+        st.session_state.data_inputs = {}
+
+        for field in fields:
+            if field.name not in st.session_state.data.columns:
+                # Field doesn't exist in input dataframe, add it with a default value
+                st.session_state.data_inputs[field.name] = None
+                show_field(field, index)
+            else:
+                show_field(field, index)
+
+        submitted = st.form_submit_button("Submit")
+        if submitted:
+            with st.spinner(text="saving"):
+                save_data({
+                    'user_id': st.session_state.user_id,
+                    'index': st.session_state.current_index,
+                    **st.session_state.data.iloc[index][COLS_TO_SAVE].to_dict(),
+                    **st.session_state.data_inputs
+                })
+            st.success("Feedback submitted successfully!")
+            navigate(1)
+
+else:
+    st.write("Finished all data points!")
+
+# Navigation buttons
+if st.session_state.current_index > 0:
+    if st.button("Previous"):
+        with st.spinner(text="in progress"):
+            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)}")
+
+# disable text input enter to submit
+# https://discuss.streamlit.io/t/text-input-how-to-disable-press-enter-to-apply/14457/6
+import streamlit.components.v1 as components
+
+components.html(
+    """
+    <script>
+    const inputs = window.parent.document.querySelectorAll('input');
+    inputs.forEach(input => {
+        input.addEventListener('keydown', function(event) {
+            if (event.key === 'Enter') {
+                event.preventDefault();
+            }
+        });
+    });
+    </script>
+    """,
+    height=0
+)
+st.markdown(
+    """<style>
+    div[data-testid="InputInstructions"] {
+        visibility: hidden;
+    }
+    </style>""", unsafe_allow_html=True
+)
+