cornflow-client¶

Core classes¶

ApplicationCore¶

Base code for the application core.

class cornflow_client.core.application.ApplicationCore¶

Bases: ABC

The application template.

abstract property name: str¶

Mandatory property

Returns:: the name of the class.

property description: str | None¶

Optional property

Returns:: the description of the class

property default_args: Dict | None¶

Optional property

Returns:: the default args for the DAG

property extra_args: Dict | None¶

Optional property

Returns:: dictionary with optional arguments for the DAG

property default_args_checks_kpis: Dict | None¶

Optional property

Returns:: the default args for the DAG for checks and kpis generation

property extra_args_checks_kpis: Dict | None¶

Optional property

Returns:: dictionary with optional arguments for the DAG for checks and kpis generation

abstract property instance: Type[InstanceCore]¶

Mandatory property

Returns:: the constructor for the instance.

abstract property solution: Type[SolutionCore]¶

Mandatory property

Returns:: the constructor for the solution.

abstract property schema: dict¶

Mandatory property

Returns:: the configuration schema used for the solve() method.

abstract property test_cases: List[Dict[str, str | Dict]]¶

Mandatory property

Returns:

a list of datasets following the json-schema. if each element in the list is:

dict: each element is an instance
tuple: the first part is the instance, the second its solution

it can also return a list of dicts, where the keys are:

name: name of the test case.
description optional field with a description of the test case.
instance: the instance data.
solution: the solution data (optional)

abstract property solvers: Dict[str, Type[ExperimentCore]]¶

Mandatory property

Returns:: a dictionary of constructors for solution methods for this particular problem.

solve(data: dict, config: dict, solution_data: dict = None) → Tuple[Dict, Dict | None, Dict | None, Dict | None, str, Dict]¶

Solves the problem instance using the specified configuration.

Parameters:

data – Dictionary representing the problem instance.
config – Dictionary containing execution configuration.
solution_data – Optional dictionary with an initial solution.

Returns:

Tuple containing (solution, solution_checks, instance_checks, log_txt, log_json).

check_generate_kpis(instance_data: dict, solution_data: dict = None) → Tuple[Dict, Dict, Dict, Dict]¶: Checks the instance and solution data and generates KPIs :param instance_data: json data of the instance :param solution_data: json data of the solution :return: instance checks, solution checks, kpis, log

get_solver(name: str = 'default') → Type[ExperimentCore] | None¶

Parameters:: name – name of the solver to find
Returns:: the constructor for a solver matching the name

get_default_solver_name() → str¶

Returns:: the name of the default solver

get_schemas() → Dict[str, Dict]¶

Returns:: a dictionary with the three schemas that define the solution method

InstanceCore¶

class cornflow_client.core.instance.InstanceCore(data: dict)¶

Bases: InstanceSolutionCore, CheckCore, ABC

The instance template.

data_checks() → dict¶

Method that executes the InstanceCore.check() method and validates the result against the schema_checks

Returns:: The dictionary returned by the InstanceCore.check() method
Return type:: dict
Raises:: BadInstanceChecks – if the instance checks do not match the schema
Author:: baobab soluciones

check() → dict¶

Method that checks if there are inconsistencies in the data of the instance and if the problem is feasible

This method can be overridden if necessary

Returns:: An dictionary containing the inconsistencies found and indicating if

the problem is infeasible

abstract property schema_checks: dict¶: A dictionary representation of the json-schema for the dictionary returned by the method Instance.check()

SolutionCore¶

class cornflow_client.core.solution.SolutionCore(data: dict)¶

Bases: InstanceSolutionCore, ABC

The solution template.

InstanceSolution¶

class cornflow_client.core.instance_solution.InstanceSolutionCore(data: dict)¶

Bases: ABC

Common interface for the instance and solution templates

property data: dict¶: input data (not necessarily in json-format)

classmethod from_dict(data: dict) → InstanceSolutionCore¶

Parameters:: data – json-schema in a dictionary format
Returns:: an object initialized from the dict json-schema

to_dict() → dict¶

Returns:: a dictionary with the json-schema representation

classmethod from_json(path: str) → InstanceSolutionCore¶

Parameters:: path – path to json-schema json file
Returns:: an object initialized from the json-schema formatted json file

to_json(path: str) → None¶

Parameters:: path – path to json-schema json file

writes a json file with the json-schema representation of the object

abstract property schema: dict¶: a dictionary representation of the json-schema for the object

check_schema() → List¶

checks that the json-schema export complies with the defined schema

Returns:: a list of errors

generate_schema() → dict¶

Returns:: a dict json-schema based on the current data

classmethod from_excel(path: str) → InstanceSolutionCore¶

Read an entire excel file.

Parameters:: path – path of the excel file
Returns:: a dict with a list of dict (records format) for each table.

to_excel(path: str)¶

Write data to excel.

Parameters:: path – path or name of the excel file
Returns:: nothing

static dict_to_int_or_float(data_dict)¶

Recursively transforms a dictionary (and nested structures) by converting string representations of numbers into actual int or float types. Returns a new dictionary with converted values, does not modify the original.

For example: Transforms {a: ‘4’, b: {c: ‘7’, d: [‘8.7’, ‘9’]}}: into {a: 4, b: {c: 7, d: [8.7, 9]}}

static from_element_or_list_to_dict(element_or_list)¶: Converts a list into a dictionary indexed by the field ‘index’ of each element of the list. If the input is not a list, it is converted into a list before converting to a dictionary For example: [{‘index’: 4, ‘value’: 5}, {‘index’: 7, ‘value’: 8}]

is transformed to {4: {‘index’: 4, ‘value’: 5}, 7: {‘index’: 7, ‘value’: 8}}

static get_date_from_string(string: str) → datetime¶: Returns a datetime object from an hour-string in format ‘YYYY-MM-DD’

static get_datetime_from_string(string: str) → datetime¶: Returns a datetime object from an hour-string in format ‘YYYY-MM-DDTh:m’

static get_datetimesec_from_string(string: str) → datetime¶: Returns a datetime object from an hour-string in format ‘YYYY-MM-DDTh:m:s’

static get_datetime_from_date_hour(date: str, hour: int) → datetime¶: Returns a datetime object from a date and an hour

static get_date_hour_from_string(string: str, zero_to_twenty_four=False)¶: Returns a tuple (date, hour) from an hour-string

static get_date_string_from_ts(ts: datetime) → str¶: Returns the string of a given date as ‘YYYY-MM-DD’

static get_datetime_string_from_ts(ts: datetime) → str¶: Returns the string of a given date as ‘YYYY-MM-DDTh:m’

static get_datetimesec_string_from_ts(ts: datetime) → str¶: Returns the string of a given date as ‘YYYY-MM-DDTh:m:s’

static get_next_hour_datetime_string(string: str) → str¶: Returns the hour following the given hour, as a string

static get_next_hour_datetimesec_string(string: str) → str¶: Returns the hour following the given hour, as a string

static get_next_hour(ts: datetime) → datetime¶: Returns the hour following the given hour

static get_previous_hour_datetime_string(string: str) → str¶: Returns the hour preceding the given hour, as a string

static get_previous_hour_datetimesec_string(string: str) → str¶: Returns the hour preceding the given hour, as a string

static get_previous_hour(ts: datetime) → datetime¶: Returns the hour preceding the given hour

static get_date_string_from_ts_string(ts: str) → str¶: Returns the date in format ‘YYYY-MM-DD’ from a datetime string

static get_hour_from_ts(ts: datetime) → float¶: Returns the hours (in number) of the given time slot

static add_time_to_ts(ts: datetime, weeks=0, days=0, minutes=0, seconds=0) → datetime¶: Adds time to a datetime

static add_time_to_date_string(string: str, weeks=0, days=0, minutes=0, seconds=0) → str¶: Adds time to a date string

static add_time_to_datetime_string(string: str, weeks=0, days=0, minutes=0, seconds=0) → str¶: Adds time to a datetime

static add_time_to_datetimesec_string(string: str, weeks=0, days=0, hours=0, minutes=0, seconds=0) → str¶: Adds time to a datetime

static get_week_from_ts(ts: datetime) → int¶: Returns the integer value of the week for the given time slot

static get_week_from_date_string(string: str) → int¶: Returns the integer value of the week for the given string

static get_week_from_datetime_string(string: str) → int¶: Returns the integer value of the week for the given string

static get_week_from_datetimesec_string(string: str) → int¶: Returns the integer value of the week for the given string

static get_weekday_from_ts(ts: datetime) → int¶: Returns the number of the weekday from a ts

static get_weekday_from_date_string(string: str) → int¶: Returns the number of the weekday from a date string in format ‘YYYY-MM-DD’

static get_weekday_from_datetime_string(string: str) → int¶: Returns the number of the weekday from a date string in format ‘YYYY-MM-DDTh:m’

static get_weekday_from_datetimesec_string(string: str) → int¶: Returns the number of the weekday from a date string in format ‘YYYY-MM-DDT:h:m:s’

static get_hour_from_datetime_string(string: str) → float¶: Returns the integer value of the hour (in number) from ts string in format ‘YYYY-MM-DDTh:m’

static get_hour_from_datetimesec_string(string: str) → float¶: Returns the integer value of the hour (in number) from ts string in format ‘YYYY-MM-DDTh:m:s’

ExperimentCore¶

Base code for the experiment template.

class cornflow_client.core.experiment.ExperimentCore(instance: InstanceCore, solution: SolutionCore | None = None)¶

Bases: CheckCore, ABC

The solver template.

property instance: InstanceCore¶

Returns:: the instance

property solution: SolutionCore¶

Returns:: the solution

abstractmethod solve(options: dict) → dict¶

Mandatory method

Parameters:: options – configuration for solving the problem
Returns:: a dictionary with status codes and other information

This method produces and stores a solution

abstractmethod get_objective() → float¶

Mandatory method

Returns:: the value of the current solution, represented by a number

data_checks() → dict¶: Method that executes the ExperimentCore.check() method and validates the result against the schema_checks

get_kpis() → dict¶: Method that generates KPIs for the solution and validates the result against the schema_kpis

kpis_checks() → dict¶: Method that checks the solution with the KPIs and validates the result against the schema_checks.

check() → Dict[str, List | Dict]¶

Method that runs all the checks for the solution.

This method can be overridden by the user to modify the behaviour of the checks if wanted.

Returns:: a dictionary of dictionaries. Each dictionary represents one type of error. Each of the elements inside represents one error of that particular type.

check_solution() → Dict[str, List | Dict]¶

Method that runs all the checks for the solution. This method will be deprecated on cornflow-client version 2.0.0

This method can be overridden by the user to modify the behaviour of the checks if wanted.

Returns:: a dictionary of dictionaries. Each dictionary represents one type of error. Each of the elements inside represents one error of that particular type.

abstract property schema_checks: dict¶

A dictionary representation of the json-schema for the dictionary returned by: the method ExperimentCore.check_solution()

generate_kpis() → Dict[str, List | Dict]¶

Method that generates KPIs for the solution. By default, launches all methods starting with kpis_ and returns the result in a dictionary.

This method can be overridden by the user to modify the behaviour of the KPI generation if wanted.

Returns:: a dictionary of lists of dictionaries. Each list represents a table of KPIs. Each of the elements inside represents one row of that particular table.

check_kpis() → dict¶: Method that checks the solution with the KPIs. By default, it does not perform any check. This method can be overridden by the user to modify the behavior of the KPI checks if wanted. :return: a dictionary of lists of dictionaries. Each list represents a table of KPI checks. Each of the elements inside represents one row of that particular table.

kpis_to_dict() → dict¶: Method that transforms the KPIs generated in self.kpis into a dictionary. By default, it returns self.kpis. This method can be overridden by the user to modify the behavior of the KPI transformation if wanted. :return: a dictionary with the KPIs. The format of the dictionary should be validated against self.schema_kpis

generate_output_files(instance_checks, instance_has_errors, solution_checks, solution_has_errors) → Dict[str, str | BytesIO] | None¶

Method that generates the output files. This method can be overriden by the user. :return: a dictionary, where: - the keys are the names of the files or directories - the values are either:

an io.BytesIO instance

an io.StringIO instance

a string representing an absolute path to a file or directory.

If the values are paths, the files or directories will be deleted after generating the zip file.

static get_solver_config(config, lib='pyomo', default_solver='cbc', remove_unknown=False)¶

Format the configuration used to solve the problem. Solver configuration can either be directly in config using cornflow mapping name

or in a config[“solver_config”] using the solver names. Nothing is ever removed from solver_config.

Example:

config = {: “solver”:”milp.cbc”, “time_limit”:60, “rel_gap”:0.1, “solver_config”:{“heur”:1, “pumpC”:0}

}

Parameters:

config – dict config argument of the solver method
lib – str library used to create the model (pulp or pyomo)
default_solver – str default solver to use if none is present inf config.
remove_unknown – bool. if True, the unknown parameters will be deleted. Otherwise, they will remain but will not be translated.

Returns:

the solver name and the config dict.

cornflow client¶

Constants¶

Solving status¶

cornflow_client.constants.STATUS_NOT_SOLVED¶

cornflow_client.constants.STATUS_OPTIMAL¶

cornflow_client.constants.STATUS_INFEASIBLE¶

cornflow_client.constants.STATUS_UNBOUNDED¶

cornflow_client.constants.STATUS_UNDEFINED¶

cornflow_client.constants.STATUS_FEASIBLE¶

cornflow_client.constants.STATUS_MEMORY_LIMIT¶

cornflow_client.constants.STATUS_NODE_LIMIT¶

cornflow_client.constants.STATUS_TIME_LIMIT¶

cornflow_client.constants.STATUS_LICENSING_PROBLEM¶

cornflow_client.constants.STATUS_QUEUED¶

Constant	Value
`cornflow_client.constants.STATUS_NOT_SOLVED`	0
`cornflow_client.constants.STATUS_OPTIMAL`	1
`cornflow_client.constants.STATUS_INFEASIBLE`	-1
`cornflow_client.constants.STATUS_UNBOUNDED`	-2
`cornflow_client.constants.STATUS_UNDEFINED`	-3
`cornflow_client.constants.STATUS_FEASIBLE`	2
`cornflow_client.constants.STATUS_MEMORY_LIMIT`	3
`cornflow_client.constants.STATUS_NODE_LIMIT`	4
`cornflow_client.constants.STATUS_TIME_LIMIT`	5
`cornflow_client.constants.STATUS_LICENSING_PROBLEM`	-5
`cornflow_client.constants.STATUS_QUEUED`	-7

Solution status¶

cornflow_client.constants.SOLUTION_STATUS_INFEASIBLE¶

cornflow_client.constants.SOLUTION_STATUS_FEASIBLE¶

Constant	Value
`cornflow_client.constants.SOLUTION_STATUS_INFEASIBLE`	0
`cornflow_client.constants.SOLUTION_STATUS_FEASIBLE`	2

Constants values used in schemas functions.

exception cornflow_client.constants.OrchError(error=None, status_code=None, payload=None, log_txt=None)¶

Bases: Exception

status_code = 400¶

to_dict()¶

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.AirflowError(error=None, status_code=None, payload=None, log_txt=None)¶

Bases: OrchError

log_txt = 'Airflow error'¶

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

status_code = 400¶

to_dict()¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.DatabricksError(error=None, status_code=None, payload=None, log_txt=None)¶

Bases: OrchError

log_txt = 'Databricks error'¶

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

status_code = 400¶

to_dict()¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.NoSolverException¶

Bases: Exception

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.BadConfiguration¶

Bases: Exception

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.BadInstance¶

Bases: Exception

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.BadSolution¶

Bases: Exception

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.BadInstanceChecks¶

Bases: Exception

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.BadSolutionChecks¶

Bases: Exception

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

exception cornflow_client.constants.BadKPIs¶

Bases: Exception

add_note()¶: Exception.add_note(note) – add a note to the exception

args¶

with_traceback()¶: Exception.with_traceback(tb) – set self.__traceback__ to tb and return self.

Schemas¶

Schema utility functions¶

cornflow_client.schema.tools.get_pulp_jsonschema(filename='pulp_json_schema.json', path='data')¶: returns the PuLP model schema

cornflow_client.schema.tools.get_empty_schema(properties=None, solvers=None)¶: assumes the first solver is the default

cornflow_client.schema.tools.clean_none(dic)¶

Remove empty values from a dict

Parameters:: dic – a dict
Returns:: the filtered dict

cornflow_client.schema.tools.check_fk(fk_dic)¶

Check the format of foreign keys

Parameters:: fk_dic – a dict of foreign keys values
Returns:: None (raise an error if problems are detected)

cornflow_client.schema.tools.schema_from_excel(path_in, param_tables=None, path_out=None, fk=False, date_format=False, path_methods=None, path_access=None)¶

Create a jsonschema based on an Excel data file.

Parameters:

path_in – path of the Excel file
param_tables – array containing the names of the parameter tables
path_out – path where to save the json schema as a json file.
fk – True if foreign key are described in the second row.
date_format – if format is true special format (like date, time or datetime) are specified in the third row.
path_methods – path where to save the methods dict as a json file
path_access – path where to save the access dict as a json file

Returns:

the jsonschema

cornflow_client.schema.tools.add_details(name, details, schema)¶

Add a detail attribute to a json schema property.

Example: add_details(“foreign_key”, {first_table:{“name”:”other_table.name”}}, schema) # generate:

“name”: {
“type”: “string” “foreign_key”: “other_table.name”

}

Parameters:

name – name of the attribute to add
details – dict of dict in format {table:{column_name:value}}
schema – schema to update

Returns:

None

cornflow_client.schema.tools.str_key(dic)¶

Apply str to the keys of a dict. This must be a applied to a dict in order to transform it into json.

Parameters:: dic – a dict
Returns:: the dict with keys as strings.

cornflow_client.schema.tools.str_columns(table)¶

Transform the columns of a table (the keys of a list of dict) into strings.

Parameters:: table – a list of dict.
Returns:: the modified list of dict

cornflow_client.schema.tools.fix_required(schema)¶

Fix required property in schema: if a field is allowed null, it is not required

Parameters:: schema – the json schema
Returns:: None

Schema manager¶

Class to help create and manage data schema and to validate json files.

class cornflow_client.schema.manager.SchemaManager(schema, validator=<class 'jsonschema.validators.Draft7Validator'>)¶

Bases: object

A schema manager between json-schema, dict-schema and marshmallow

Class to help create and manage data schema. Once a schema is loaded, allow the validation of data.

Parameters:: schema – a json schema

classmethod from_filepath(path)¶

Load a json schema from a json file.

:param path the file path

return The SchemaManager instance

get_jsonschema()¶: Return a copy of the stored jsonschema.

get_validation_errors(data)¶

Validate json data according to the loaded jsonschema and return a list of errors. Return an empty list if data is valid.

Parameters:: data (dict) – data to validate.
Returns:: A list of validation errors.

For more details about the error format, see: https://python-jsonschema.readthedocs.io/en/latest/errors/#jsonschema.exceptions.ValidationError

validate_data(data, print_errors=False)¶

Validate json data according to the loaded jsonschema.

Parameters:

data (dict) – the data to validate.
print_errors (bool) – If true, will print the errors.

Returns:

True if data format is valid, else False.

validate_schema(print_errors=False)¶

Validate the loaded jsonschema :param bool print_errors: If true, will print the errors

Returns:: True if jsonschema is valid, else False

get_file_errors(path)¶

Get json file errors according to the loaded jsonschema.

:param path the file path

Returns:: A list of validation errors. For more details about the error format, see: https://python-jsonschema.readthedocs.io/en/latest/errors/#jsonschema.exceptions.ValidationError

validate_file(path, print_errors=False)¶

Validate a json file according to the loaded jsonschema.

:param path the file path :param print_errors: If true, will print the errors.

Returns:: True if the data is valid and False if it is not.

to_dict_schema()¶

Transform a jsonschema into a dictionary format

Returns:: The schema dictionary

to_schema_dict_obj()¶: Returns an DictSchema object equivalent of the jsonschema

to_marshmallow()¶

Create marshmallow schemas

Returns:: a dict containing the flask marshmallow schemas
Return type:: Schema()

export_schema_dict(path)¶

Print the schema_dict in a json file.

Parameters:: path – the path where to save the dict.format
Returns:: nothing

draft_schema_from(path, save_path=None)¶

Create a draft jsonschema from a json file of data.

Parameters:

path – path to the json file.
save_path – path where to save the generated schema.

Returns:

the generated schema.

to_template()¶: This function assumes certain structure for the jsonschema. For now, three types of tables exist: array of objects, arrays and objects. { table1: [{col1: a, col2: b}, {col1: aa, col2: bb}, …], table2: [1, 2, 3, ], table3: {config1: a, config2: b}, }

static load_json(path)¶

Load a json file

Parameters:: path – the path of the json file.json

return the json content.

dict_to_flask()¶

Create marshmallow schemas

Returns:: a dict containing the flask marshmallow schemas
Return type:: Schema()

classmethod load_schema(path)¶

Load a json schema from a json file.

:param path the file path

return The SchemaManager instance

jsonschema_to_flask()¶

Create marshmallow schemas

Returns:: a dict containing the flask marshmallow schemas
Return type:: Schema()

jsonschema_to_dict()¶

Transform a jsonschema into a dictionary format

Returns:: The schema dictionary

Dictionary schema¶

class cornflow_client.schema.dictSchema.DictSchema(jsonschema)¶

Bases: object

A json-schema to dict-schema parser

Class to manage internal dictionary schema

Parameters:: jsonschema – a json schema

static get_empty_schema()¶: Create un empty schema dict