import abc
import logging
import traceback
import typing
import datetime
from divera.api.v2.pull import All as _All
from divera.api.v2 import models as _models
[docs]class Model(abc.ABC):
sorting_key: str = 'id'
@property
def id(self) -> int:
return self.data['id']
@property
def cluster_id(self) -> int:
return self.data['cluster_id']
def __init__(self,
data: dict = None,
):
super().__init__()
if data is None:
data = {}
self.data = data
def __getattribute__(self, item):
try:
# Try the usual way
return super().__getattribute__(item)
except AttributeError as original_error:
try:
# There is a chance for the value to be included in `self.data`, but not as `self.<<attribute>>`.
# This might happen if the version of either the instance
# or the library is not up-to-date with the other.
res = self.data[item]
logging.debug(
f"{self.__class__.__name__}.{item} has been accessed, but is not available as property yet."
)
return res
except KeyError:
raise original_error
def __eq__(self, other):
return self.__getattribute__(self.sorting_key) == other.__getattribute__(other.sorting_key)
def __repr__(self):
res = '<' + ' '.join([
type(self).__name__,
str(self.sorting_key(self) if callable(self.sorting_key) else self.data[self.sorting_key]),
f'"{self.title}"'
if 'title' in self.data
and self.title is not None
else f'"{self.name}"'
if 'name' in self.data
and self.name is not None
else '',
]).strip() + '>'
return res
def __sub__(self, other):
different_attributes = []
for x in self.data.keys():
try:
if self.data[x] != other.data[x]:
different_attributes.append(x)
except Exception as e:
different_attributes.append(x)
logging.error(f'{type(e).__name__} with objects of types {type(self)} and {type(other)}')
logging.error(''.join(traceback.format_exception(e)))
return different_attributes
[docs] @staticmethod
@abc.abstractmethod
def get_all(
):
pass
[docs]class APIGetModel(
Model,
abc.ABC,
):
[docs] def get(
self,
obj=None,
object_id=None,
):
request = _models.Get(
obj=obj or (type(self)(id=int(object_id)) if object_id is not None else None) or self,
)
request.process_result = lambda result: type(self)(data=result['data'])
return request
[docs]class APIConfirmModel(
Model,
abc.ABC,
):
[docs] def confirm(
self,
response_id: int,
response_text: str = None,
obj=None,
object_id: str = None,
):
request = _models.Confirm(
obj=obj or (type(self)(id=int(object_id)) if object_id is not None else None) or self,
response_id=response_id,
response_text=response_text,
)
return request
[docs]class APIReadModel(
Model,
abc.ABC,
):
[docs] def read(
self,
obj=None,
object_id=None,
):
request = _models.Read(
obj=obj or (type(self)(id=int(object_id)) if object_id is not None else None) or self,
)
request.process_result = lambda result: result['success']
return request
[docs]class AttachmentModel(
abc.ABC,
):
[docs] def add_attachment(
self,
file,
file_name: str = None,
description: str = None,
):
"""
Attach a file to the object.
.. code-block:: python
obj.add_attachment(
open('path/to/file.png', 'rb').read(),
'my_file.png',
'my image'
)
:param file: The binary data of your file.
:param file_name: A custom name for your file.
:param description: A description.
:return:
"""
request = _models.Attachment(
obj=self,
file=file,
file_name=file_name,
description=description,
)
return request
[docs]class PullModel(Model, abc.ABC):
"""
Parent class for all models that get their data from pull_data
"""
path: typing.List = [
]
path_to_sorting: typing.List = None
[docs] def get(
self,
object_id=None,
):
request = _All(
)
def process_result(result):
data = result
for p in self.__class__.path:
data = data[p]
sorting_data = result
for p in self.__class__.path_to_sorting:
sorting_data = sorting_data[p]
all_objects = [
self.__class__(
data={
**data[str(key)],
**(
{
'id': int(key),
}
if 'id' not in data[str(key)]
and key not in data[str(key)].values()
else {}
)
},
)
for key in sorting_data
]
key = (
'id'
if 'id' in self.data else
self.sorting_key
)
relevant_objects = [
pm
for pm in all_objects
if pm.__getattribute__(key) == self.__getattribute__(key)
]
if len(relevant_objects) == 1:
return relevant_objects[0]
elif len(relevant_objects) > 1:
raise ValueError('There should only be one object')
else:
raise ValueError('Object could not be found')
request.process_result = process_result
return request
[docs] @staticmethod
def get_all(
model: typing.Type,
):
request = _All(
)
def process_result(result):
data = result
for p in model.path:
data = data[p]
all_objects = [
model(
data={
**(
{
'id': int(key),
'cluster_id': result['data']['cluster']['id'],
}
if 'id' not in data[key]
and key not in data[key].values()
else {}
),
**data[key],
},
)
for key in data
]
def get_by_sorting_key(obj, key):
# automatically handle lambdas and str as self.sorting_key
if callable(key):
return key(obj)
else:
return obj.__getattribute__(key)
if model.path_to_sorting is not None:
sorting = result
for p in model.path_to_sorting:
sorting = sorting[p]
objects_sorted = []
for ident in sorting:
# Pick the item from wherever it is located the original list and append it to the output
l = [
get_by_sorting_key(obj, obj.sorting_key)
for obj in all_objects
]
try:
index = l.index(ident)
except ValueError as e:
logging.warning('List contains an id that could not be found')
objects_sorted = all_objects
break
objects_sorted.append(all_objects.pop(index))
else:
objects_sorted = sorted(
all_objects,
key=lambda x: get_by_sorting_key(x, x.sorting_key),
)
all_objects = objects_sorted
return all_objects
request.process_result = process_result
return request
[docs]class FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle(
abc.ABC,
):
@property
def name(self) -> str:
return self.data['name']
[docs]class AlarmAndEventAndMessageAndMessageChannelAndNewsAndStatus(
abc.ABC,
):
@property
def hidden(self) -> bool:
return self.data['hidden']
[docs]class AlarmAndEventAndMessageAndMessageChannelAndNews(
abc.ABC,
):
@property
def author_id(self) -> int:
return self.data['author_id']
@property
def deleted(self) -> bool:
return self.data['deleted']
@property
def ts_create(self) -> datetime.datetime:
if self.data['ts_create'] is not None:
return datetime.datetime.fromtimestamp(self.data['ts_create'])
@property
def ts_update(self) -> datetime.datetime:
if self.data['ts_update'] is not None:
return datetime.datetime.fromtimestamp(self.data['ts_update'])
[docs]class AlarmAndEventAndGroupAndMessageChannelAndNews(
abc.ABC,
):
@property
def foreign_id(self) -> str:
return self.data['foreign_id']
[docs]class AlarmAndEventAndMessageAndNews(
abc.ABC,
):
@property
def attachment_count(self) -> int:
return self.data['attachment_count']
@property
def message_channel_id(self) -> int:
return self.data['message_channel_id']
@property
def text(self) -> str:
return self.data['text']
[docs]class AlarmAndEventAndMessageChannelAndNews(
abc.ABC,
):
@property
def editable(self) -> bool:
return self.data['editable']
@property
def private_mode(self) -> bool:
return self.data['private_mode']
@property
def title(self) -> str:
return self.data['title']
@property
def ts_publish(self) -> datetime.datetime:
if self.data['ts_publish'] is not None:
return datetime.datetime.fromtimestamp(self.data['ts_publish'])
[docs]class AlarmAndEventAndNewsAndVehicle(
abc.ABC,
):
@property
def lat(self) -> float:
return self.data['lat']
@property
def lng(self) -> float:
return self.data['lng']
[docs]class AlarmAndEventAndNews(
abc.ABC,
):
@property
def address(self) -> str:
return self.data['address']
@property
def answerable(self) -> bool:
return self.data['answerable']
@property
def cluster(self) -> list:
return self.data['cluster']
@property
def count_read(self) -> int:
return self.data['count_read']
@property
def count_recipients(self) -> int:
return self.data['count_recipients']
@property
def group(self) -> list:
return self.data['group']
@property
def message_channel(self) -> bool:
return self.data['message_channel']
@property
def new(self) -> bool:
return self.data['new']
@property
def notification_type(self) -> int:
return self.data['notification_type']
@property
def ucr_addressed(self) -> list:
return self.data['ucr_addressed']
@property
def ucr_read(self) -> list:
return self.data['ucr_read']
@property
def ucr_self_addressed(self) -> bool:
return self.data['ucr_self_addressed']
@property
def user_cluster_relation(self) -> list:
return self.data['user_cluster_relation']
[docs]class OperationRoleAndQualificationAndVehicle(
abc.ABC,
):
@property
def shortname(self) -> str:
return self.data['shortname']
[docs]class AlarmAndNews(
abc.ABC,
):
@property
def date(self) -> datetime.datetime:
if self.data['date'] is not None:
return datetime.datetime.fromtimestamp(self.data['date'])
[docs]class AlarmAndReportType(
abc.ABC,
):
@property
def vehicle(self) -> int:
return self.data['vehicle']
[docs]class AlarmAndEvent(
abc.ABC,
):
@property
def custom_answers(self) -> bool:
return self.data['custom_answers']
@property
def ts_response(self) -> datetime.datetime:
if self.data['ts_response'] is not None:
return datetime.datetime.fromtimestamp(self.data['ts_response'])
[docs]class EventAndMessage(
abc.ABC,
):
@property
def attachments(self) -> list:
return self.data['attachments']
[docs]class FMSStatusAndVehicle(
abc.ABC,
):
@property
def number(self) -> str:
return self.data['number']
[docs]class FMSStatusAndStatus(
abc.ABC,
):
@property
def color_hex(self) -> str:
return self.data['color_hex']
@property
def color_id(self) -> int:
return self.data['color_id']
[docs]class MessageAndMessageChannel(
abc.ABC,
):
@property
def messages(self) -> list:
return self.data['messages']
[docs]class MessageChannelAndReportType(
abc.ABC,
):
@property
def description(self) -> str:
return self.data['description']
[docs]class MessageChannelAndNews(
abc.ABC,
):
@property
def ts_archive(self) -> datetime.datetime:
if self.data['ts_archive'] is not None:
return datetime.datetime.fromtimestamp(self.data['ts_archive'])
[docs]class Alarm(
PullModel,
AttachmentModel,
AlarmAndEvent,
AlarmAndNews,
AlarmAndReportType,
abc.ABC,
):
path = [
'data',
'alarm',
'items',
]
path_to_sorting = [
'data',
'alarm',
'sorting',
]
def __init__(
self,
id: int = None,
author_id: int = None,
cluster_id: int = None,
alarmcode_id: int = None,
message_channel_id: int = None,
foreign_id: str = None,
title: str = None,
text: str = None,
report: str = None,
address: str = None,
lat: float = None,
lng: float = None,
priority: bool = None,
date: [int, float, datetime.datetime] = None,
new: bool = None,
editable: bool = None,
answerable: bool = None,
notification_type: int = None,
vehicle: list = None,
group: list = None,
cluster: list = None,
user_cluster_relation: list = None,
hidden: bool = None,
deleted: bool = None,
message_channel: bool = None,
custom_answers: bool = None,
attachment_count: int = None,
closed: bool = None,
duration: str = None,
ts_response: [int, float, datetime.datetime] = None,
response_time: [int, float, datetime.timedelta] = None,
ucr_addressed: list = None,
ucr_answered: dict = None,
ucr_answeredcount: dict = None,
ucr_read: list = None,
ucr_self_addressed: bool = None,
count_recipients: int = None,
count_read: int = None,
private_mode: bool = None,
ts_publish: [int, float, datetime.datetime] = None,
ts_create: [int, float, datetime.datetime] = None,
ts_update: [int, float, datetime.datetime] = None,
ts_close: [int, float, datetime.datetime] = None,
notification_filter_vehicle: bool = None,
notification_filter_status: bool = None,
notification_filter_shift_plan: int = None,
notification_filter_access: bool = None,
notification_filter_status_access: bool = None,
ucr_self_status_id: int = None,
ucr_self_note: str = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'author_id': author_id,
'cluster_id': cluster_id,
'alarmcode_id': alarmcode_id,
'message_channel_id': message_channel_id,
'foreign_id': foreign_id,
'title': title,
'text': text,
'report': report,
'address': address,
'lat': lat,
'lng': lng,
'priority': priority,
'date': date.timestamp() if type(date) is datetime.datetime else date,
'new': new,
'editable': editable,
'answerable': answerable,
'notification_type': notification_type,
'vehicle': vehicle,
'group': group,
'cluster': cluster,
'user_cluster_relation': user_cluster_relation,
'hidden': hidden,
'deleted': deleted,
'message_channel': message_channel,
'custom_answers': custom_answers,
'attachment_count': attachment_count,
'closed': closed,
'duration': duration,
'ts_response': ts_response.timestamp() if type(ts_response) is datetime.datetime else ts_response,
'response_time': response_time.timestamp() if type(response_time) is datetime.timedelta else response_time,
'ucr_addressed': ucr_addressed,
'ucr_answered': ucr_answered,
'ucr_answeredcount': ucr_answeredcount,
'ucr_read': ucr_read,
'ucr_self_addressed': ucr_self_addressed,
'count_recipients': count_recipients,
'count_read': count_read,
'private_mode': private_mode,
'ts_publish': ts_publish.timestamp() if type(ts_publish) is datetime.datetime else ts_publish,
'ts_create': ts_create.timestamp() if type(ts_create) is datetime.datetime else ts_create,
'ts_update': ts_update.timestamp() if type(ts_update) is datetime.datetime else ts_update,
'ts_close': ts_close.timestamp() if type(ts_close) is datetime.datetime else ts_close,
'notification_filter_vehicle': notification_filter_vehicle,
'notification_filter_status': notification_filter_status,
'notification_filter_shift_plan': notification_filter_shift_plan,
'notification_filter_access': notification_filter_access,
'notification_filter_status_access': notification_filter_status_access,
'ucr_self_status_id': ucr_self_status_id,
'ucr_self_note': ucr_self_note,
},
**(data or {}),
}
)
@property
def alarmcode_id(self) -> int:
return self.data['alarmcode_id']
@property
def closed(self) -> bool:
return self.data['closed']
@property
def duration(self) -> str:
return self.data['duration']
@property
def notification_filter_access(self) -> bool:
return self.data['notification_filter_access']
@property
def notification_filter_shift_plan(self) -> int:
return self.data['notification_filter_shift_plan']
@property
def notification_filter_status(self) -> bool:
return self.data['notification_filter_status']
@property
def notification_filter_status_access(self) -> bool:
return self.data['notification_filter_status_access']
@property
def notification_filter_vehicle(self) -> bool:
return self.data['notification_filter_vehicle']
@property
def priority(self) -> bool:
return self.data['priority']
@property
def report(self) -> str:
return self.data['report']
@property
def response_time(self) -> datetime.timedelta:
if self.data['response_time'] is not None:
return datetime.timedelta(seconds=self.data['response_time'])
@property
def ts_close(self) -> datetime.datetime:
if self.data['ts_close'] is not None:
return datetime.datetime.fromtimestamp(self.data['ts_close'])
@property
def ucr_answered(self) -> dict:
return self.data['ucr_answered']
@property
def ucr_answeredcount(self) -> dict:
return self.data['ucr_answeredcount']
@property
def ucr_self_note(self) -> str:
return self.data['ucr_self_note']
@property
def ucr_self_status_id(self) -> int:
return self.data['ucr_self_status_id']
[docs]class Consumer(
PullModel,
abc.ABC,
):
path = [
'data',
'cluster',
'consumer',
]
path_to_sorting = [
'data',
'cluster',
'consumersorting',
]
def __init__(
self,
id: int = None,
cluster_id: int = None,
firstname: str = None,
lastname: str = None,
stdformat_name: str = None,
groups: list = None,
qualifications: list = None,
phonenumbers: list = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'firstname': firstname,
'lastname': lastname,
'stdformat_name': stdformat_name,
'groups': groups,
'qualifications': qualifications,
'phonenumbers': phonenumbers,
},
**(data or {}),
}
)
@property
def firstname(self) -> str:
return self.data['firstname']
@property
def groups(self) -> list:
return self.data['groups']
@property
def lastname(self) -> str:
return self.data['lastname']
@property
def phonenumbers(self) -> list:
return self.data['phonenumbers']
@property
def qualifications(self) -> list:
return self.data['qualifications']
@property
def stdformat_name(self) -> str:
return self.data['stdformat_name']
[docs]class Event(
PullModel,
AttachmentModel,
APIGetModel,
APIConfirmModel,
APIReadModel,
AlarmAndEvent,
EventAndMessage,
abc.ABC,
):
path = [
'data',
'events',
'items',
]
path_to_sorting = [
'data',
'events',
'sorting',
]
def __init__(
self,
id: int = None,
foreign_id: str = None,
author_id: int = None,
title: str = None,
text: str = None,
address: str = None,
cluster: list = None,
group: list = None,
user_cluster_relation: list = None,
private_mode: bool = None,
notification_type: int = None,
send_reminder: bool = None,
reminder: dict = None,
new: bool = None,
editable: bool = None,
answerable: bool = None,
hidden: bool = None,
deleted: bool = None,
show_result_count: int = None,
show_result_names: int = None,
custom_answers: bool = None,
response_type: int = None,
response_until: bool = None,
ts_response: [int, float, datetime.datetime] = None,
access_count: bool = None,
access_names: bool = None,
participation: int = None,
note: str = None,
ucr_self_addressed: bool = None,
count_recipients: int = None,
count_read: int = None,
ts_create: [int, float, datetime.datetime] = None,
ts_update: [int, float, datetime.datetime] = None,
cluster_id: int = None,
message_channel_id: int = None,
lat: float = None,
lng: float = None,
start: [int, float, datetime.datetime] = None,
end: [int, float, datetime.datetime] = None,
fullday: bool = None,
days: int = None,
ucr_addressed: list = None,
ucr_read: list = None,
message_channel: bool = None,
attachment_count: int = None,
participationlist: dict = None,
participationcount: dict = None,
participationnotes: list = None,
ts_publish: [int, float, datetime.datetime] = None,
attachments: list = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'foreign_id': foreign_id,
'author_id': author_id,
'title': title,
'text': text,
'address': address,
'cluster': cluster,
'group': group,
'user_cluster_relation': user_cluster_relation,
'private_mode': private_mode,
'notification_type': notification_type,
'send_reminder': send_reminder,
'reminder': reminder,
'new': new,
'editable': editable,
'answerable': answerable,
'hidden': hidden,
'deleted': deleted,
'show_result_count': show_result_count,
'show_result_names': show_result_names,
'custom_answers': custom_answers,
'response_type': response_type,
'response_until': response_until,
'ts_response': ts_response.timestamp() if type(ts_response) is datetime.datetime else ts_response,
'access_count': access_count,
'access_names': access_names,
'participation': participation,
'note': note,
'ucr_self_addressed': ucr_self_addressed,
'count_recipients': count_recipients,
'count_read': count_read,
'ts_create': ts_create.timestamp() if type(ts_create) is datetime.datetime else ts_create,
'ts_update': ts_update.timestamp() if type(ts_update) is datetime.datetime else ts_update,
'cluster_id': cluster_id,
'message_channel_id': message_channel_id,
'lat': lat,
'lng': lng,
'start': start.timestamp() if type(start) is datetime.datetime else start,
'end': end.timestamp() if type(end) is datetime.datetime else end,
'fullday': fullday,
'days': days,
'ucr_addressed': ucr_addressed,
'ucr_read': ucr_read,
'message_channel': message_channel,
'attachment_count': attachment_count,
'participationlist': participationlist,
'participationcount': participationcount,
'participationnotes': participationnotes,
'ts_publish': ts_publish.timestamp() if type(ts_publish) is datetime.datetime else ts_publish,
'attachments': attachments,
},
**(data or {}),
}
)
@property
def access_count(self) -> bool:
return self.data['access_count']
@property
def access_names(self) -> bool:
return self.data['access_names']
@property
def days(self) -> int:
return self.data['days']
@property
def end(self) -> datetime.datetime:
if self.data['end'] is not None:
return datetime.datetime.fromtimestamp(self.data['end'])
@property
def fullday(self) -> bool:
return self.data['fullday']
@property
def note(self) -> str:
return self.data['note']
@property
def participation(self) -> int:
return self.data['participation']
@property
def participationcount(self) -> dict:
return self.data['participationcount']
@property
def participationlist(self) -> dict:
return self.data['participationlist']
@property
def participationnotes(self) -> list:
return self.data['participationnotes']
@property
def reminder(self) -> dict:
return self.data['reminder']
@property
def response_type(self) -> int:
return self.data['response_type']
@property
def response_until(self) -> bool:
return self.data['response_until']
@property
def send_reminder(self) -> bool:
return self.data['send_reminder']
@property
def show_result_count(self) -> int:
return self.data['show_result_count']
@property
def show_result_names(self) -> int:
return self.data['show_result_names']
@property
def start(self) -> datetime.datetime:
if self.data['start'] is not None:
return datetime.datetime.fromtimestamp(self.data['start'])
[docs]class File(
Model,
abc.ABC,
):
# sorting_key = 'file_reference_id'
def __init__(
self,
id: int = None,
file_reference_id: int = None,
foreign_type: str = None,
foreign_id: id = None,
type: int = None,
title: str = None,
description: str = None,
encrypted: bool = None,
encryption_iv: str = None,
name: str = None,
mime_type: str = None,
deleted: bool = None,
size: int = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'file_reference_id': file_reference_id,
'foreign_type': foreign_type,
'foreign_id': foreign_id,
'type': type,
'title': title,
'description': description,
'encrypted': encrypted,
'encryption_iv': encryption_iv,
'name': name,
'mime_type': mime_type,
'deleted': deleted,
'size': size,
},
**(data or {}),
}
)
@property
def id(self):
return self.data['id']
@property
def file_reference_id(self):
return self.data['file_reference_id']
@property
def foreign_type(self):
return self.data['foreign_type']
@property
def foreign_id(self):
return self.data['foreign_id']
@property
def type(self):
return self.data['type']
@property
def title(self):
return self.data['title']
@property
def description(self):
return self.data['description']
@property
def encrypted(self):
return self.data['encrypted']
@property
def encryption_iv(self):
return self.data['encryption_iv']
@property
def name(self):
return self.data['name']
@property
def mime_type(self):
return self.data['mime_type']
@property
def deleted(self):
return self.data['deleted']
@property
def size(self):
return self.data['size']
[docs]class FMSStatus(
PullModel,
FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle,
FMSStatusAndStatus,
FMSStatusAndVehicle,
abc.ABC,
):
path = [
'data',
'cluster',
'fms_status',
'items',
]
path_to_sorting = [
'data',
'cluster',
'fms_status',
'sorting',
]
sorting_key = 'number'
def __init__(
self,
name: str = None,
number: str = None,
color_id: int = None,
color_hex: str = None,
data: dict = None,
):
super().__init__(
data={
**{
'name': name,
'number': number,
'color_id': color_id,
'color_hex': color_hex,
},
**(data or {}),
}
)
[docs]class Group(
PullModel,
AlarmAndEventAndGroupAndMessageChannelAndNews,
FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle,
abc.ABC,
):
path = [
'data',
'cluster',
'group',
]
path_to_sorting = [
'data',
'cluster',
'groupsorting',
]
def __init__(
self,
id: int = None,
cluster_id: int = None,
name: str = None,
ric: str = None,
foreign_id: str = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'name': name,
'ric': ric,
'foreign_id': foreign_id,
},
**(data or {}),
}
)
@property
def ric(self) -> str:
return self.data['ric']
[docs]class Message(
PullModel,
AttachmentModel,
EventAndMessage,
MessageAndMessageChannel,
abc.ABC,
):
def __init__(
self,
id: int = None,
message_channel_id: int = None,
author_id: int = None,
parent_id: int = None,
attachment_count: int = None,
attachments: list = None,
text: str = None,
ts_create: [int, float, datetime.datetime] = None,
ts_update: [int, float, datetime.datetime] = None,
messages: list = None,
thread: bool = None,
thread_first_message_id: int = None,
thread_last_message_id: int = None,
thread_message_count: int = None,
hidden: bool = None,
deleted: bool = None,
deleted_by_user_cluster_relation_id: int = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'message_channel_id': message_channel_id,
'author_id': author_id,
'parent_id': parent_id,
'attachment_count': attachment_count,
'attachments': attachments,
'text': text,
'ts_create': ts_create.timestamp() if type(ts_create) is datetime.datetime else ts_create,
'ts_update': ts_update.timestamp() if type(ts_update) is datetime.datetime else ts_update,
'messages': messages,
'thread': thread,
'thread_first_message_id': thread_first_message_id,
'thread_last_message_id': thread_last_message_id,
'thread_message_count': thread_message_count,
'hidden': hidden,
'deleted': deleted,
'deleted_by_user_cluster_relation_id': deleted_by_user_cluster_relation_id,
},
**(data or {}),
}
)
@property
def deleted_by_user_cluster_relation_id(self) -> int:
return self.data['deleted_by_user_cluster_relation_id']
@property
def parent_id(self) -> int:
return self.data['parent_id']
@property
def thread(self) -> bool:
return self.data['thread']
@property
def thread_first_message_id(self) -> int:
return self.data['thread_first_message_id']
@property
def thread_last_message_id(self) -> int:
return self.data['thread_last_message_id']
@property
def thread_message_count(self) -> int:
return self.data['thread_message_count']
[docs]class MessageChannel(
Model,
MessageAndMessageChannel,
MessageChannelAndNews,
MessageChannelAndReportType,
abc.ABC,
):
def __init__(
self,
id: int = None,
foreign_type: str = None,
foreign_id: int = None,
cluster_id: int = None,
author_id: int = None,
first_message_id: int = None,
last_message_id: int = None,
title: str = None,
description: str = None,
access_all: bool = None,
config: dict = None,
ucr_access: dict = None,
self_access: dict = None,
private_mode: bool = None,
editable: bool = None,
ts_publish: [int, float, datetime.datetime] = None,
ts_archive: [int, float, datetime.datetime] = None,
ts_delete: [int, float, datetime.datetime] = None,
ts_create: [int, float, datetime.datetime] = None,
ts_update: [int, float, datetime.datetime] = None,
ts_last_message: [int, float, datetime.datetime] = None,
messages: list = None,
message_count: int = None,
hidden: bool = None,
deleted: bool = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'foreign_type': foreign_type,
'foreign_id': foreign_id,
'cluster_id': cluster_id,
'author_id': author_id,
'first_message_id': first_message_id,
'last_message_id': last_message_id,
'title': title,
'description': description,
'access_all': access_all,
'config': config,
'ucr_access': ucr_access,
'self_access': self_access,
'private_mode': private_mode,
'editable': editable,
'ts_publish': ts_publish.timestamp() if type(ts_publish) is datetime.datetime else ts_publish,
'ts_archive': ts_archive.timestamp() if type(ts_archive) is datetime.datetime else ts_archive,
'ts_delete': ts_delete.timestamp() if type(ts_delete) is datetime.datetime else ts_delete,
'ts_create': ts_create.timestamp() if type(ts_create) is datetime.datetime else ts_create,
'ts_update': ts_update.timestamp() if type(ts_update) is datetime.datetime else ts_update,
'ts_last_message': ts_last_message.timestamp() if type(ts_last_message) is datetime.datetime else ts_last_message,
'messages': messages,
'message_count': message_count,
'hidden': hidden,
'deleted': deleted,
},
**(data or {}),
}
)
@property
def access_all(self) -> bool:
return self.data['access_all']
@property
def config(self) -> dict:
return self.data['config']
@property
def first_message_id(self) -> int:
return self.data['first_message_id']
@property
def foreign_type(self) -> str:
return self.data['foreign_type']
@property
def last_message_id(self) -> int:
return self.data['last_message_id']
@property
def message_count(self) -> int:
return self.data['message_count']
@property
def self_access(self) -> dict:
return self.data['self_access']
@property
def ts_delete(self) -> datetime.datetime:
if self.data['ts_delete'] is not None:
return datetime.datetime.fromtimestamp(self.data['ts_delete'])
@property
def ts_last_message(self) -> datetime.datetime:
if self.data['ts_last_message'] is not None:
return datetime.datetime.fromtimestamp(self.data['ts_last_message'])
@property
def ucr_access(self) -> dict:
return self.data['ucr_access']
[docs]class News(
PullModel,
AttachmentModel,
AlarmAndEventAndMessageAndNews,
AlarmAndEventAndNews,
AlarmAndEventAndNewsAndVehicle,
AlarmAndNews,
MessageChannelAndNews,
abc.ABC,
):
path = [
'data',
'alarm',
'items',
]
path_to_sorting = [
'data',
'alarm',
'sorting',
]
def __init__(
self,
id: int = None,
title: str = None,
text: str = None,
author_id: int = None,
cluster_id: int = None,
message_channel_id: int = None,
foreign_id: str = None,
address: str = None,
lat: float = None,
lng: float = None,
archive: bool = None,
date: [int, float, datetime.datetime] = None,
ts_archive: [int, float, datetime.datetime] = None,
new: bool = None,
editable: bool = None,
answerable: bool = None,
notification_type: int = None,
group: list = None,
cluster: list = None,
user_cluster_relation: list = None,
hidden: bool = None,
deleted: bool = None,
message_channel: bool = None,
attachment_count: int = None,
count_recipients: int = None,
count_read: int = None,
survey: bool = None,
ucr_addressed: list = None,
ucr_read: list = None,
ucr_self_addressed: bool = None,
private_mode: bool = None,
cross_unit_meta: dict = None,
ts_publish: [int, float, datetime.datetime] = None,
ts_create: [int, float, datetime.datetime] = None,
ts_update: [int, float, datetime.datetime] = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'title': title,
'text': text,
'author_id': author_id,
'cluster_id': cluster_id,
'message_channel_id': message_channel_id,
'foreign_id': foreign_id,
'address': address,
'lat': lat,
'lng': lng,
'archive': archive,
'date': date.timestamp() if type(date) is datetime.datetime else date,
'ts_archive': ts_archive.timestamp() if type(ts_archive) is datetime.datetime else ts_archive,
'new': new,
'editable': editable,
'answerable': answerable,
'notification_type': notification_type,
'group': group,
'cluster': cluster,
'user_cluster_relation': user_cluster_relation,
'hidden': hidden,
'deleted': deleted,
'message_channel': message_channel,
'attachment_count': attachment_count,
'count_recipients': count_recipients,
'count_read': count_read,
'survey': survey,
'ucr_addressed': ucr_addressed,
'ucr_read': ucr_read,
'ucr_self_addressed': ucr_self_addressed,
'private_mode': private_mode,
'cross_unit_meta': cross_unit_meta,
'ts_publish': ts_publish.timestamp() if type(ts_publish) is datetime.datetime else ts_publish,
'ts_create': ts_create.timestamp() if type(ts_create) is datetime.datetime else ts_create,
'ts_update': ts_update.timestamp() if type(ts_update) is datetime.datetime else ts_update,
},
**(data or {}),
}
)
@property
def archive(self) -> bool:
return self.data['archive']
@property
def cross_unit_meta(self) -> dict:
return self.data['cross_unit_meta']
@property
def survey(self) -> bool:
return self.data['survey']
[docs]class OperationRole(
PullModel,
FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle,
OperationRoleAndQualificationAndVehicle,
abc.ABC,
):
path = [
'data',
'cluster',
'operationrole',
]
path_to_sorting = [
'data',
'cluster',
'operationrolesorting',
]
def __init__(
self,
id: int = None,
cluster_id: int = None,
name: str = None,
shortname: str = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'name': name,
'shortname': shortname,
},
**(data or {}),
}
)
[docs]class Qualification(
PullModel,
FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle,
OperationRoleAndQualificationAndVehicle,
abc.ABC,
):
path = [
'data',
'cluster',
'qualification',
]
path_to_sorting = [
'data',
'cluster',
'qualificationsorting',
]
def __init__(
self,
id: int = None,
cluster_id: int = None,
name: str = None,
shortname: str = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'name': name,
'shortname': shortname,
},
**(data or {}),
}
)
[docs]class ReportType(
PullModel,
AlarmAndReportType,
FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle,
MessageChannelAndReportType,
abc.ABC,
):
path = [
'data',
'cluster',
'reporttypes',
'items',
]
path_to_sorting = [
'data',
'cluster',
'reporttypes',
'sorting',
]
def __init__(
self,
id: int = None,
cluster_id: int = None,
name: str = None,
description: str = None,
anonym: int = None,
location: int = None,
vehicle: int = None,
fields: list = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'name': name,
'description': description,
'anonym': anonym,
'location': location,
'vehicle': vehicle,
'fields': fields,
},
**(data or {}),
}
)
@property
def anonym(self) -> int:
return self.data['anonym']
@property
def fields(self) -> list:
return self.data['fields']
@property
def location(self) -> int:
return self.data['location']
[docs]class Status(
PullModel,
AlarmAndEventAndMessageAndMessageChannelAndNewsAndStatus,
FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle,
FMSStatusAndStatus,
abc.ABC,
):
path = [
'data',
'cluster',
'status',
]
path_to_sorting = [
'data',
'cluster',
'statussorting',
]
def __init__(
self,
id: int = None,
cluster_id: int = None,
name: str = None,
show_on_alarmmonitor: bool = None,
show_on_alarm: bool = None,
show_on_statusgeber: bool = None,
show_on_statusplaner: bool = None,
show_on_geofence: bool = None,
color_id: int = None,
color_hex: str = None,
time: int = None,
sorting: int = None,
hidden: bool = None,
phonenumber: str = None,
alias: str = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'name': name,
'show_on_alarmmonitor': show_on_alarmmonitor,
'show_on_alarm': show_on_alarm,
'show_on_statusgeber': show_on_statusgeber,
'show_on_statusplaner': show_on_statusplaner,
'show_on_geofence': show_on_geofence,
'color_id': color_id,
'color_hex': color_hex,
'time': time,
'sorting': sorting,
'hidden': hidden,
'phonenumber': phonenumber,
'alias': alias,
},
**(data or {}),
}
)
@property
def alias(self) -> str:
return self.data['alias']
@property
def phonenumber(self) -> str:
return self.data['phonenumber']
@property
def show_on_alarm(self) -> bool:
return self.data['show_on_alarm']
@property
def show_on_alarmmonitor(self) -> bool:
return self.data['show_on_alarmmonitor']
@property
def show_on_geofence(self) -> bool:
return self.data['show_on_geofence']
@property
def show_on_statusgeber(self) -> bool:
return self.data['show_on_statusgeber']
@property
def show_on_statusplaner(self) -> bool:
return self.data['show_on_statusplaner']
@property
def sorting(self) -> int:
return self.data['sorting']
@property
def time(self) -> int:
return self.data['time']
[docs]class StatusPlanCategory(
PullModel,
FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle,
abc.ABC,
):
path = [
'data',
'cluster',
'statusplancategory',
]
path_to_sorting = [
'data',
'cluster',
'statusplancategorysorting',
]
def __init__(
self,
id: int = None,
cluster_id: int = None,
name: str = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'name': name,
},
**(data or {}),
}
)
[docs]class Vehicle(
PullModel,
AlarmAndEventAndNewsAndVehicle,
FMSStatusAndGroupAndOperationRoleAndQualificationAndReportTypeAndStatusAndStatusPlanCategoryAndVehicle,
FMSStatusAndVehicle,
OperationRoleAndQualificationAndVehicle,
abc.ABC,
):
path = [
'data',
'cluster',
'vehicle',
]
path_to_sorting = [
'data',
'cluster',
'vehiclesorting',
]
sorting_key = 'name'
def __init__(
self,
id: int = None,
cluster_id: int = None,
fullname: str = None,
shortname: str = None,
name: str = None,
fmsstatus_id: int = None,
fmsstatus_note: str = None,
fmsstatus_ts: int = None,
crew: list = None,
lat: float = None,
lng: float = None,
opta: str = None,
issi: str = None,
number: str = None,
data: dict = None,
):
super().__init__(
data={
**{
'id': id,
'cluster_id': cluster_id,
'fullname': fullname,
'shortname': shortname,
'name': name,
'fmsstatus_id': fmsstatus_id,
'fmsstatus_note': fmsstatus_note,
'fmsstatus_ts': fmsstatus_ts,
'crew': crew,
'lat': lat,
'lng': lng,
'opta': opta,
'issi': issi,
'number': number,
},
**(data or {}),
}
)
@property
def crew(self) -> list:
return self.data['crew']
@property
def fmsstatus_id(self) -> int:
return self.data['fmsstatus_id']
@property
def fmsstatus_note(self) -> str:
return self.data['fmsstatus_note']
@property
def fmsstatus_ts(self) -> int:
return self.data['fmsstatus_ts']
@property
def fullname(self) -> str:
return self.data['fullname']
@property
def issi(self) -> str:
return self.data['issi']
@property
def opta(self) -> str:
return self.data['opta']