plexus.reports.blocks package

class plexus.reports.blocks.BaseReportBlock(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)

Bases: ABC

Abstract base class for all report blocks.

Each subclass is responsible for generating a specific section or data point within a report.

DEFAULT_DESCRIPTION: str | None = None
DEFAULT_NAME: str | None = None
__init__(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
attach_detail_file(report_block_id: str, file_name: str, content: bytes, content_type: str | None = None) → str

Attach a detail file to this report block.

This uploads the file to S3 and adds its path to the ReportBlock’s attachedFiles field. The Amplify Gen2 storage expects paths to be stored, not complex objects.

Args:

report_block_id: ID of the report block to attach the file to file_name: Name of the file to create content: Bytes content of the file (changed from str to bytes) content_type: Optional MIME type for the file

Returns:

The S3 path to the file

abstractmethod async generate() → Tuple[Dict[str, Any] | None, str | None]

Generates the data for this report block asynchronously.

Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.

Returns:
A tuple containing:

  • A dictionary containing the generated data (JSON-serializable), or None on failure.

  • A string containing concatenated log messages, or None if no logs.

get_resolved_dataset_id() → str | None

Get the resolved dataset ID that this block actually used.

Returns:

The dataset ID if one was resolved, None otherwise

set_resolved_dataset_id(dataset_id: str)

Set the resolved dataset ID that this block actually used.

This should be called by blocks after they have resolved and used a dataset, to record which specific dataset was actually processed.

Args:

dataset_id: The ID of the dataset that was resolved and used

class plexus.reports.blocks.CostAnalysis(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)

Bases: BaseReportBlock

ReportBlock: Cost analysis over ScoreResults with summary or detail outputs.

Config:

  • scorecard (optional): id/key/name/externalId (any identifier)

  • score (optional): id/key/name/externalId (resolved within scorecard when provided)

  • hours (int, default 1): preferred time window

  • days (int, default 0): used when hours not provided

  • group_by (str|None): None | ‘scorecard’ | ‘score’ | ‘scorecard_score’

  • mode (str): ‘summary’ (default) or ‘detail’

  • breakdown (bool): when summary mode, include grouped summaries

DEFAULT_DESCRIPTION: str | None = 'ScoreResult cost metrics and breakdowns'
DEFAULT_NAME: str | None = 'Cost Analysis'
async generate() → Tuple[Dict[str, Any] | None, str | None]

Generates the data for this report block asynchronously.

Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.

Returns:
A tuple containing:

  • A dictionary containing the generated data (JSON-serializable), or None on failure.

  • A string containing concatenated log messages, or None if no logs.

class plexus.reports.blocks.FeedbackAnalysis(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)

Bases: BaseReportBlock

Analyzes feedback data using Gwet’s AC1 agreement coefficient.

This block retrieves FeedbackItem records and compares initial and final answer values to calculate agreement scores using Gwet’s AC1.

If a specific ‘score_id’ is provided in the config, it analyzes only that score. Otherwise, it analyzes all scores associated with the provided ‘scorecard’ that have a mapping to a Plexus Score with an externalId.

Config:

scorecard (str): Scorecard identifier. This is REQUIRED. days (int, optional): Number of days in the past to analyze (default: 14).

FeedbackItems updated within this period will be considered.

start_date (str, optional): Start date for analysis in YYYY-MM-DD format.

If provided, overrides ‘days’.

end_date (str, optional): End date for analysis in YYYY-MM-DD format.

Defaults to today if not specified.

score_id (str, optional): Specific score ID to analyze.

If specified, only this score will be analyzed.

DEFAULT_DESCRIPTION: str | None = 'Inter-rater Reliability Assessment'
DEFAULT_NAME: str | None = 'Feedback Analysis'
async generate() → Tuple[Dict[str, Any] | None, str | None]

Fetches feedback data and performs agreement analysis.

class plexus.reports.blocks.ScoreInfo(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)

Bases: BaseReportBlock

Generates a block of information about a specific Score.

Config:
score (str): The ID or name of the Score to display.

(Note: Current mock implementation uses this as ID)

include_variant (bool, optional): Whether to include variant details.

Defaults to False.

async generate() → Tuple[Dict[str, Any] | None, str | None]

Fetches (mock) Score data and returns it with logs.

class plexus.reports.blocks.TopicAnalysis(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)

Bases: BaseReportBlock

DEFAULT_NAME: str | None = 'Topic Analysis'
DEFAULT_PROMPT = '\n    I have a topic from call center transcripts that is described by the following keywords: [KEYWORDS]\n    In this topic, these customer-agent conversations are representative examples:\n    [DOCUMENTS]\n\n    Based on the keywords and representative examples above, provide a short, descriptive label for this topic in customer service context. Return only the label, no other text or formatting.\n    '
__init__(config: Dict[str, Any], params: Dict[str, Any] | None, api_client: PlexusDashboardClient)
async generate() → Tuple[Dict[str, Any] | None, str | None]

Generates the data for this report block asynchronously.

Access block configuration via self.config and report parameters via self.params. Use self.api_client for data fetching. Use self._log(“message”) to record log information.

Returns:
A tuple containing:

  • A dictionary containing the generated data (JSON-serializable), or None on failure.

  • A string containing concatenated log messages, or None if no logs.

Submodules