Source code for materials_commons.api.models

from .util import get_date


[docs]def from_list(cls, data): if data is None: return [] return [cls(d) for d in data]
[docs]def pretty_print(clas, indent=0): """ Prints to stdout a formatted version of the object. This will recursively print sub-objects as well as iterate over lists of objects maintaining proper indenting. Private attributes are ignored. """ print(' ' * indent + type(clas).__name__ + ':') indent += 4 for k, v in clas.__dict__.items(): if '__dict__' in dir(v): pretty_print(v, indent) elif isinstance(v, list): print(' ' * indent + k + ': ') for item in v: pretty_print(item, indent + 4) else: if k != '_data': print(' ' * indent + k + ': ' + str(v))
[docs]class Common(object): """ Base class for most models. Contains common attributes shared across most model objects. Attributes ---------- id : int The id of the object. uuid : str The uuid of the object. name : str Name of model object. description : str Description of the model instance, for example a description of a project. summary : str A short description suitable for display in a table. owner_id : int The id of the owner of the model instance. owner : mcapi.Owner The full owner model associated with the owner_id. created_at : str Formatted string datetime when the object was created. String format is "%Y-%m-%dT%H:%M:%S.%fZ". updated_at : str Formatted string datetime when the object was last updated. String format is "%Y-%m-%dT%H:%M:%S.%fZ". project_id : int The project_id is an optional field that exists only if the underlying model has a project_id field. The project_id is the id of the project the object is associated with. Methods ------- pretty_print() Prints to stdout a formatted version of the object. This will recursively print sub-objects as well as iterate over lists of objects maintaining proper indenting. Private attributes are ignored. """ def __init__(self, data): self._data = data.copy() self.id = data.get('id', None) self.uuid = data.get('uuid', None) self.name = data.get('name', None) self.description = data.get('description', None) self.summary = data.get('summary', None) self.owner_id = data.get('owner_id', None) self.created_at = get_date('created_at', data) self.updated_at = get_date('updated_at', data) project_id = data.get('project_id', None) if project_id: self.project_id = project_id owner = data.get('owner', None) if owner: self.owner = User(owner)
[docs] def pretty_print(self): pretty_print(self)
[docs]class Community(Common): """ Community represents a Materials Commons Community of Practice. A Community of Practice is place to gather similar published datasets together. In addition it contains links and files that are specific to a community. For example it may contain a link to a forum associated with the community, or a file with suggested naming conventions. Attributes: ----------- public : bool A flag that is true if this community is public and viewable by anyone. files : list of mcapi.File Files associated with the community. links: list of mcapi.Link Links associated with the community. datasets: list of mcapi.Dataset List of published datasets associated with the community. """ def __init__(self, data={}): super(Community, self).__init__(data) self.public = data.get('public', None) self.files = File.from_list_attr(data) self.links = Link.from_list_attr(data) self.datasets = Dataset.from_list_attr(data)
[docs] @staticmethod def from_list(data): return from_list(Community, data)
[docs] @staticmethod def from_list_attr(data, attr='communities'): return Community.from_list(data.get(attr, []))
[docs]class Activity(Common): """ An activity represents a step that operates on one or more Entities. For example an Entity maybe have been heat treated. In that case an Activity representing the heat treatment step could be associated with the Entity. An activity may also have files associated with it. These may represent files produced by that activity. For example the images produced from running an SEM. All the files from the SEM will be associated with the activity. A subset of these files may be represented with an Entity that the SEM operated on. Attributes: ---------- entities: list of mcapi.Entity The list of entities associated with this activity. files: list of mcapi.File The list of files associated with this activity. """ def __init__(self, data={}): super(Activity, self).__init__(data) self.entities = Entity.from_list_attr(data) self.files = File.from_list_attr(data)
[docs] @staticmethod def from_list(data): return from_list(Activity, data)
[docs] @staticmethod def from_list_attr(data, attr='activities'): return Activity.from_list(data.get(attr, []))
[docs]class Dataset(Common): """ A dataset represents a collection of files, activities and entities, along with other meta data such as authors, papers, etc... that is meant to be shared as a whole. A dataset can be published, in which case the dataset is available to the public, and its associated files can be downloaded. Attributes: ----------- license : str The license (if any) associated with the dataset. license_url : str The url of the license associated with the dataset. Currently licenses all come from Open Data Commons. doi : str The DOI associated with the dataset. authors : FIX THIS file_selection : dict The file_selection is a selection of files and directories to include/exclude in a dataset when it is published. The file_selection has the following fields (each field is a list): include_files, exclude_files, include_dirs, exclude_dirs. zipfile_size : int If a zipfile was built for this dataset then this is the size of the zipfile in bytes. zipfile_name : str If a zipfile was build for this dataset then this is the name of the zipfile. workflows : list of mcapi.Workflow The list of workflows associated with the dataset. experiments : list of mcapi.Experiment The list of experiments associated with the dataset. activities: list of mcapi.Activity The list of activities included with the dataset. entities: list of mcapi.Entity The list of entities included with the dataset. files : list of mcapi.File The list of files included with the dataset. globus_path : str The globus path for using globus to access the dataset files. globus_endpoint_id : str The globus endpoint the dataset files are stored on. workflows_count : int Count of workflows included with dataset. activities_count : int Count of activities included with dataset. entities_count : int Count of entities included with dataset. comments_count : int Count of comments associated with dataset. published_at : str The date the dataset was published on. tags : list of mcapi.Tag The tags associated with the dataset. """ def __init__(self, data={}): super(Dataset, self).__init__(data) self.license = data.get('license', None) self.license_link = self._get_license_link(data) self.doi = data.get('doi', None) self.authors = data.get('authors', None) self.file_selection = data.get('file_selection', None) self.zipfile_size = data.get('zipfile_size', None) self.zipfile_name = data.get('zipfile_name', None) self.workflows = Workflow.from_list_attr(data) self.experiments = Experiment.from_list_attr(data) self.activities = Activity.from_list_attr(data) self.entities = Entity.from_list_attr(data) self.files = File.from_list_attr(data) self.globus_path = data.get('globus_path', None) self.globus_endpoint_id = data.get('globus_endpoint_id', None) self.experiments_count = data.get('experiments_count', None) self.files_count = data.get('files_count', None) self.workflows_count = data.get('workflows_count', None) self.activities_count = data.get('activities_count', None) self.entities_count = data.get('entities_count', None) self.comments_count = data.get('comments_count', None) self.published_at = get_date('published_at', data) self.tags = Tag.from_list_attr(data) def _get_license_link(self, data): if not self.license: return None license_link = data.get('license_link', None) if license_link: return license_link if self.license == "Public Domain Dedication and License (PDDL)": return "https://opendatacommons.org/licenses/pddl/summary" elif self.license == "Attribution License (ODC-By)": return "https://opendatacommons.org/licenses/by/summary" elif self.license == "Open Database License (ODC-ODbL)": return "https://opendatacommons.org/licenses/odbl/summary" else: return "https://opendatacommons.org"
[docs] @staticmethod def from_list(data): return from_list(Dataset, data)
[docs] @staticmethod def from_list_attr(data, attr='datasets'): return Dataset.from_list(data.get(attr, []))
[docs]class Entity(Common): """ An entity represent a virtual or physical specimen, sample or object. An entity is what is being measured or transformed. An example of an entity would be a sheet of metal that is be tested. That sheet might be heated (Activity), cut (Activity) then viewed on a SEM (Activity). Attributes: ---------- activities: list of mcapi.Activity The list of activities associated with this entity. files: list of mcapi.File The list of files associated with this entity. """ def __init__(self, data={}): super(Entity, self).__init__(data) self.activities = Activity.from_list_attr(data) self.files = File.from_list_attr(data)
[docs] @staticmethod def from_list(data): return from_list(Entity, data)
[docs] @staticmethod def from_list_attr(data, attr='entities'): return Entity.from_list(data.get(attr, []))
[docs]class Experiment(Common): """ An experiment is a container for entities, activities, and files. Attributes: ----------- workflows : list of mcapi.Workflow The list of workflows used in the experiment. activities : list of mcapi.Activity The list of activities used in the experiment. entities : list of mcapi.Entity The list of entities used in the experiment. files : list of mcapi.File The list of files used in the experiment. """ def __init__(self, data={}): super(Experiment, self).__init__(data) self.workflows = Workflow.from_list_attr(data) self.activities = Activity.from_list_attr(data) self.entities = Entity.from_list_attr(data) self.files = File.from_list_attr(data)
[docs] @staticmethod def from_list(data): return from_list(Experiment, data)
[docs] @staticmethod def from_list_attr(data, attr='experiments'): return Experiment.from_list(data.get(attr, []))
[docs]class File(Common): """ A file is an uploaded file associated with a project in Materials Commons. Attributes: ----------- mime_type : str path : str directory_id : int size : int checksum : str experiments_count : int activities_count : int entities_count : int entity_states_count : int previous_versions_count : int directory : mcapi.File """ def __init__(self, data={}): super(File, self).__init__(data) self.mime_type = data.get('mime_type', None) self.path = data.get('path', None) self.directory_id = data.get('directory_id', None) self.size = data.get('size', None) self.checksum = data.get('checksum', None) self.experiments_count = data.get('experiments_count', None) self.activities_count = data.get('activities_count', None) self.entities_count = data.get('entities_count', None) self.entity_states_count = data.get('entity_states_count', None) self.previous_versions_count = data.get('previous_versions_count', None) directory = data.get('directory', None) if directory: self.directory = File(directory) self._make_path() else: self.directory = None def _make_path(self): if self.directory.path == "/": self.path = self.directory.path + self.name else: self.path = self.directory.path + "/" + self.name
[docs] @staticmethod def from_list(data): return from_list(File, data)
[docs] @staticmethod def from_list_attr(data, attr='files'): return File.from_list(data.get(attr, []))
[docs]class GlobusUpload(Common): def __init__(self, data={}): super(GlobusUpload, self).__init__(data) self.globus_endpoint_id = data.get('globus_endpoint_id', None) self.globus_url = data.get('globus_url', None) self.globus_path = data.get('globus_path', None) self.status = data.get('status', None)
[docs] @staticmethod def from_list(data): return from_list(GlobusUpload, data)
[docs] @staticmethod def from_list_attr(data, attr="globus_uploads"): return GlobusUpload.from_list(data.get(attr, []))
[docs]class GlobusDownload(Common): def __init__(self, data={}): super(GlobusDownload, self).__init__(data) self.globus_endpoint_id = data.get('globus_endpoint_id', None) self.globus_url = data.get('globus_url', None) self.globus_path = data.get('globus_path', None) self.status = data.get('status', None)
[docs] @staticmethod def from_list(data): return from_list(GlobusDownload, data)
[docs] @staticmethod def from_list_attr(data, attr="globus_uploads"): return GlobusDownload.from_list(data.get(attr, []))
[docs]class GlobusTransfer(object): """ A GlobusTransfer represents a started globus transfer, whether its an upload or a download. Attributes: ----------- id : int uuid : str globus_endpoint_id : str globus_url : str globus_path : str state : str last_globus_transfer_id_completed : str latest_globus_transfer_completed_date : str project_id : int owner_id : int transfer_request_id : in created_at : str updated_at : str """ def __init__(self, data={}): self._data = data.copy() self.id = data.get('id', None) self.uuid = data.get('uuid', None) self.globus_endpoint_id = data.get('globus_endpoint_id', None) self.globus_url = data.get('globus_url', None) self.globus_path = data.get('globus_path', None) self.state = data.get('state', None) self.last_globus_transfer_id_completed = data.get('last_globus_transfer_id_completed', None) self.latest_globus_transfer_completed_date = data.get('latest_globus_transfer_completed_date', None) self.project_id = data.get('project_id', None) self.owner_id = data.get('owner_id', None) self.transfer_request_id = data.get('transfer_request_id', None) self.created_at = get_date('created_at', data) self.updated_at = get_date('updated_at', data)
[docs] def pretty_print(self): pretty_print(self)
[docs] @staticmethod def from_list(data): return from_list(GlobusTransfer, data)
[docs] @staticmethod def from_list_attr(data, attr='globus_transfers'): return GlobusTransfer.from_list(data.get(attr, []))
[docs]class Project(Common): """ A project is the top level object that stores files and meta data about your research project. Attributes: ----------- activities : list of mcapi.Activity workflows : list of mcapi.Workflow experiments : list of mcapi.Experiment activities : list of mcapi.Activity entities : list of mcapi.Entity members : list of mcapi.User admins : list of mcapi.User root_dir : mcapi.File """ def __init__(self, data={}): super(Project, self).__init__(data) self.is_active = data.get('is_active', None) self.activities = Activity.from_list_attr(data) self.workflows = Workflow.from_list_attr(data) self.experiments = Experiment.from_list_attr(data) self.activities = Activity.from_list_attr(data) self.entities = Entity.from_list_attr(data) self.members = User.from_list_attr(data, 'members') self.admins = User.from_list_attr(data, 'admins') self.root_dir = None root_dir = data.get('rootDir', None) if root_dir: self.root_dir = File(root_dir)
[docs] @staticmethod def from_list(data): return from_list(Project, data)
[docs] @staticmethod def from_list_attr(data, attr='projects'): return Project.from_list(data.get(attr, []))
[docs]class Server(object): """ A Server contains information about the Materials Commons server hosting the API. Attributes: ----------- globus_endpoint_id : str institution : str version : str last_updated_at : str first_deployed_at : str contact : str description : str name : str uuid : str """ def __init__(self, data={}): self._data = data.copy() self.globus_endpoint_id = data.get('globus_endpoint_id', None) self.institution = data.get('institution', None) self.version = data.get('version', None) self.last_updated_at = data.get('last_updated_at', data) self.first_deployed_at = data.get('first_deployed_at', data) self.contact = data.get('contact', None) self.description = data.get('description', None) self.name = data.get('name', None) self.uuid = data.get('uuid', None)
[docs] def pretty_print(self): pretty_print(self)
[docs]class Tag(object): """ A tag is an attribute that can be added to different objects in the system. Currently only datasets support tags. Attributes: ----------- id : int name : str slug : str created_at : str updated_at : str """ def __init__(self, data={}): self._data = data.copy() self.id = data.get('id', None) self.name = data.get('name', None) self.slug = data.get('slug', None) self.created_at = get_date('created_at', data) self.updated_at = get_date('updated_at', data)
[docs] def pretty_print(self): pretty_print(self)
[docs] @staticmethod def from_list(data): return from_list(Tag, data)
[docs] @staticmethod def from_list_attr(data, attr='tags'): return Tag.from_list(data.get(attr, []))
[docs]class User(object): """ A User represents a user account on Materials Commons. Attributes: ----------- id : int uuid : str name : str email : str description : str affiliation : str created_at : str updated_at : str """ def __init__(self, data={}): self._data = data.copy() self.id = data.get('id', None) self.uuid = data.get('uuid', None) self.name = data.get('name', None) self.email = data.get('email', None) self.description = data.get('description', None) self.affiliation = data.get('affiliation', None) self.created_at = get_date('created_at', data) self.updated_at = get_date('updated_at', data)
[docs] def pretty_print(self): pretty_print(self)
[docs] @staticmethod def from_list(data): return from_list(User, data)
[docs] @staticmethod def from_list_attr(data, attr='users'): return User.from_list(data.get(attr, []))
[docs]class Searchable(object): """ A searchable represents the results of a search. Attributes: ----------- title : str url : str type : str "datasets" || "communities" item : mcapi.Dataset or mcapi.Community depending on what type is set to """ def __init__(self, data={}): self._data = data.copy() self.title = data.get('title') self.url = data.get('url') self.type = data.get('type') self._fill_item()
[docs] def pretty_print(self): pretty_print(self)
def _fill_item(self): if self.type == "datasets": self.item = Dataset(self._data["searchable"]) elif self.type == "communities": self.item = Community(self._data["searchable"])
[docs] @staticmethod def from_list(data): return from_list(Searchable, data)
[docs]class Workflow(Common): """ A workflow is a graphical and textual representation a user created for an experimental workflow. """ def __init__(self, data={}): super(Workflow, self).__init__(data)
[docs] @staticmethod def from_list(data): return from_list(Workflow, data)
[docs] @staticmethod def from_list_attr(data, attr='workflows'): return Workflow.from_list(data.get(attr, []))