timelink.kleio package

Subpackages

timelink.kleio.groups package

Submodules

timelink.kleio.importer module

class timelink.kleio.importer.KleioContext(value)[source]: Kleio context enumeration

class timelink.kleio.importer.SaxHandler(kleio_handler)[source]

SAX Handler for Kleio XML files

characters(content)[source]

Receive notification of character data.

The Parser will call this method to report each chunk of character data. SAX parsers may return all contiguous character data in a single chunk, or they may split it into several chunks; however, all of the characters in any single event must come from the same external entity so that the Locator provides useful information.

endDocument()[source]

Receive notification of the end of a document.

The SAX parser will invoke this method only once, and it will be the last method invoked during the parse. The parser shall not invoke this method until it has either abandoned parsing (because of an unrecoverable error) or reached the end of input.

endElement(name)[source]

Signals the end of an element in non-namespace mode.

The name parameter contains the name of the element type, just as with the startElement event.

ignorableWhitespace(whitespace)[source]

Receive notification of ignorable whitespace in element content.

Validating Parsers must use this method to report each chunk of ignorable whitespace (see the W3C XML 1.0 recommendation, section 2.10): non-validating parsers may also use this method if they are capable of parsing and using content models.

SAX parsers may return all contiguous whitespace in a single chunk, or they may split it into several chunks; however, all of the characters in any single event must come from the same external entity, so that the Locator provides useful information.

processingInstruction(target, data)[source]

Receive notification of a processing instruction.

The Parser will invoke this method once for each processing instruction found: note that processing instructions may occur before or after the main document element.

A SAX parser should never report an XML declaration (XML 1.0, section 2.8) or a text declaration (XML 1.0, section 4.3.1) using this method.

startDocument()[source]

Receive notification of the beginning of a document.

The SAX parser will invoke this method only once, before any other methods in this interface or in DTDHandler (except for setDocumentLocator).

startElement(name: str, attrs)[source]

Signals the start of an element in non-namespace mode.

The name parameter contains the raw XML 1.0 name of the element type as a string and the attrs parameter holds an instance of the Attributes class containing the attributes of the element.

timelink.kleio.importer.import_from_xml(filespec: str | Path, session: Session, options: dict | None = None) → dict[source]

Import data from file or url into a timelink-mhk database.

The data file must be a XML file in the format exported by the kleio translator.

Parameters:

filespec (str,Path) – a file path, URL of a data file or path in a Kleio Server.
conn_string (str) – SQLAlchemy connection string to database.
options (dict) –
a dictionnary with options
- ’return_stats’: if True import stats will be returned
- ’kleio_url’: the url of kleio server;
- ’kleio_token’: the authorization token for the kleio server.
- ’mode’: the mode of the import, either ‘TL’(Timelink) or ‘MHK’
from (If kleio_url and kleio_token are specified the data will be fetched) –
file (a KleioServer and the filespec should contain the "xml_path" of the) –
server (in the) –
autentication (the kleio_token will be inserted in the header for) –

Returns:

If stats is True in options a dict with statistical information will be returned.

’datetime’: the time stamp of the start of the import

’machine’: local machine name

’file’: path to imported file or url

’import_time_seconds’: elapsed time during import

’entities_processed’: number of entities processed

’entity_rate’: number of entities processed per second

’person_rate’: number of persons (entities of class ‘person’)

’nerrors’: number of errors during import

’errors’: list of error messages

Examples

Returned statistical information when stats=True

{
'datetime': '2022-01-02 18:11:12',
'machine': 'joaquims-mbpr.local',
'file': 'https://...b1685.xml',
'import_time_seconds': 7.022288084030151,
'entities_processed': 747,
'entity_rate': 106.37558457603042,
'person_rate': 27.483919441999827
'nerrors': 0
'errors': []
}

TODO: should use https when the kleio_url not local.

timelink.kleio.kleio_server module

Interface to Kleio Server

class timelink.kleio.kleio_server.KleioServer(container: Container | None = None, url: str | None = None, token: str | None = None, kleio_home: str | None = None)[source]

This class interfaces to a Kleio server through its JSON-RPC api. It also provides convenience methods to start a server in Docker locally.

This class is not intended to be used directly. Use KleioServer.start() and KleioServer.attach() to create instances of KleioServer.

Parameters:

container (docker.models.containers.Container) – runing kleio server container
url (str) – kleio server url if running in a different machine (container=None)
token (str) – kleio server token if running in a different machine (container=None)
kleio_home (str) – kleio server home directory. If None and container is not None then kleio_home is obtained from the container. If not none

Not to be used directly.

See KleioServer.start() and KleioServer.attach()

To start a kleio server locally in docker use KleioServer.start()

To attach to a running kleio server, local or remote, use KleioServer.attach()

Parameters:

container (docker.models.containers.Container) – runing kleio server container; if container is None, then url, token and kleio_home must be provided.
url (str) – kleio server url if running in a different machine (container=None)
token (str) – kleio server token if running in a different machine (container=None)
kleio_home (str) – kleio server home directory. If None and container is not None then kleio_home is obtained from the container. If not none

static attach(url: str, token: str, kleio_home: str | None = None)[source]

Attach to a already running Kleio Server.

Use this either to attach to a running server outside docker (for instance in a Prolog session) or to a server running in another machine (should use htpps url in that case)

Parameters:

url (str) – kleio server url
token (str) – kleio server token
kleio_home (str, optional) – kleio server home directory

Returns:

KleioServer object

Return type:

KleioServer

call(method: str, params: dict)[source]

Call kleio server API

Parameters:

method (str) – kleio server API method
params (dict) – kleio server API method parameters

Returns:

kleio server API response

Return type:

dict

container: Container: kleio server container

static find_local_kleio_home(path: str | None = None)[source]

Find kleio home directory.

Kleio home directory is the directory where Kleio Server finds sources and auxiliary files like structures, mappings and inferences.

It can be in the current directory, parent directory, or tests directory. It can be named “kleio-home”, “timelink-home”, or “mhk-home”.

A special case is when the current directory is “notebooks”. In this case, kleio-home is assumed to be the parent directory of “notebooks” and thus set up as the timelink-project template.

Parameters:: path (str, optional) – path to start searching from; defaults to None -> current directory.
Returns:: kleio home directory
Return type:: str

generate_token(user: str, info: TokenInfo)[source]

Generate a token for a user

Parameters:

user (str) – user to generate token for
info (TokenInfo) – token information

Returns:

kleio server API response

Return type:

dict

get_container()[source]

Get the kleio server container

Returns:: kleio server container
Return type:: docker.models.containers.Container

get_kleio_home()[source]

Get the kleio server home directory

Returns:: kleio server home directory
Return type:: str

get_report(rpt_url: str) → str[source]

Get report from kleio server

Parameters:: rpt_url (str) – url of the report in the Kleio Server
Returns:: kleio server API response
Return type:: dict

static get_server(kleio_home: str | None = None)[source]

Check if a kleio server is running in docker mapped to a given kleio home directory.

If yes return a KleioServer object, otherwise return None

Parameters:: kleio_home (str, optional) – kleio home directory; defaults to None -> any kleio home.
Returns:: KleioServer object or None
Return type:: KleioServer or None

get_sources(path: str, recurse: str)[source]

Get sources from kleio server

Parameters:

path (str) – path to the directory in sources
recurse (str) – if “yes” recurse in subdirectories

Returns:

kleio server API response

Return type:

dict

get_token()[source]

Get the kleio server token

Returns:: kleio server token
Return type:: str

get_url()[source]

Get the kleio server url

Returns:: kleio server url
Return type:: str

get_url_content(server_url: str) → str[source]

Get content from Kleio Server

Parameters:: server_url (str) – url of the content in the Kleio Server

host: str: kleio server host

invalidate_user(user: str)[source]

Invalidate a user

Parameters:: user (str) – user to invalidate
Returns:: kleio server API response
Return type:: dict

static is_server_running(kleio_home: str | None = None)[source]

Check if a kleio server is running in docker mapped to a given kleio home directory.

Return True of False

Parameters:: kleio_home (str, optional) – kleio home directory; defaults to None -> any kleio home.
Returns:: True if server is running, False otherwise
Return type:: bool

kleio_home: str: kleio server home directory

static make_token()[source]

Get the kleio server token from environment or generate a new one if not set.

Returns:: kleio server token
Return type:: str

static start(kleio_image: str = 'timelinkserver/kleio-server', kleio_version: str | None = 'latest', kleio_home: str | None = None, kleio_token: str | None = None, consistency: str = 'cached', port: int | None = None, update: bool = False, reuse: bool = True)[source]

Starts a kleio server in docker

Parameters:

image (str) – kleio server image, defaults to “timelinkserver/kleio-server”
version (str, optional) – kleio-server version, defaults to “latest”
kleio_home (str, optional) – kleio home directory, defaults to None -> current directory
token (str, optional) – kleio server token, defaults to None -> generate a random token
consistency (str, optional) – consistency of the volume mount, defaults to “cached”
port (int, optional) – port to map to 8088, defaults to None -> find a free port starting at 8088
update (bool, optional) – update kleio server image, defaults to False
reuse (bool, optional) – if True, reuse an existing kleio server container with same keio_home, defaults to True.

Returns:

KleioServer object

Return type:

KleioServer

stop()[source]: Stop the kleio server container

token: str: kleio server token

translate(path: str, recurse: str = 'yes', spawn: str = 'yes')[source]

Translate sources from kleio server

Parameters:

path (str) – path to the directory in sources
recurse (str, optional) – if “yes” recurse in subdirectories
spawn (str, optional) – if “yes” spawn a translation process for each file

Returns:

kleio server API response

Return type:

dict

translation_clean(path: str, recurse: str)[source]

clean translations from kleio server

Removes translation results from kleio server.

Parameters:

path (str) – path to the directory in sources
recurse (str) – if “yes” recurse in subdirectories

Returns:

kleio server API response

Return type:

dict

translation_status(path: str, recurse: str, status: str | None = None) → List[KleioFile][source]

Get translations from kleio server

Parameters:

path (str) – path to the directory in sources
recurse (str) – if “yes” recurse in subdirectories
status (str, optional) – filter by translation status V = valid translations T = need translation (source more recent than translation) E = translation with errors W = translation with warnings P = translation being processed Q = file queued for translation

Returns:

list of KleioFile objects

Return type:

list[KleioFile]

url: str: kleio server url

timelink.kleio.kleio_server.find_local_kleio_home(path: str | None = None)[source]

Find kleio home directory.

Kleio home directory is the directory where Kleio Server finds sources and auxiliary files like structures, mappings and inferences.

It can be in the current directory, parent directory, or tests directory. It can be named “kleio-home”, “timelink-home”, or “mhk-home”.

Special cases:

if the current directory is “notebooks”, kleio-home is assumed that Kleio home
to be the parent directory of “notebooks”
if there is a “tests” subdirectory, kleio-home is searched in childs of “tests”

timelink.kleio.kleio_server.get_kserver_container(kleio_home: str | None = None)[source]: Check if a kleio server is running in docker, possibly mapped to a given kleio home directory.

timelink.kleio.kleio_server.get_kserver_container_list() → None | List[Container][source]: Get the Kleio server container :returns: the Kleio server container :rtype: docker.models.containers.Container

timelink.kleio.kleio_server.get_kserver_home(container: Container | None = None, container_number: int = 0)[source]

Get the kleio server home directory

Parameters:

container (docker.models.containers.Container, optional) – kleio server container; defaults to None -> get by number.
container_number (int, optional) – container number. Defaults to 0.

Returns the volume mapped to /kleio-home in the kleio server container

timelink.kleio.kleio_server.get_kserver_token(container: Container | None = None, container_number: int = 0) → str[source]

Get the Kleio server container admin token

Parameters:

container (docker.models.containers.Container, optional) – kleio server container; defaults to None -> get the first container.
container_number (int, optional) – container number. Defaults to 0.

Returns:

the kleio server container token

Return type:

str

timelink.kleio.kleio_server.is_docker_running()[source]: Check if docker is running

timelink.kleio.kleio_server.random_token(length=32)[source]: Generate a random token

timelink.kleio.kleio_server.start_kleio_server(image: str = 'timelinkserver/kleio-server', version: str | None = 'latest', kleio_home: str | None = None, token: str | None = None, consistency: str = 'cached', port: int | None = None, update: bool = False, reuse: bool = True)[source]

Starts a kleio server in docker

Parameters:

image (str, optional) – kleio server image. Defaults to “time-link/kleio-server”.
version (str | None, optional) – kleio-server version; defaults to “latest”.
kleio_home (str | None, optional) – kleio home directory. Defaults to current directory.
token (str | None, optional) – kleio server token; if None -> generate a random token.
consistency (str, optional) – consistency of the volume mount. Defaults to “cached”.
port (int, optional) – port to map to 8088, if none find a free port starting at 8088.
update (bool, optional) – update kleio server image. Defaults to False.
reuse (bool, optional) – if True, reuse an existing kleio server container with same keio_home; defaults to True.

timelink.kleio.kleio_server.stop_kleio_server(container: Container | None = None)[source]: Stop kleio server

timelink.kleio.schemas module

class timelink.kleio.schemas.ApiPermissions(value)[source]: An enumeration.

class timelink.kleio.schemas.KleioFile(*, path: str, name: str, size: int, directory: str, modified: datetime, modified_iso: datetime, modified_string: str, qtime: datetime, qtime_string: str, source_url: str, status: translation_status_enum, translated: datetime | None = None, translated_string: str | None = None, errors: int | None = None, warnings: int | None = None, version: str | None = None, rpt_url: str | None = None, xml_url: str | None = None, import_status: import_status_enum | None = None, import_errors: int | None = None, import_warnings: int | None = None, import_error_rpt: str | None = None, import_warning_rpt: str | None = None, imported: datetime | None = None, imported_string: str | None = None)[source]

Represents the information about a kleio file and its translation and import status

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'directory': FieldInfo(annotation=str, required=True, description='The directory containing the file'), 'errors': FieldInfo(annotation=Union[int, NoneType], required=False, description='The number of errors encountered during translation'), 'import_error_rpt': FieldInfo(annotation=Union[str, NoneType], required=False, description='Error report from import'), 'import_errors': FieldInfo(annotation=Union[int, NoneType], required=False, description='The number of errors encountered during import'), 'import_status': FieldInfo(annotation=Union[import_status_enum, NoneType], required=False, description='The status of the file import:\n I = imported\n E = imported with error\n W = imported with warnings no errors\n N = not imported\n U = translation updated need to reimport'), 'import_warning_rpt': FieldInfo(annotation=Union[str, NoneType], required=False, description='Warning report from import'), 'import_warnings': FieldInfo(annotation=Union[int, NoneType], required=False, description='The number of warnings encountered during import'), 'imported': FieldInfo(annotation=Union[datetime, NoneType], required=False, description='Date of import of the file'), 'imported_string': FieldInfo(annotation=Union[str, NoneType], required=False, description='Date of import of the file as a string'), 'modified': FieldInfo(annotation=datetime, required=True, description='The last modified time of the file'), 'modified_iso': FieldInfo(annotation=datetime, required=True, description='The last modified time of the file in ISO format'), 'modified_string': FieldInfo(annotation=str, required=True, description='The last modified time of the file as a string'), 'name': FieldInfo(annotation=str, required=True, description='The name of the file'), 'path': FieldInfo(annotation=str, required=True, description='The path of the file'), 'qtime': FieldInfo(annotation=datetime, required=True, description='The time the file was queued for translation'), 'qtime_string': FieldInfo(annotation=str, required=True, description='The time the file was queued for translation as a string'), 'rpt_url': FieldInfo(annotation=Union[str, NoneType], required=False, description='The URL of the report file'), 'size': FieldInfo(annotation=int, required=True, description='The size of the file in bytes'), 'source_url': FieldInfo(annotation=str, required=True, description='The URL of the source file'), 'status': FieldInfo(annotation=translation_status_enum, required=True, description='The status of the file:\n V = valid translations\n T = need translation (source more recent than translation)\n E = translation with errors\n W = translation with warnings\n P = translation being processed\n Q = file queued for translation'), 'translated': FieldInfo(annotation=Union[datetime, NoneType], required=False, description='The time the file was translated'), 'translated_string': FieldInfo(annotation=Union[str, NoneType], required=False, description='The time the file was translated as a string'), 'version': FieldInfo(annotation=Union[str, NoneType], required=False, description='The version of the kleio translator'), 'warnings': FieldInfo(annotation=Union[int, NoneType], required=False, description='The number of warnings encountered during translation'), 'xml_url': FieldInfo(annotation=Union[str, NoneType], required=False, description='The URL of the XML file')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

needs_import()[source]

Return True if the file needs import

A file needs import if it has not been imported before: or if it has been translated again since the last import

needs_translation()[source]: Return True if the file needs translation

source_url: str: The status of the file: .. code-block:: python

V = “valid translations” T = “need translation (source more recent than translation)” E = “translation with errors” W = “translation with warnings” P = “translation being processed” Q = “file queued for translation”

class timelink.kleio.schemas.TokenInfo(*, comment: str, api: list[ApiPermissions], structures: str, sources: str)[source]

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'api': FieldInfo(annotation=list[ApiPermissions], required=True, description='API permissions'), 'comment': FieldInfo(annotation=str, required=True, description='Comment'), 'sources': FieldInfo(annotation=str, required=True, description='Sources'), 'structures': FieldInfo(annotation=str, required=True, description='Structures')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class timelink.kleio.schemas.import_status_enum(value)[source]: An enumeration.

class timelink.kleio.schemas.translation_status_enum(value)[source]: An enumeration.

timelink.kleio.utilities module

Kleio Groups are the building blocks for transcription of historical sources.

timelink.kleio.utilities.kleio_escape(v: str) → str[source]

Checks for Kleio special characters and quotes if needed:

>>> print(kleio_escape('normal string'))
normal string
>>> print(kleio_escape('oops we have a / in the middle'))
"oops we have a / in the middle"

timelink.kleio.utilities.quote_long_text(txt, initial_indent=' ', indent=' ', width=2048) → str[source]

Surround long text with triple quotes, wraps and indents lines if needed. Some of the parameters are passed on to textwrap.fill().

Sphynx style markup

Parameters:

txt (str) – The text to be transformed
initial_indent (str) – string to ident the first line of paragraphs. Default is 4 spaces. See textwrap.fill().
indent (str) – string to ident the wrap lines of paragraphs (after the first). Default is 2 spaces. See textwrap.fill().
width (int) – width of line for wrapping. See textwrap.fill().

Return type:

str