py4ai.data.model.core module

Module with base abstraction of common objects.

class py4ai.data.model.core.BaseIterable(*args, **kwds)

Bases: Generic[T]

Abstract class defining interface for iterables.

abstract property cached: bool

Whether the iterable is cached in memory or lazy.

Returns: boolean indicating whether iterable is fully-stored in memory

abstract classmethod empty() → BaseIterableType: Return an empty iterable instance.

abstract property items: Iterable[T]

Return an iterator over the items.

Returns: Iterable[T]

abstract property type: Type[T]: Return the type of the objects in the Iterable.

class py4ai.data.model.core.BaseRange

Bases: ABC

Abstract Range Class.

property business_days: List[Timestamp]

Create date range with daily frequency.

Returns: list of pd.Timestamp from start to end with daily frequency including only days from Mon to Fri

property days: List[Timestamp]

Create date range with daily frequency.

Returns: list of pd.Timestamp from start to end with daily frequency

abstract property end: Timestamp

Return the last timestamp.

Returns: Timestamp

property minutes_15: List[Timestamp]

Create date range with daily frequency.

Returns: list of pd.Timestamp from start to end with 15 minutes frequency

abstract overlaps(other: BaseRange) → bool

Return whether two ranges overlaps.

Parameters: other – other range to be compared with
Returns: True if the two ranges intersect, False otherwise

abstract range(freq: str = 'H') → List[Timestamp]

Return list of timestamps, spaced by given frequency.

Parameters: freq – frequency of timestamps, valid values are “D” (day), “H” (hours), “M”(minute), “S” (seconds).
Returns: list of timestamps

abstract property start: Timestamp

Return the first timestamp.

Returns: Timestamp

class py4ai.data.model.core.CachedIterable(items: Sequence[T])

Bases: BaseIterable[T]

Base class to be used for implementing cached iterables.

Return instance of a class to be used for implementing cached iterables.

Parameters: items – sequence or iterable of elements

property cached: bool

Whether the iterable is cached in memory or lazy.

Returns: boolean indicating whether iterable is fully-stored in memory

classmethod empty() → CachedIterableType

Return an empty cached iterable.

Returns: Empty instance

classmethod from_iterable(iterable: BaseIterable[T]) → CachedIterableType

Create a new instance of this class from a BaseIterable instance.

Parameters: iterable – iterable instance
Returns: cached iterable

property items: Sequence[T]

Return an iterator over the items.

Returns: Iterable[T]

class py4ai.data.model.core.CompositeRange(ranges: List[Range])

Bases: BaseRange

Class representing a composition of ranges.

Return a range made up of multiple ranges.

Parameters: ranges – List of Ranges

property end: Timestamp

Return the last timestamp.

Returns: Timestamp

overlaps(other: BaseRange) → bool

Return whether two ranges overlaps.

Parameters: other – BaseRange, other range to be compared with
Returns: bool, True if the two ranges intersect, False otherwise

range(freq: str = 'H') → List[Timestamp]

Return list of timestamps, spaced by given frequency.

Parameters: freq – given frequency
Returns: list of timestamps

simplify() → Union[CompositeRange, Range]

Simplify the list into disjoint Range objects, aggregating non-disjoint ranges.

If only one range would be present, a simple Range object is returned.

Returns: BaseRange

property start: Timestamp

Return the first timestamp.

Returns: Timestamp

class py4ai.data.model.core.DillSerialization

Bases: Serializable

Serialization based on dill package.

classmethod load(filename: Union[str, os.PathLike[str]]) → Any

Load instance from file.

Parameters: filename – Name of the file to be read
Returns: Instance of the read Model

write(filename: Union[str, os.PathLike[str]]) → None

Write instance as pickle.

Parameters: filename – Name of the file where to save the instance

class py4ai.data.model.core.IterGenerator(generator_function: Callable[[], Iterator[T]], _type: Optional[Type[T]] = None)

Bases: Generic[T]

Base class representing any generator.

Class that allows a given generator to be accessed as an Iterator via .iterator property.

Parameters

generator_function – function that outputs a generator
_type – type returned by the generartor, required when the generator is empty

Raises

TypeError – when type mismatch happens between generator and provided type
ValueError – when an empty generator is provided without _type specification

property iterator: Iterator[T]

Return an iterator over the given generator function.

Returns: an iterator

class py4ai.data.model.core.IterableUtilsMixin(*args, **kwargs)

Bases: Generic[T, LazyIterableType, CachedIterableType], BaseIterable[T], ABC

Class to provide base interfaces and methods for enhancing iterables classes and enable more functional approaches.

In particular, the class provides among others implementation for map, filter and foreach methods.

Create a new instance of this class.

Parameters

cls – parent object class
args – passed to the super class __new__ method
kwargs – passed to the super class __new__ method

Raises

RuntimeError – if the cached and lazy versions were not defined before instantiating the class

Returns

an instance of this class

batch(size: int = 100) → Iterator[CachedIterableType]

Return an iterator of batches of size size.

Parameters: size – dimension of the batch
Yield: iterator of batches

cached_type: Type[CachedIterableType]

filter(f: Callable[[T], bool]) → LazyIterableType

Return an iterable where elements have been filtered based on a boolean function.

Parameters: f – boolean function that selects items
Returns: lazy iterable with elements filtered

foreach(f: Callable[[T], Any]) → None

Execute the provided function on each element of the iterable.

Parameters: f – function to be executed for each element

from_element(value: T, cached: bool = True) → Union[LazyIterableType, CachedIterableType]

Instantiate a new object of this class from a single element.

Parameters

value – element
cached – whether a cached iterable should be returned, defaults to True

Returns

iterable object

lazy_type: Type[LazyIterableType]

map(f: Callable[[T], T_co]) → LazyIterableType

Map all elements of an iterable with the provided function.

Parameters: f – function to be used to map the elements
Returns: mapped iterable

take(size: int) → CachedIterableType

Take the first n elements of the iterables.

Parameters: size – number of elements to be taken
Returns: cached iterable with the first elements

to_cached() → CachedIterableType

Create a new cached instance of this instance.

Returns: cached iterable

to_lazy() → LazyIterableType

Create a new lazy instance of this instance.

Returns: lazy iterable

class py4ai.data.model.core.LazyIterable(items: IterGenerator[T])

Bases: BaseIterable[T]

Base class to be used for implementing lazy iterables.

Return an instance of the class to be used for implementing lazy iterables.

Parameters: items – IterGenerator containing the generator of items

property cached: bool

Whether the iterable is cached in memory or lazy.

Returns: boolean indicating whether iterable is fully-stored in memory

classmethod empty() → LazyIterableType

Return an empty lazy iterable.

Returns: Empty instance

classmethod from_iterable(iterable: BaseIterable[T]) → LazyIterableType

Create a new instance of this class from a BaseIterable instance.

Parameters: iterable – iterable instance
Returns: lazy iterable

property items: Iterator[T]

Return an iterator over the items.

Returns: Iterable[T]

class py4ai.data.model.core.PickleSerialization

Bases: Serializable

Serialization based on pickle package.

classmethod load(filename: Union[str, os.PathLike[str]]) → Any

Load instance from pickle.

Parameters: filename – Name of the file to be read
Returns: Instance of the read Model

write(filename: Union[str, os.PathLike[str]]) → None

Write instance as pickle.

Parameters: filename – Name of the file where to save the instance

class py4ai.data.model.core.Range(start: Union[float, str, datetime], end: Union[float, str, datetime])

Bases: BaseRange

Base class for a continuous range.

Return a simple Range Class.

Parameters

start – starting datetime for the range
end – ending datetime for the range

Raises

ValueError – if start > end

property end: Timestamp

Return the last timestamp.

Returns: Timestamp

overlaps(other: BaseRange) → bool

Return whether two ranges overlaps.

Parameters: other – other range to be compared with
Returns: True or False whether the two overlaps

range(freq: str = 'H') → List[Timestamp]

Return list of timestamps, spaced by given frequency.

Parameters: freq – given frequency
Returns: list of timestamps

property start: Timestamp

Return the first timestamp.

Returns: Timestamp

class py4ai.data.model.core.RegisterLazyCachedIterables(class_object_first: Type[IterableUtilsMixin[T, LazyIterableType, CachedIterableType]], unidirectional_link: bool = False)

Bases: Generic[T, LazyIterableType, CachedIterableType]

Register the lazy and cached version of the iterables.

Initialize an instance of this class.

Parameters

class_object_first – the first iterable class object (the lazy or chached version)
unidirectional_link – if True, only set the link in the second class passed to the __call__ method

static register_cached(class_object_lazy: Type[IterableUtilsMixin[T, LazyIterableType, CachedIterableType]], class_object_cached: Type[IterableUtilsMixin[T, LazyIterableType, CachedIterableType]]) → None

Link the lazy and cached versions.

Parameters

class_object_lazy – the lazy iterable class object
class_object_cached – the cached iterable class object

static register_lazy(class_object_lazy: Type[IterableUtilsMixin[T, LazyIterableType, CachedIterableType]], class_object_cached: Type[IterableUtilsMixin[T, LazyIterableType, CachedIterableType]]) → None

Link the lazy and cached versions.

Parameters

class_object_lazy – the lazy iterable class object
class_object_cached – the cached iterable class object

class py4ai.data.model.core.Serializable

Bases: ABC

Abstract Class to be used to extend objects that can be serialised.

abstract classmethod load(filename: Union[str, os.PathLike[str]]) → Any

Load class from a file.

Parameters: filename – filename

abstract write(filename: Union[str, os.PathLike[str]]) → None

Write class to a file.

Parameters: filename – filename