API Reference

API Reference#

Props#

class port.api.props.PropsUIDataSubmissionButtons(donate_question: Translatable | None = None, donate_button: Translatable | None = None, waiting: bool = False)#

Buttons for data submission actions

donate_question#

Optional question text above buttons

Type:: port.api.props.Translatable | None

donate_button#

Optional text for donate button

Type:: port.api.props.Translatable | None

waiting#

Whether the data submission is in progress

Type:: bool

class port.api.props.PropsUIFooter(progressPercentage: float)#

Page footer

progressPercentage#

float indicating the progress in the flow

Type:: float

class port.api.props.PropsUIHeader(title: Translatable)#

Page header

title#

title of the page

Type:: port.api.props.Translatable

A multi-purpose page that gets shown to the user

platform#

the platform name the user is curently in the process of donating data from

Type:: str

header#

page header

Type:: port.api.props.PropsUIHeader

body#

main body of the page, see the individual classes for an explanation

Type:: port.api.props.PropsUIPromptRadioInput | port.api.props.PropsUIPromptConsentForm | port.api.props.PropsUIPromptFileInput | port.api.props.PropsUIPromptConfirm | port.api.props.PropsUIPromptProgress | port.api.props.PropsUIPromptHelloWorld | port.api.props.PropsUIPromptConsentFormTable | port.api.props.PropsUIDataSubmissionButtons | list | Any

class port.api.props.PropsUIPageEnd#: An ending page to show the user they are done

class port.api.props.PropsUIPromptConfirm(text: Translatable, ok: Translatable, cancel: Translatable)#

Retry submitting a file page

Prompt the user if they want to submit a new file. This can be used in case a file could not be processed.

text#

message to display

Type:: port.api.props.Translatable

ok#

message to display if the user wants to try again

Type:: port.api.props.Translatable

cancel#

message to display if the user wants to continue regardless

Type:: port.api.props.Translatable

class port.api.props.PropsUIPromptConsentForm(tables: list[PropsUIPromptConsentFormTable], description: Translatable | None = None, donate_question: Translatable | None = None, donate_button: Translatable | None = None)#

Tables to be shown to the participant prior to data submission

tables#

a list of tables, including both editable and read-only tables

Type:: list[port.api.props.PropsUIPromptConsentFormTable]

description#

Optional description text

Type:: port.api.props.Translatable | None

donate_question#

Optional question text for data submission button

Type:: port.api.props.Translatable | None

donate_button#

Optional text for data submission button

Type:: port.api.props.Translatable | None

class port.api.props.PropsUIPromptConsentFormTable(id: str, title: Translatable, description: Translatable, data_frame: pandas.DataFrame, headers: dict[str, Translatable] | None = None)#

Table to be shown to the participant prior to data_submission

id#

a unique string to itentify the table after data_submission

Type:: str

title#

title of the table

Type:: port.api.props.Translatable

data_frame#

table to be shown

Type:: pandas.DataFrame

class port.api.props.PropsUIPromptFileInput(description: Translatable, extensions: str)#

Prompt the user to submit a file

description#

text with an explanation

Type:: port.api.props.Translatable

extensions#

accepted mime types, example: “application/zip, text/plain”

Type:: str

class port.api.props.PropsUIPromptHelloWorld(text: Translatable)#

Hello world component to welcome users

text#

welcome message to display

Type:: port.api.props.Translatable

class port.api.props.PropsUIPromptProgress(description: Translatable, message: str, percentage: int | None = None)#

Prompt the user information during the extraction

description#

text with an explanation

Type:: port.api.props.Translatable

message#

can be used to show extraction progress

Type:: str

class port.api.props.PropsUIPromptRadioInput(title: Translatable, description: Translatable, items: list[RadioItem])#

Radio group

This radio group can be used get a mutiple choice answer from a user

title#

title of the radio group

Type:: port.api.props.Translatable

description#

short description of the radio group

Type:: port.api.props.Translatable

items#

a list of radio buttons

Type:: list[port.api.props.RadioItem]

class port.api.props.PropsUIPromptText(text: Translatable, title: Translatable | None = None)#

Text block to display information to the user

title#

optional title for the text block

Type:: port.api.props.Translatable | None

text#

main text content to display

Type:: port.api.props.Translatable

class port.api.props.RadioItem#

Radio button

id#

id of radio button

Type:: int

value#

text to be displayed

Type:: str

class port.api.props.Translatable(translations: Translations)#: Wrapper class for Translations

class port.api.props.Translations#

Typed dict containing text that is display in a speficic language

en#

English string to display

Type:: str

nl#

Dutch string to display

Type:: str

Extraction helpers#

This module contains helper functions that can be used during the data extraction process

exception port.helpers.extraction_helpers.FileNotFoundInZipError#: The File you are looking for is not present in a zipfile

port.helpers.extraction_helpers.dict_denester(inp: dict[Any, Any] | list[Any], new: dict[Any, Any] | None = None, name: str = '', run_first: bool = True) → dict[Any, Any]#

Denests a dictionary or list, returning a new flattened dictionary.

Parameters:

inp (dict[Any, Any] | list[Any]) – The input dictionary or list to be denested.
new (dict[Any, Any] | None, optional) – The dictionary to store denested key-value pairs. Defaults to None.
name (str, optional) – The current key name in the denesting process. Defaults to “”.
run_first (bool, optional) – Flag to indicate if this is the first run of the function. Defaults to True.

Returns:

A new denested dictionary.

Return type:

dict[Any, Any]

Examples:

>>> nested_dict = {"a": {"b": {"c": 1}}, "d": [2, 3]}
>>> dict_denester(nested_dict)
{"a-b-c": 1, "d-0": 2, "d-1": 3}

port.helpers.extraction_helpers.epoch_to_iso(epoch_timestamp: str | int | float) → str#

Convert epoch timestamp to an ISO 8601 string, assuming UTC.

Parameters:: epoch_timestamp (str | int) – The epoch timestamp to convert.
Returns:: The ISO 8601 formatted string, or the original input if conversion fails.
Return type:: str
Raises:: Exception – Logs an error message if conversion fails.

Examples:

>>> epoch_to_iso(1632139200)
"2021-09-20T12:00:00+00:00"

port.helpers.extraction_helpers.extract_file_from_zip(zfile: str, file_to_extract: str) → BytesIO#

Extracts a specific file from a zipfile and returns it as a BytesIO buffer.

Parameters:

zfile (str) – Path to the zip file.
file_to_extract (str) – Name or path of the file to extract from the zip.

Returns:

A BytesIO buffer containing the extracted file’s content of the first file found.: Returns an empty BytesIO if the file is not found or an error occurs.

Return type:

io.BytesIO

Raises:

FileNotFoundInZipError – Logs an error if the specified file is not found in the zip.
zipfile.BadZipFile – Logs an error if the zip file is invalid.
Exception – Logs any other unexpected errors.

Examples:

>>> extracted_file = extract_file_from_zip("archive.zip", "data.txt")
>>> content = extracted_file.getvalue().decode('utf-8')

port.helpers.extraction_helpers.find_item(d: dict[Any, Any], key_to_match: str) → str#

Finds the least nested value in a denested dictionary whose key contains the given key_to_match.

Parameters:

d (dict[Any, Any]) – A denested dictionary to search in.
key_to_match (str) – The substring to match in the keys.

Returns:

The value of the least nested key containing key_to_match.: Returns an empty string if no match is found.

Return type:

str

Raises:

Exception – Logs an error message if an exception occurs during the search.

Examples:

>>> d = {"asd-asd-asd": 1, "asd-asd": 2, "qwe": 3}
>>> find_item(d, "asd")
"2"

port.helpers.extraction_helpers.find_items(d: dict[Any, Any], key_to_match: str) → list#

Finds all values in a denested dictionary whose keys contain the given key_to_match.

Parameters:

d (dict[Any, Any]) – A denested dictionary to search in.
key_to_match (str) – The substring to match in the keys.

Returns:

A list of all values whose keys contain key_to_match.

Return type:

list

Raises:

Exception – Logs an error message if an exception occurs during the search.

Examples:

>>> d = {"asd-1": "a", "asd-2": "b", "qwe": "c"}
>>> find_items(d, "asd")
["a", "b"]

port.helpers.extraction_helpers.fix_ascii_string(input: str) → str#

Fixes the string encoding by removing non-ASCII characters.

Parameters:: input (str) – The input string that needs to be fixed.
Returns:: The fixed string with only ASCII characters, or the original string if an exception occurs.
Return type:: str

Examples:

>>> fix_ascii_string("Hello, 世界!")
"Hello, !"

port.helpers.extraction_helpers.fix_latin1_string(input: str) → str#

Fixes the string encoding by attempting to encode it using the ‘latin1’ encoding and then decoding it.

Parameters:: input (str) – The input string that needs to be fixed.
Returns:: The fixed string after encoding and decoding, or the original string if an exception occurs.
Return type:: str

Examples:

>>> fix_latin1_string("café")
"café"

port.helpers.extraction_helpers.json_dumper(zfile: str) → pandas.DataFrame#

Reads all JSON files in a zip file, flattens them, and combines them into a single DataFrame.

Parameters:: zfile (str) – Path to the zip file containing JSON files.
Returns:: A DataFrame containing flattened data from all JSON files in the zip.
Return type:: pd.DataFrame
Raises:: Exception – Logs an error message if an exception occurs during the process.

Examples:

>>> df = json_dumper("data.zip")
>>> print(df.head())

port.helpers.extraction_helpers.read_csv_from_bytes(json_bytes: BytesIO) → list[dict[Any, Any]]#

Reads CSV data from a BytesIO buffer and returns it as a list of dictionaries.

Parameters:

json_bytes (io.BytesIO) – A BytesIO buffer containing CSV data.

Returns:

A list of dictionaries, where each dictionary represents a row in the CSV.: Returns an empty list if parsing fails.

Return type:

list[dict[Any, Any]]

Examples

>>> buffer = io.BytesIO(b'name,age\nAlice,30\nBob,25')
>>> data = read_csv_from_bytes(buffer)
>>> print(data)
[{'name': 'Alice', 'age': '30'}, {'name': 'Bob', 'age': '25'}]

port.helpers.extraction_helpers.read_csv_from_bytes_to_df(json_bytes: BytesIO) → pandas.DataFrame#

Reads CSV data from a BytesIO buffer and returns it as a pandas DataFrame.

Parameters:: json_bytes (io.BytesIO) – A BytesIO buffer containing CSV data.
Returns:: A pandas DataFrame containing the CSV data.
Return type:: pd.DataFrame

Examples

>>> buffer = io.BytesIO(b'name,age\nAlice,30\nBob,25')
>>> df = read_csv_from_bytes_to_df(buffer)
>>> print(df)
   name  age
0  Alice   30
1    Bob   25

port.helpers.extraction_helpers.read_json_from_bytes(json_bytes: BytesIO) → dict[Any, Any] | list[Any]#

Reads JSON data from a BytesIO buffer.

Parameters:

json_bytes (io.BytesIO) – A BytesIO buffer containing JSON data.

Returns:

The parsed JSON data as a dictionary or list.: Returns an empty dictionary if parsing fails.

Return type:

dict[Any, Any] | list[Any]

Examples:

>>> buffer = io.BytesIO(b'{"key": "value"}')
>>> data = read_json_from_bytes(buffer)
>>> print(data)
{'key': 'value'}

port.helpers.extraction_helpers.read_json_from_file(json_file: str) → dict[Any, Any] | list[Any]#

Reads JSON data from a file.

Parameters:

json_file (str) – Path to the JSON file.

Returns:

The parsed JSON data as a dictionary or list.: Returns an empty dictionary if parsing fails.

Return type:

dict[Any, Any] | list[Any]

Examples:

>>> data = read_json_from_file("data.json")
>>> print(data)
{'key': 'value'}

port.helpers.extraction_helpers.replace_months(input_string: str) → str#

Replaces Dutch month abbreviations with English equivalents in the input string.

Parameters:: input_string (str) – The input string containing potential Dutch month abbreviations.
Returns:: The input string with Dutch month abbreviations replaced by English equivalents.
Return type:: str

Examples:

>>> replace_months("15 mei 2023")
"15 may 2023"

port.helpers.extraction_helpers.sort_isotimestamp_empty_timestamp_last(timestamp_series: pandas.Series) → pandas.Series#

Creates a key for sorting a pandas Series of ISO timestamps, placing empty timestamps last.

Parameters:: timestamp_series (pd.Series) – A pandas Series containing ISO formatted timestamps.
Returns:: A Series of sorting keys, with -timestamp for valid dates and infinity for invalid/empty dates.
Return type:: pd.Series

Examples:

>>> df = df.sort_values(by="Date", key=sort_isotimestamp_empty_timestamp_last)

Port helpers#

port.helpers.port_helpers.donate(key: str, json_string: str) → CommandSystemDonate#

Initiates a donation process using the provided key and data.

This function triggers the donation process by passing a key and a JSON-formatted string that contains donation information.

Parameters:

key (str) – The key associated with the donation process. The key will be used in the file name.
json_string (str) – A JSON-formatted string containing the donated data.

Returns:

A system command that initiates the donation process. Must be yielded.

Return type:

CommandSystemDonate

port.helpers.port_helpers.exit(code: int, info: str) → CommandSystemExit#

Exits Next with the provided exit code and additional information. This if the code reaches this function, it will return to the task list in Next.

Parameters:

code (int) – The exit code representing the type or status of the exit.
info (str) – A string containing additional information about the exit.

Returns:

A system command that initiates the exit process in Next.

Return type:

CommandSystemExit

Examples:

yield exit(0, "Success")

port.helpers.port_helpers.generate_file_prompt(extensions: str) → PropsUIPromptFileInput#

Generates a file input prompt for selecting a file for a platform.

This function creates a bilingual (English and Dutch) file input prompt that instructs the user to select a file they’ve received from a platform and stored on their device.

Parameters:

extensions (str) – A collection of allowed MIME types.
example (For) – “application/zip, text/plain, application/json”

Returns:

A file input prompt object containing the description text and allowed file extensions.

Return type:

props.PropsUIPromptFileInput

port.helpers.port_helpers.generate_radio_prompt(title: Translatable, description: Translatable, items: list[str]) → PropsUIPromptRadioInput#: General purpose prompt selection menu

port.helpers.port_helpers.generate_retry_prompt(platform_name: str) → PropsUIPromptConfirm#

Generates a confirmation prompt for retrying file processing.

This function creates a bilingual (English and Dutch) confirmation prompt when a file from a specific platform cannot be processed. It allows the user to either try again with a different file or continue with the current file.

Parameters:: platform_name (str) – The name of the platform associated with the file that couldn’t be processed. This is inserted into the prompt text.
Returns:: A confirmation prompt object containing the message, and labels for the “OK” (try again) and “Cancel” (continue) buttons.
Return type:: props.PropsUIPromptConfirm

port.helpers.port_helpers.generate_review_data_prompt(description: Translatable, table_list: list[PropsUIPromptConsentFormTableViz]) → PropsUIPromptConsentFormViz#

Generates a data review form with a list of tables and a description, including default donate question and button. The participant can review these tables before they will be send to the researcher. If the participant consents to sharing the data the data will be stored at the configured storage location.

Parameters:

table_list (list[props.PropsUIPromptConsentFormTableViz]) – A list of consent form tables to be included in the prompt.
description (props.Translatable) – A translatable description text for the consent prompt.

Returns:

A structured consent form object containing the provided table list, description, and default values for donate question and button.

Return type:

props.PropsUIPromptConsentForm

port.helpers.port_helpers.render_page(header_text: Translatable, body: PropsUIPromptRadioInput | PropsUIPromptConsentForm | PropsUIPromptConsentFormViz | PropsUIPromptFileInput | PropsUIPromptConfirm) → CommandUIRender#

Renders the UI components for a donation page.

This function assembles various UI components including a header, body, and footer to create a complete donation page. It uses the provided header text and body content to customize the page.

Parameters:

header_text (props.Translatable) – The text to be displayed in the header. This should be a translatable object to support multiple languages.
( (body) – props.PropsUIPromptRadioInput | props.PropsUIPromptConsentForm | props.PropsUIPromptFileInput | props.PropsUIPromptConfirm |
) – The main content of the page. It must be compatible with props.PropsUIPageDonation.

Returns:

A render command object containing the fully assembled page. Must be yielded.

Return type:

CommandUIRender

Validation#

Contains classes to deal with input validation of DDPs

The idea of this module is to provide a uniform way to assign a validation status to a DDP validation Which can be used and acted upon

class port.helpers.validate.BaseValidation(status_code: int)#: Base validation class that can be used for validation purposes

class port.helpers.validate.DDPCategory(id: str, ddp_filetype: DDPFiletype, language: Language, known_files: list[str])#

Represents characteristics that define a DDP (Data Delivery Package) category.

Parameters:

id (str) – Unique identifier for the DDP category.
ddp_filetype (DDPFiletype) – The file type of the DDP.
language (Language) – The language of the DDP.
known_files (List[str]) – A list of known files associated with this DDP category.

Examples

>>> category = DDPCategory("cat1", DDPFiletype.JSON, Language.EN, ["file1.json", "file2.json"])
>>> print(category.id)
cat1
>>> print(category.language)
<Language.EN: 1>

class port.helpers.validate.DDPFiletype(*values)#: Enumeration of supported DDP file types.

class port.helpers.validate.Language(*values)#: Enumeration of supported languages.

class port.helpers.validate.StatusCode(id: int, description: str)#

Represents a status code that can be used to set a DDP status.

Parameters:

id (int) – The numeric identifier of the status code.
description (str) – A brief description of what the status code represents.

Examples

>>> status = StatusCode(0, "Success")
>>> print(status.id)
0
>>> print(status.description)
Success

class port.helpers.validate.ValidateInput(all_status_codes: list[StatusCode], all_ddp_categories: list[DDPCategory], current_status_code: StatusCode | None = None, current_ddp_category: DDPCategory | None = None)#

A class for validating input data against predefined categories and status codes.

Parameters:

all_status_codes (List[StatusCode]) – A list of valid status codes.
all_ddp_categories (List[DDPCategory]) – A list of valid DDP categories.
current_status_code (Optional[StatusCode]) – The current status code. Defaults to None.
current_ddp_category (Optional[DDPCategory]) – The current DDP category. Defaults to None.

ddp_categories_lookup#

A lookup dictionary for DDP categories.

Type:: Dict[str, DDPCategory]

status_codes_lookup#

A lookup dictionary for status codes.

Type:: Dict[int, StatusCode]

Examples

>>> status_codes = [StatusCode(id=0, description="Success"), StatusCode(id=1, description="Error")]
>>> ddp_categories = [DDPCategory(id="cat1", ddp_filetype=DDPFiletype.JSON, language=Language.EN, known_files=["file1.txt", "file2.txt"])]
>>> validator = ValidateInput(all_status_codes=status_codes, all_ddp_categories=ddp_categories)

get_status_code_id() → int#

Return the current assigned status code ID. Note: zero is always used for OK. Non-zero otherwise.

Returns:: The ID of the current status code, or 1 if no status code is set.
Return type:: int

Examples

>>> validator.get_status_code_id()

infer_ddp_category(file_list_input: list[str]) → bool#

Compares a list of files to a list of known files and infers the DDPCategory.

Parameters:: file_list_input (List[str]) – A list of input files to compare against known files.
Returns:: True if a valid DDP category is inferred, False otherwise. It sets the current_status_code and current_ddp_category to either the DDP catogory match, or to an unknown category.
Return type:: bool

Examples

>>> validator.infer_ddp_category(["file1.txt", "file2.txt"])

set_current_status_code_by_id(id: int) → None#

Set the status code based on the provided ID.

Parameters:: id (int) – The ID of the status code to set.

Examples

>>> validator.set_current_status_code_by_id(0)

port.helpers.validate.validate_zip(ddp_categories: list[DDPCategory], path_to_zip: str) → ValidateInput#

Validates a DDP zip file against a list of DDP categories.

This function attempts to open and read the contents of a zip file, then uses the ValidateInput class to infer the DDP category based on the files in the zip. If the zip file is invalid or cannot be read, it sets an error status code (an integer greather than 0).

Parameters:

ddp_categories (List[DDPCategory]) – A list of valid DDP categories to compare against.
path_to_zip (str) – The file path to the zip file to be validated.

Returns:

An instance of ValidateInput containing the validation results.

Return type:

ValidateInput

Raises:

zipfile.BadZipFile – This exception is caught internally and results in an error status code.

Examples

>>> categories = [DDPCategory(id="cat1", ddp_filetype=DDPFiletype.JSON, language=Language.EN, known_files=["file1.txt", "file2.txt"])]
>>> result = validate_zip(categories, "path/to/valid.zip")
>>> result.get_status_code_id()
0

>>> result = validate_zip(categories, "path/to/invalid.zip")
>>> result.get_status_code_id()
1

API Reference

Contents

API Reference#

Props#

Extraction helpers#

Port helpers#

Validation#