core

Shared utilities used across the dashboard: configuration, graph layout, colour schemes, database abstraction, and data processing.

exception fstg_toolkit.app.core.config.NotConfiguredError(name: str)

Bases: Exception

Raised when a required configuration attribute has not been set.

__init__(name: str)

Initialise with the name of the missing configuration attribute.

Parameters:

name (str) – The name of the attribute that is not configured.

class fstg_toolkit.app.core.geometry.Arc(begin: float, end: float)

Bases: object

An arc on the unit circle defined by start and end angles (in radians).

Parameters:

• begin (float) – Start angle in radians.

• end (float) – End angle in radians.

property angle: float

Angular span of the arc in radians (end - begin).

begin: float

end: float

static from_proportions(proportions: list[float], begin: float = 0, length: float = 6.283185307179586, gap_size: float = 0.005) → list[Arc]

Create a list of arcs from proportional sizes with uniform gaps.

Parameters:

• proportions (list[float]) – Relative sizes of each arc. Need not sum to 1.

• begin (float, optional) – Starting angle in radians (default 0).

• length (float, optional) – Total angular length to distribute (default 2π).

• gap_size (float, optional) – Gap between arcs as a fraction of the full circle (default 0.005).

Returns:

One Arc per proportion, laid out consecutively with gaps.

Return type:

list[Arc]

sample(radius: float, angular_res: float = 0.04908738521234052) → ndarray

Sample points along the arc at the given radius.

Parameters:

• radius (float) – Radial distance from the origin.

• angular_res (float, optional) – Angular resolution between samples (default π/64).

Returns:

Complex-valued array of sampled points (real=x, imag=y).

Return type:

numpy.ndarray

class fstg_toolkit.app.core.geometry.ArcShape(arc: Arc, thickness: float, radius: float)

Bases: Shape

A filled arc shape with a given radius and thickness.

Parameters:

• arc (Arc) – The angular extent of the shape.

• thickness (float) – Radial thickness of the filled band.

• radius (float) – Inner radius of the shape.

arc: Arc

property exterior_edge: ndarray

Sampled points along the outer arc edge.

property interior_edge: ndarray

Sampled points along the inner arc edge.

radius: float

thickness: float

to_path() → Path

Convert to a closed Path by joining exterior and reversed interior edges.

class fstg_toolkit.app.core.geometry.Line(length: float, orientation: float, origin: tuple[float, float] = (0, 0))

Bases: object

A straight line segment defined by a length, orientation, and origin.

Parameters:

• length (float) – Length of the segment.

• orientation (float) – Angle of the segment in radians (measured from the positive x-axis).

• origin (tuple[float, float], optional) – Start point of the segment (default (0, 0)).

static from_proportions(proportions: list[float], total_length: float, orientation: float, origin: tuple[float, float] = (0, 0), gap_length: float = 0.25) → list[Line]

Create proportionally-sized line segments laid end-to-end with gaps.

Parameters:

• proportions (list[float]) – Relative sizes of each segment.

• total_length (float) – Total available length (gaps included).

• orientation (float) – Shared orientation angle in radians.

• origin (tuple[float, float], optional) – Starting point for the first segment (default (0, 0)).

• gap_length (float, optional) – Fixed gap length between consecutive segments (default 0.25).

Returns:

One Line per proportion, laid out consecutively with gaps.

Return type:

list[Line]

length: float

orientation: float

origin: tuple[float, float] = (0, 0)

sample(offset: float = 0) → ndarray

Sample the two endpoints of the line, optionally perpendicular-shifted.

Parameters:

offset (float, optional) – Perpendicular offset from the centreline (default 0).

Returns:

Array of shape (2, 2) containing the start and end points.

Return type:

numpy.ndarray

property terminus: tuple[float, float]

End point of the line segment computed from length and orientation.

class fstg_toolkit.app.core.geometry.LineShape(line: Line, thickness: float)

Bases: Shape

A filled rectangular shape centred on a line segment.

Parameters:

• line (Line) – The centreline of the shape.

• thickness (float) – Total perpendicular width of the shape.

property left_edge: ndarray

Sampled start and end points of the left edge (negative offset side).

line: Line

property right_edge: ndarray

Sampled start and end points of the right edge (positive offset side).

thickness: float

to_path() → Path

Convert to a closed Path by joining left and reversed right edges.

class fstg_toolkit.app.core.geometry.Path(points: list[tuple[float, float]] = <factory>)

Bases: object

A closed polygonal path defined by a sequence of 2-D points.

Parameters:

points (list[tuple[float, float]], optional) – Ordered list of (x, y) vertices (default empty list).

static from_components(x: list[float], y: list[float]) → Path

Create a Path from separate x and y coordinate lists.

Parameters:

• x (list[float]) – x-coordinates of the vertices.

• y (list[float]) – y-coordinates of the vertices.

Returns:

A Path whose points are zip(x, y).

Return type:

Path

points: list[tuple[float, float]]

to_svg() → str

Serialise the path as an SVG path d attribute string.

Returns:

An SVG path string starting with M, joined by L segments, and closed with Z.

Return type:

str

class fstg_toolkit.app.core.geometry.Ribbon(begin: float, end: float)

Bases: object

A curved ribbon connecting two angular positions on the unit circle via a quadratic Bezier.

Parameters:

• begin (float) – Angle (radians) of the ribbon’s start point on the circle.

• end (float) – Angle (radians) of the ribbon’s end point on the circle.

begin: float

end: float

sample(radius: float = 1.0, origin: tuple[float, float] = (0, 0), strength: float = 0.3, nb_points: int = 50) → ndarray

Sample points along the ribbon curve.

Parameters:

• radius (float, optional) – Circle radius at which the ribbon endpoints sit (default 1.0).

• origin (tuple[float, float], optional) – Centre of the circle (default (0, 0)).

• strength (float, optional) – Pull strength of the Bezier control point towards the origin (default 0.3).

• nb_points (int, optional) – Number of sample points along the curve (default 50).

Returns:

Array of shape (2, nb_points) with x- and y-coordinates.

Return type:

numpy.ndarray

class fstg_toolkit.app.core.geometry.RibbonShape(left_ribbon: Ribbon, right_ribbon: Ribbon, radius: float, strength: float)

Bases: Shape

A filled ribbon bounded by two Bézier curves and two arc caps.

Parameters:

• left_ribbon (Ribbon) – The left boundary ribbon (connects arc_source.begin to arc_target.begin).

• right_ribbon (Ribbon) – The right boundary ribbon (connects arc_source.end to arc_target.end).

• radius (float) – Circle radius for the ribbon endpoints.

• strength (float) – Bezier control-point strength for both ribbons.

left_ribbon: Ribbon

radius: float

right_ribbon: Ribbon

strength: float

to_path() → Path

Convert to a closed Path by joining both ribbon curves and arc caps.

The path traces: left ribbon (Bezier) → arc cap at target end → right ribbon reversed (Bezier) → arc cap at source end (reversed).

class fstg_toolkit.app.core.geometry.Shape

Bases: ABC

Abstract base class for geometric shapes that can be converted to a Path.

abstractmethod to_path() → Path

Convert the shape to a Path suitable for SVG rendering.

Returns:

The closed polygonal approximation of the shape.

Return type:

Path

class fstg_toolkit.app.core.color.ColorsInterpolator

Bases: ABC

Abstract base class for color interpolators that generate a palette of n colors.

abstractmethod sample(n: int) → list[tuple[float, float, float]]

Generate n evenly-distributed colors.

Parameters:

n (int) – Number of colors to generate.

Returns:

A list of n RGB tuples with components in [0, 1].

Return type:

list[tuple[float, float, float]]

class fstg_toolkit.app.core.color.HueInterpolator(saturation: float = 0.7, value: float = 0.9)

Bases: ColorsInterpolator

Color interpolator that steps evenly through the HSV hue wheel.

Parameters:

• saturation (float, optional) – HSV saturation for all generated colors (default 0.7).

• value (float, optional) – HSV value (brightness) for all generated colors (default 0.9).

sample(n: int) → list[tuple[float, float, float]]

Generate n colors by stepping uniformly through the hue wheel.

Parameters:

n (int) – Number of colors to generate.

Returns:

A list of n RGB tuples with components in [0, 1].

Return type:

list[tuple[float, float, float]]

saturation: float = 0.7

value: float = 0.9

exception fstg_toolkit.app.core.datafilesdb.AbstractClassNotMeantToBeUsedDirectly

Bases: NotImplementedError

class fstg_toolkit.app.core.datafilesdb.DataFilesDB(token_nb_bytes: int = 3, debug: bool = False)

Bases: ABC

Abstract base class for managing a database of data files, each associated with a unique token.

Parameters:

token_nb_bytes (int, optional) – Number of bytes to use when generating unique tokens for data files (default is 3).

add(file_path: Path) → str

Adds a data file to the database and returns its unique token.

get(token: str) → pathlib.Path | None

Retrieves the file path associated with the given token.

list() → Generator[tuple[str, pathlib.Path], None, None]

Lists all token-file path pairs in the database.

add(file_path: Path) → str

Adds a data file to the database and returns its unique token.

Parameters:

file_path (Path) – The path to the data file to be added.

Returns:

A unique token associated with the added data file.

Return type:

str

abstractmethod get(token: str) → Path | None

Retrieves the file path associated with the given token.

Parameters:

token (str) – The unique token associated with the data file.

Returns:

The path to the data file if found, otherwise None.

Return type:

pathlib.Path or None

abstractmethod list() → Generator[tuple[str, Path], None, None]

Lists all token-file path pairs in the database.

Returns:

A generator yielding tuples of tokens and their associated file paths.

Return type:

Generator of tuple[str, pathlib.Path]

class fstg_toolkit.app.core.datafilesdb.MemoryDataFilesDB(*args, **kwargs)

Bases: DataFilesDB

In-memory implementation of the DataFilesDB abstract base class.

This class manages a database of data files using a Python dictionary, mapping unique tokens to file paths. It is suitable for use cases where persistence is not required and the number of files is relatively small.

Parameters:

token_nb_bytes (int, optional) – Number of bytes to use when generating unique tokens for data files (default is 3).

add(file_path: Path) → str

Adds a data file to the database and returns its unique token.

get(token: str) → Path | None

Retrieves the file path associated with the given token.

list() → Generator[tuple[str, Path], None, None]

Lists all token-file path pairs in the database.

get(token: str) → Path | None

Retrieves the file path associated with the given token.

Parameters:

token (str) – The unique token associated with the data file.

Returns:

The path to the data file if found, otherwise None.

Return type:

pathlib.Path or None

list() → Generator[tuple[str, Path], None, None]

Lists all token-file path pairs in the database.

Returns:

A generator yielding tuples of tokens and their associated file paths.

Return type:

Generator of tuple[str, pathlib.Path]

class fstg_toolkit.app.core.datafilesdb.SQLiteDataFilesDB(db_path: Path, *args, **kwargs)

Bases: DataFilesDB, SQLiteConnected

SQLite-backed implementation of the DataFilesDB.

Stores token -> file path mappings in a SQLite database. Pass a Path or string via the db_path parameter to persist to disk.

Parameters:

• db_path (pathlib.Path | str) – Path to the sqlite database file.

• token_nb_bytes (Other keyword args are forwarded to the parent DataFilesDB (e.g.)

:param : :param debug).:

get(token: str) → Path | None

Retrieves the file path associated with the given token.

Parameters:

token (str) – The unique token associated with the data file.

Returns:

The path to the data file if found, otherwise None.

Return type:

pathlib.Path or None

list() → Generator[tuple[str, Path], None, None]

Lists all token-file path pairs in the database.

Returns:

A generator yielding tuples of tokens and their associated file paths.

Return type:

Generator of tuple[str, pathlib.Path]

fstg_toolkit.app.core.datafilesdb.get_data_file_db(requested_type: Type[DataFilesDB] | None = None, **kwargs) → DataFilesDB

Returns the singleton instance of the data files database.

If the singleton instance does not exist, it is created using the specified type or defaults to SQLiteDataFilesDB.

Parameters:

requested_type (type[DataFilesDB] or None, optional) – The class type of the data files database to instantiate. If None, defaults to SQLiteDataFilesDB for database creation.

Returns:

The singleton instance of the data files database.

Return type:

DataFilesDB

Raises:

RuntimeError – If a database instance already exists with a different type than requested_type.

class fstg_toolkit.app.core.processing.DatasetProcessingManager(db_path: Path)

Bases: SQLiteConnected

list(limit: int = 30) → list[DatasetResult]

submit(dataset: SubmittedDataset)

class fstg_toolkit.app.core.processing.DatasetResult(dataset: fstg_toolkit.app.core.processing.SubmittedDataset, job_status: fstg_toolkit.app.core.processing.ProcessingJobStatus, submitted_at: datetime.datetime, result: str, error: str | None)

Bases: object

dataset: SubmittedDataset

error: str | None

job_status: ProcessingJobStatus

result: str

submitted_at: datetime

exception fstg_toolkit.app.core.processing.InvalidSubmittedDataset(message: str)

Bases: Exception

class fstg_toolkit.app.core.processing.JobStatusMonitor(db_path: Path)

Bases: SQLiteConnected, ProcessingQueueListener

on_job_completed(job_id: str, result: T | None)

Called when a job finishes successfully.

Parameters:

• job_id (str) – The unique identifier of the completed job.

• result (any or None) – The return value of the job function.

on_job_failed(job_id: str, error: Exception)

Called when a job raises an unhandled exception.

Parameters:

• job_id (str) – The unique identifier of the failed job.

• error (Exception) – The exception that caused the failure.

on_job_started(job_id: str)

Called when a worker thread picks up the job and starts executing it.

Parameters:

job_id (str) – The unique identifier of the started job.

on_job_submitted(job_id: str)

Called immediately after a job is submitted to the queue.

Parameters:

job_id (str) – The unique identifier of the submitted job.

class fstg_toolkit.app.core.processing.ProcessingJobStatus(*values)

Bases: Enum

COMPLETED = 'Completed'

FAILED = 'Failed'

PENDING = 'Pending'

RUNNING = 'Running'

static from_value(name: str) → ProcessingJobStatus | None

class fstg_toolkit.app.core.processing.ProcessingQueue(max_workers: int = 1, listener: ProcessingQueueListener | None = None)

Bases: object

A thread-pool-backed job queue with optional lifecycle event notifications.

Each submitted callable receives a unique job ID as its first argument. An optional ProcessingQueueListener is notified at each stage of a job’s lifecycle (submitted, started, completed, failed).

__init__(max_workers: int = 1, listener: ProcessingQueueListener | None = None)

Initialise the processing queue.

Parameters:

• max_workers (int, optional) – Maximum number of concurrent worker threads (default 1).

• listener (ProcessingQueueListener or None, optional) – Observer to notify on job lifecycle events.

get_result(job_id: str) → T | None

submit(func: Callable[[str, ...], T], *args, **kwargs) → str

Submit a callable for asynchronous execution.

Parameters:

• func (Callable) – The function to execute. It will receive a unique job ID string as its first positional argument, followed by *args and **kwargs.

• *args – Additional arguments forwarded to func.

• **kwargs – Additional arguments forwarded to func.

Returns:

The UUID string identifying the submitted job.

Return type:

str

class fstg_toolkit.app.core.processing.ProcessingQueueListener

Bases: ABC

Abstract observer interface for ProcessingQueue lifecycle events.

Implement this interface to receive notifications when jobs are submitted, started, completed, or failed.

abstractmethod on_job_completed(job_id: str, result: T | None)

Called when a job finishes successfully.

Parameters:

• job_id (str) – The unique identifier of the completed job.

• result (any or None) – The return value of the job function.

abstractmethod on_job_failed(job_id: str, error: Exception)

Called when a job raises an unhandled exception.

Parameters:

• job_id (str) – The unique identifier of the failed job.

• error (Exception) – The exception that caused the failure.

abstractmethod on_job_started(job_id: str)

Called when a worker thread picks up the job and starts executing it.

Parameters:

job_id (str) – The unique identifier of the started job.

abstractmethod on_job_submitted(job_id: str)

Called immediately after a job is submitted to the queue.

Parameters:

job_id (str) – The unique identifier of the submitted job.

class fstg_toolkit.app.core.processing.SubmittedDataset(name: str, include_raw: bool, compute_metrics: bool, compute_frequent: bool, areas_file: pathlib.Path, matrices_files: list[pathlib.Path])

Bases: object

areas_file: Path

check()

compute_frequent: bool

compute_metrics: bool

static from_record(record: dict[str, Any]) → SubmittedDataset

include_raw: bool

matrices_files: list[Path]

name: str

fstg_toolkit.app.core.processing.get_dataset_processing_manager() → DatasetProcessingManager

fstg_toolkit.app.core.processing.get_processing_queue() → ProcessingQueue

class fstg_toolkit.app.core.utils.SQLiteConnected(db_path: Path | None, timeout: float = 30.0)

Bases: object

A class that brings a connection to an SQLite database as a feature.

db_path

The file path to the SQLite database.

Type:

Path

db_path: Path | None

timeout: float = 30.0

fstg_toolkit.app.core.utils.join(l: list[Any], sep: Any) → list[Any]

Interleave a separator between elements of a list.

Parameters:

• l (list) – Sequence of elements to be joined. If empty, an empty list is returned.

• sep (Any) – Separator value to insert between consecutive elements of l.

Returns:

A new list with sep inserted between each pair of elements from l. If l is empty, returns []. If l contains a single element, a shallow copy of l is returned.

Return type:

list

Examples

>>> join([1, 2, 3], 0)
[1, 0, 2, 0, 3]
>>> join(['a'], '-')
['a']
>>> join([], None)
[]