App

app

---

Assembly:                Chonky
Filename:                app.py
Author:                  Terry D. Eppler
Created:                 05-31-2022

Last Modified By:        Terry D. Eppler
Last Modified On:        05-01-2025

---

       app.py
       Copyright ©  2022  Terry Eppler

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

You can contact me at: terryeppler@gmail.com or eppler.terry@epa.gov

Provides the Streamlit application shell for the Chonky document-processing workflow. Purpose: Coordinates Chonky's staged interface for document loading, text processing, semantic analysis, tokenization diagnostics, hosted and local embedding generation, vector persistence, and similarity retrieval. The module initializes Streamlit session state, configures the page, wires loader, processor, and embedder wrappers into tabbed UI sections, and preserves shared workflow outputs for downstream tabs.

---

style_subheaders

style_subheaders() -> None

Style Streamlit subheaders.

Purpose

Injects the CSS used to keep Chonky's secondary headings visually aligned with the application accent color. The helper affects Streamlit markdown containers and chat message headings without changing application state or downstream workflow data.

Source code in app.py

def style_subheaders( ) -> None:
	"""Style Streamlit subheaders.

	Purpose:
		Injects the CSS used to keep Chonky's secondary headings visually aligned with the
		application accent color. The helper affects Streamlit markdown containers and chat
		message headings without changing application state or downstream workflow data.
	"""
	st.markdown(
		"""
		<style>
		div[data-testid="stMarkdownContainer"] h2,
		div[data-testid="stMarkdownContainer"] h3,
		div[data-testid="stChatMessage"] div[data-testid="stMarkdownContainer"] h2,
		div[data-testid="stChatMessage"] div[data-testid="stMarkdownContainer"] h3 {
			color: rgb(0, 120, 252) !important;
		}
		</style>
		""",
		unsafe_allow_html=True,
	)

encode_image_base64

encode_image_base64(path: str) -> str

Encode an image file as base64 text.

Purpose

Reads an image from disk and converts its bytes into a base64 string for Streamlit HTML or markdown rendering where inline image content is required.

Parameters:

Name	Type	Description	Default
path	str	Filesystem path to the image file.	required

Returns:

Name	Type	Description
str	str	Base64-encoded image content.

Source code in app.py

def encode_image_base64( path: str ) -> str:
	"""Encode an image file as base64 text.

	Purpose:
		Reads an image from disk and converts its bytes into a base64 string for Streamlit
		HTML or markdown rendering where inline image content is required.

	Args:
		path: Filesystem path to the image file.

	Returns:
		str: Base64-encoded image content.
	"""
	data = Path( path ).read_bytes( )
	return base64.b64encode( data ).decode( "utf-8" )

clear_if_active

clear_if_active(loader_name: str) -> None

Clear loader-specific session state when a loader is active.

Purpose

Resets document, token, chunking, embedding, dataframe, and table state only when the requested loader is the active loader. The helper protects unrelated workflow state while allowing individual loader panels to clear their own downstream outputs.

Parameters:

Name	Type	Description	Default
loader_name	str	Name of the loader whose state should be cleared when active.	required

Source code in app.py

def clear_if_active( loader_name: str ) -> None:
	"""Clear loader-specific session state when a loader is active.

	Purpose:
		Resets document, token, chunking, embedding, dataframe, and table state only when the
		requested loader is the active loader. The helper protects unrelated workflow state while
		allowing individual loader panels to clear their own downstream outputs.

	Args:
		loader_name: Name of the loader whose state should be cleared when active.
	"""
	if st.session_state.active_loader == loader_name:
		st.session_state.documents = None
		st.session_state.active_loader = None
		st.session_state.tokens = None
		st.session_state.vocabulary = None
		st.session_state.token_counts = None
		st.session_state.chunks = None
		st.session_state.chunk_modes = None
		st.session_state.chunked_documents = None
		st.session_state.embeddings = None
		st.session_state.active_table = None
		st.session_state.df_frequency = None
		st.session_state.df_tables = None
		st.session_state.df_schema = None
		st.session_state.df_preview = None
		st.session_state.df_count = None
		st.session_state.df_chunks = None
		st.session_state.lines = None

metric_with_tooltip

metric_with_tooltip(label: str, value: str, tooltip: str)

Render a Streamlit metric with optional hover guidance.

Purpose

Displays a metric value beside an informational tooltip icon for Chonky analysis panels. The helper keeps metric layout consistent while suppressing tooltip icons for core text length metrics that do not require additional explanation.

Parameters:

Name	Type	Description	Default
label	str	Metric label displayed by Streamlit.	required
value	str	Metric value displayed by Streamlit.	required
tooltip	str	Tooltip text rendered in the hover title attribute.	required

Source code in app.py

def metric_with_tooltip( label: str, value: str, tooltip: str ):
	"""Render a Streamlit metric with optional hover guidance.

	Purpose:
		Displays a metric value beside an informational tooltip icon for Chonky analysis panels.
		The helper keeps metric layout consistent while suppressing tooltip icons for core text
		length metrics that do not require additional explanation.

	Args:
		label: Metric label displayed by Streamlit.
		value: Metric value displayed by Streamlit.
		tooltip: Tooltip text rendered in the hover title attribute.
	"""
	col_metric, col_info = st.columns( [ 0.5, 0.5 ] )

	with col_metric:
		st.metric( label, value )

	with col_info:
		if label not in [ 'Characters', 'Tokens', 'Unique Tokens', 'Avg Length' ]:
			st.markdown(
				f"""
	            <span style="
	                cursor: help;
	                font-size: 0.85rem;
	                color:#888;
	                vertical-align: super;
	            " title="{tooltip}">ℹ️ </span>
	            """,
				unsafe_allow_html=True,
			)

normalize_embeddings

normalize_embeddings(emb_array: ndarray) -> np.ndarray

Normalize an embedding array to two dimensions.

Purpose

Converts a single one-dimensional embedding vector into a row-shaped array so Chonky embedding diagnostics, dimensionality reduction, and vector persistence paths can operate on a consistent matrix-like structure.

Parameters:

Name	Type	Description	Default
emb_array	ndarray	NumPy embedding array to normalize.	required

Returns:

Type	Description
ndarray	np.ndarray: Original array or reshaped two-dimensional embedding array.

Source code in app.py

def normalize_embeddings( emb_array: np.ndarray ) -> np.ndarray:
	"""Normalize an embedding array to two dimensions.

	Purpose:
		Converts a single one-dimensional embedding vector into a row-shaped array so Chonky
		embedding diagnostics, dimensionality reduction, and vector persistence paths can operate
		on a consistent matrix-like structure.

	Args:
		emb_array: NumPy embedding array to normalize.

	Returns:
		np.ndarray: Original array or reshaped two-dimensional embedding array.
	"""
	if isinstance( emb_array, np.ndarray ) and emb_array.ndim == 1:
		return emb_array.reshape( 1, -1 )
	return emb_array

rebuild_raw_text_from_documents

rebuild_raw_text_from_documents() -> str | None

Rebuild raw text from loaded documents.

Purpose

Reconstructs the shared raw-text buffer from the current Streamlit documents state after loader-specific clear operations. The helper preserves downstream processing compatibility by joining non-empty page_content values from remaining LangChain documents.

Returns:

Type	Description
str | None	str | None: Rebuilt raw text, or None when no usable document text remains.

Source code in app.py

def rebuild_raw_text_from_documents( ) -> str | None:
	"""Rebuild raw text from loaded documents.

	Purpose:
		Reconstructs the shared raw-text buffer from the current Streamlit ``documents`` state after
		loader-specific clear operations. The helper preserves downstream processing compatibility
		by joining non-empty ``page_content`` values from remaining LangChain documents.

	Returns:
		str | None: Rebuilt raw text, or ``None`` when no usable document text remains.
	"""
	docs = st.session_state.get( "documents" ) or [ ]
	if not docs:
		return None
	text = '\n\n'.join( d.page_content for d in docs
	                    if hasattr( d, 'page_content' )
	                    and isinstance( d.page_content, str )
	                    and d.page_content.strip( ) )
	return text if text.strip( ) else None

chunk_characters

chunk_characters(
    text: str, size: int, overlap: int
) -> List[str]

Chunk text by character windows.

Purpose

Splits source text into overlapping character-based chunks for semantic analysis, embedding preparation, and display workflows. The function clamps chunk size and overlap to safe values before producing non-empty chunk strings.

Parameters:

Name	Type	Description	Default
text	str	Source text to chunk.	required
size	int	Maximum number of characters in each chunk.	required
overlap	int	Number of overlapping characters between adjacent chunks.	required

Returns:

Type	Description
List[str]	List[str]: Character-based chunk strings.

Source code in app.py

def chunk_characters( text: str, size: int, overlap: int ) -> List[ str ]:
	"""Chunk text by character windows.

	Purpose:
		Splits source text into overlapping character-based chunks for semantic analysis,
		embedding preparation, and display workflows. The function clamps chunk size and overlap
		to safe values before producing non-empty chunk strings.

	Args:
		text: Source text to chunk.
		size: Maximum number of characters in each chunk.
		overlap: Number of overlapping characters between adjacent chunks.

	Returns:
		List[str]: Character-based chunk strings.
	"""
	if not isinstance( text, str ) or not text.strip( ):
		return [ ]

	size = int( size )
	overlap = int( overlap )
	size = max( 1, size )
	overlap = max( 0, min( overlap, size - 1 ) )

	step = size - overlap
	chunks: list[ str ] = [ ]

	i = 0
	n = len( text )
	while i < n:
		part = text[ i: i + size ].strip( )
		if part:
			chunks.append( part )
		i += step

	return chunks

chunk_tokens

chunk_tokens(
    tokens: list[str], size: int, overlap: int
) -> List[str]

Chunk tokens by overlapped token windows.

Purpose

Splits a token list into overlapping windows and rejoins each window into a string for Chonky tokenization, semantic-analysis, and embedding-readiness workflows. The function validates list input and clamps chunk parameters to safe values.

Parameters:

Name	Type	Description	Default
tokens	list[str]	Token values to chunk.	required
size	int	Maximum number of tokens in each chunk.	required
overlap	int	Number of overlapping tokens between adjacent chunks.	required

Returns:

Type	Description
List[str]	List[str]: Token-window chunk strings.

Source code in app.py

def chunk_tokens( tokens: list[ str ], size: int, overlap: int ) -> List[ str ]:
	"""Chunk tokens by overlapped token windows.

	Purpose:
		Splits a token list into overlapping windows and rejoins each window into a string for
		Chonky tokenization, semantic-analysis, and embedding-readiness workflows. The function
		validates list input and clamps chunk parameters to safe values.

	Args:
		tokens: Token values to chunk.
		size: Maximum number of tokens in each chunk.
		overlap: Number of overlapping tokens between adjacent chunks.

	Returns:
		List[str]: Token-window chunk strings.
	"""
	if not isinstance( tokens, list ) or not tokens:
		return [ ]

	size = int( size )
	overlap = int( overlap )
	size = max( 1, size )
	overlap = max( 0, min( overlap, size - 1 ) )

	step = size - overlap
	out: list[ str ] = [ ]

	i = 0
	n = len( tokens )
	while i < n:
		window = tokens[ i: i + size ]
		s = " ".join( t for t in window if isinstance( t, str ) and t.strip( ) ).strip( )
		if s:
			out.append( s )
		i += step

	return out

rebuild_token_text

rebuild_token_text(tokens: List[str]) -> str

Rebuild display text from token values.

Purpose

Reconstructs readable text from tokenized output while avoiding spaces before common sentence delimiters and normalizing punctuation spacing. The helper supports Chonky's tokenization and processed-text display panels.

Parameters:

Name	Type	Description	Default
tokens	List[str]	Token values produced by tokenization, stemming, or lemmatization.	required

Returns:

Name	Type	Description
str	str	Reconstructed display text with normalized punctuation spacing.

Source code in app.py

def rebuild_token_text( tokens: List[ str ] ) -> str:
	"""Rebuild display text from token values.

	Purpose:
		Reconstructs readable text from tokenized output while avoiding spaces before common
		sentence delimiters and normalizing punctuation spacing. The helper supports Chonky's
		tokenization and processed-text display panels.

	Args:
		tokens: Token values produced by tokenization, stemming, or lemmatization.

	Returns:
		str: Reconstructed display text with normalized punctuation spacing.
	"""
	if not isinstance( tokens, list ) or not tokens:
		return ''

	delimiters = { '.', '?', '!', ';', ':', ',', ')' }
	openers = { '(', '[', '{' }
	parts: List[ str ] = [ ]

	for token in tokens:
		if not isinstance( token, str ):
			continue

		value = token.strip( )
		if not value:
			continue

		if not parts:
			parts.append( value )
		elif value in delimiters:
			parts[ -1 ] = f'{parts[ -1 ]}{value}'
		elif parts[ -1 ] in openers:
			parts.append( value )
		else:
			parts.append( f' {value}' )

	text = ''.join( parts )
	text = re.sub( r'\s+', ' ', text ).strip( )
	text = re.sub( r'\s+([.!?;:,)])', r'\1', text )
	text = re.sub( r'([.!?;:])(?=\w)', r'\1 ', text )
	return text

refresh_processing_display

refresh_processing_display() -> None

Purpose:

Increment the processing display version so Streamlit recreates the read-only text-area widgets after processing, reset, or clear actions.

Parameters:

None

Returns:

None

Source code in app.py

def refresh_processing_display( ) -> None:
	"""

		Purpose:
		--------
		Increment the processing display version so Streamlit recreates the read-only
		text-area widgets after processing, reset, or clear actions.

		Parameters:
		-----------
		None

		Returns:
		--------
		None

	"""
	st.session_state.processing_display_version = (
			int( st.session_state.get( 'processing_display_version', 0 ) ) + 1)

coerce_text

coerce_text(value: object) -> str

Purpose:

Convert parser output into a display-safe string without allowing None values to break later processing steps.

Parameters:

value : object The value returned by a processing method.

Returns:

str The string representation of the value, or an empty string.

Source code in app.py

def coerce_text( value: object ) -> str:
	"""

		Purpose:
		--------
		Convert parser output into a display-safe string without allowing None values
		to break later processing steps.

		Parameters:
		-----------
		value : object
			The value returned by a processing method.

		Returns:
		--------
		str
			The string representation of the value, or an empty string.

	"""
	if value is None:
		return ''

	if isinstance( value, str ):
		return value

	if isinstance( value, list ):
		return '\n'.join(
			str( item ) for item in value
			if item is not None and str( item ).strip( ) )

	return str( value )

pad_or_trim_row

pad_or_trim_row(row: list, size: int) -> list

Purpose:

Normalizes a token row to a fixed width by trimming or right-padding with blanks.

Parameters:

row : list The token row to normalize.

int

The required fixed width.

Returns:

list A list of length == size.

Source code in app.py

def pad_or_trim_row( row: list, size: int ) -> list:
	"""
		Purpose:
		--------
		Normalizes a token row to a fixed width by trimming or right-padding with blanks.

		Parameters:
		-----------
		row : list
			The token row to normalize.

		size : int
			The required fixed width.

		Returns:
		--------
		list
			A list of length == size.
	"""
	if not isinstance( row, list ):
		row = [ ]

	if len( row ) >= size:
		return row[ :size ]

	return row + ([ '' ] * (size - len( row )))

project_chunks_for_embedding

project_chunks_for_embedding(chunks: list) -> list[str]

Purpose:

Project chunked inputs into a clean list of embedding-ready strings.

Guarantees:

• Output is strictly List[str]
• No None, type, NaN, or non-string objects
• No empty or whitespace-only strings

Parameters:

chunks : list Input chunks (token grids, strings, or mixed structures)

Returns:

list[str] Clean embedding documents

Source code in app.py

def project_chunks_for_embedding( chunks: list ) -> list[ str ]:
	"""

		Purpose:
		--------
		Project chunked inputs into a clean list of embedding-ready strings.

		Guarantees:
		-----------
		- Output is strictly List[str]
		- No None, type, NaN, or non-string objects
		- No empty or whitespace-only strings

		Parameters:
		-----------
		chunks : list
			Input chunks (token grids, strings, or mixed structures)

		Returns:
		--------
		list[str]
			Clean embedding documents

	"""
	texts = [ ]
	if not isinstance( chunks, list ):
		return texts
	for c in chunks:
		if isinstance( c, list ):
			tokens = [ t for t in c if isinstance( t, str ) and t.strip( ) ]
			if tokens is not None:
				texts.append( " ".join( tokens ) )
		elif isinstance( c, str ):
			if c.strip( ):
				texts.append( c.strip( ) )
	return texts