API Reference¶

Props¶

class port.api.props.PropsUIFooter¶

Page footer.

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

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

Page header.

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

A multi-purpose page that gets shown to the user.

platform¶

The platform name the user is currently in the process of donating data from.

Type:: str

header¶

Page header.

Type:: PropsUIHeader

body (PropsUIPromptRadioInput | PropsUIPromptConsentForm |: PropsUIPromptFileInput | PropsUIPromptFileInputMultiple | PropsUIPromptConfirm | PropsUIPromptQuestionnaire): Main body of the page.

footer¶

Optional page footer.

Type:: Optional[PropsUIFooter]

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

class port.api.props.PropsUIPageEnd¶

An ending page to show the user they are done.

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

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:: Translatable

ok¶

Message to display if the user wants to try again.

Type:: Translatable

cancel¶

Message to display if the user wants to continue regardless.

Type:: Translatable

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

class port.api.props.PropsUIPromptConsentForm(id: str, tables: list[PropsUIPromptConsentFormTable], meta_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 donation.

id¶

will be used as part of the filename when the data is stored

Type:: str

tables¶

A list of tables.

Type:: list[PropsUIPromptConsentFormTable]

meta_tables¶

A list of optional tables, for example for logging data.

Type:: list[PropsUIPromptConsentFormTable]

description¶

Optional description of the consent form.

Type:: Optional[Translatable]

donate_question¶

Optional donation question.

Type:: Optional[Translatable]

donate_button¶

Optional text for the donate button.

Type:: Optional[Translatable]

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

translate_meta_tables()¶

Translate the meta tables to a list of dictionaries.

Returns:: A list of dictionaries representing the meta tables.
Return type:: list

translate_tables()¶

Translate the tables to a list of dictionaries.

Returns:: A list of dictionaries representing the tables.
Return type:: list

class port.api.props.PropsUIPromptConsentFormTable(id: str, title: Translatable, data_frame: pandas.DataFrame | Dict[str, Dict[str, Any]], description: Translatable | None = None, visualizations: list | None = None, folded: bool | None = False, delete_option: bool | None = True)¶

Table to be shown to the participant prior to donation.

id¶

A unique string to identify the table after donation.

Type:: str

title¶

Title of the table.

Type:: Translatable

data_frame¶

Table to be shown can be a pandas DataFrame or a dictionary.

Type:: pd.DataFrame | Dict[str, Dict[str, Any]]

description¶

Optional description of the table.

Type:: Optional[Translatable]

visualizations¶

Optional visualizations to be shown.

Type:: Optional[list]

folded¶

Whether the table should be initially folded.

Type:: Optional[bool]

delete_option¶

Whether to show a delete option for the table.

Type:: Optional[bool]

Examples:

data_frame_df = pd.DataFrame([
    {"column1": 1, "column2": 4},
    {"column1": 2, "column2": 5},
    {"column1": 3, "column2": 6},
])

example1 = PropsUIPromptConsentFormTable(
    id="example1",
    title=Translatable("Table as DataFrame"),
    data_frame=data_frame_df,
)

data_frame_dict = {
    "column1": {"0": 1, "1": 4},
    "column2": {"0": 2, "1": 5},
    "column3": {"0": 3, "1": 6},
}

example2 = PropsUIPromptConsentFormTable(
    id="example2",
    title=Translatable("Table as Dictionary"),
    data_frame=data_frame_dict,
)

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

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

Prompt the user to submit a file.

description¶

Text with an explanation.

Type:: Translatable

extensions¶

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

Type:: str

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

class port.api.props.PropsUIPromptFileInputMultiple(description: Translatable, extensions: str)¶

Prompt the user to submit multiple files.

description¶

Text with an explanation.

Type:: Translatable

extensions¶

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

Type:: str

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

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:: Translatable

message¶

Can be used to show extraction progress.

Type:: str

percentage¶

Optional percentage of progress.

Type:: Optional[int]

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

class port.api.props.PropsUIPromptQuestionnaire(description: Translatable, questions: list[PropsUIQuestionMultipleChoice | PropsUIQuestionMultipleChoiceCheckbox | PropsUIQuestionOpen])¶

Questionnaire containing multiple questions.

description¶

Description of the questionnaire.

Type:: Translatable

questions¶

List of questions in the questionnaire.

Type:: list[PropsUIQuestionMultipleChoice | PropsUIQuestionMultipleChoiceCheckbox | PropsUIQuestionOpen]

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

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

Radio group.

This radio group can be used to get a multiple choice answer from a user.

title¶

Title of the radio group.

Type:: Translatable

description¶

Short description of the radio group.

Type:: Translatable

items¶

A list of radio buttons.

Type:: list[RadioItem]

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

class port.api.props.PropsUIQuestionMultipleChoice(id: int, question: Translatable, choices: list[Translatable])¶

Multiple choice question with radio buttons.

id¶

Question ID.

Type:: int

question¶

The question text.

Type:: Translatable

choices¶

List of choices.

Type:: list[Translatable]

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

class port.api.props.PropsUIQuestionMultipleChoiceCheckbox(id: int, question: Translatable, choices: list[Translatable])¶

Multiple choice question with checkboxes.

id¶

Question ID.

Type:: int

question¶

The question text.

Type:: Translatable

choices¶

List of choices.

Type:: list[Translatable]

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

class port.api.props.PropsUIQuestionOpen(id: int, question: Translatable)¶

Open-ended question.

id¶

Question ID.

Type:: int

question¶

The question text.

Type:: Translatable

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

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.

toDict()¶

Convert the object to a dictionary.

Returns:: A dictionary representation of the object.
Return type:: dict

class port.api.props.Translations¶: Typed dict containing text that is displayed in a specific language.

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_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(id: str, description: Translatable, table_list: list[PropsUIPromptConsentFormTable]) → PropsUIPromptConsentForm¶

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:

id (str) – will be used as part of the filename when the data is stored
table_list (list[props.PropsUIPromptConsentFormTable]) – 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_end_page()¶: Renders a thank you page, must be yielded.

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.PropsUIPromptFileInputMultiple | props.PropsUIPromptConfirm | props.PropsUIPromptQuestionnaire
) – 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.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(value)¶: Enumeration of supported DDP file types.

class port.helpers.validate.Language(value)¶: 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