jebus/enviPy-bayer
1
0
Fork 0
forked from enviPath/enviPy
Code Issues Pull Requests Actions Releases Activity

[Feature] More on Legacy API (#142)

Browse Source 
Co-authored-by: Tim Lorsbach <tim@lorsba.ch>
Reviewed-on: enviPath/enviPy#142
This commit is contained in:
jebus
2025-10-03 00:07:30 +13:00
parent 7ad4112343
commit 3f2b046bd6
3 changed files with 536 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

617
epdb/legacy_api.py
View File

@ -2,11 +2,24 @@ from typing import List, Dict, Optional, Any
from django.contrib.auth import get_user_model
from django.http import HttpResponse
from django.shortcuts import redirect
from ninja import Router, Schema, Field, Form
from utilities.chem import FormatConverter
from .logic import PackageManager
from .models import Compound, CompoundStructure, Package, User, UserPackagePermission, Rule, Reaction, Scenario, Pathway
from .logic import PackageManager, UserManager, SettingManager
from .models import (
Compound,
CompoundStructure,
Package,
User,
UserPackagePermission,
Rule,
Reaction,
Scenario,
Pathway,
Node,
Edge, SimpleAmbitRule
)
def _anonymous_or_real(request):
@ -22,16 +35,9 @@ router = Router()
class Error(Schema):
message: str
class SimpleObject(Schema):
id: str = Field(None, alias="url")
name: str = Field(None, alias="name")
reviewStatus: bool = Field(None, alias="package.reviewed")
################
# Login/Logout #
################
#################
# SimpleObjects #
#################
class SimpleUser(Schema):
id: str = Field(None, alias="url")
identifier: str = 'user'
@ -39,8 +45,82 @@ class SimpleUser(Schema):
email: str = Field(None, alias='email')
class SimpleGroup(Schema):
id: str = Field(None, alias="url")
identifier: str = 'group'
name: str = Field(None, alias='name')
class SimpleSetting(Schema):
id: str = Field(None, alias="url")
identifier: str = 'setting'
name: str = Field(None, alias='name')
class SimpleObject(Schema):
id: str = Field(None, alias="url")
name: str = Field(None, alias="name")
reviewStatus: str = Field(None, alias="review_status")
@staticmethod
def resolve_review_status(obj):
if isinstance(obj, Package):
return 'reviewed' if obj.reviewed else 'unreviewed'
elif hasattr(obj, 'package'):
return 'reviewed' if obj.package.reviewed else 'unreviewed'
elif isinstance(obj, CompoundStructure):
return 'reviewed' if obj.compound.package.reviewed else 'unreviewed'
elif isinstance(obj, Node) or isinstance(obj, Edge):
return 'reviewed' if obj.pathway.package.reviewed else 'unreviewed'
else:
raise ValueError('Object has no package')
class SimplePackage(SimpleObject):
identifier: str = 'package'
class SimpleCompound(SimpleObject):
identifier: str = 'compound'
class SimpleCompoundStructure(SimpleObject):
identifier: str = 'structure'
class SimpleRule(SimpleObject):
identifier: str = 'rule'
@staticmethod
def resolve_url(obj: Rule):
return obj.url.replace('-ambit-', '-').replace('-rdkit-', '-')
class SimpleReaction(SimpleObject):
identifier: str = 'reaction'
class SimpleScenario(SimpleObject):
identifier: str = 'scenario'
class SimplePathway(SimpleObject):
identifier: str = 'pathway'
class SimpleNode(SimpleObject):
identifier: str = 'node'
class SimpleEdge(SimpleObject):
identifier: str = 'edge'
################
# Login/Logout #
################
@router.post("/", response={200: SimpleUser, 403: Error})
def login(request, loginusername: Form[str], loginpassword: Form[str], hiddenMethod: Form[str]):
def login(request, loginusername: Form[str], loginpassword: Form[str]):
from django.contrib.auth import authenticate
from django.contrib.auth import login
email = User.objects.get(username=loginusername).email
@ -49,21 +129,68 @@ def login(request, loginusername: Form[str], loginpassword: Form[str], hiddenMet
login(request, user)
return user
else:
return 403, {'message': 'Invalid username or password'}
return 403, {'message': 'Invalid username and/or password'}
class SimpleGroup(Schema):
id: str
identifier: str = 'group'
name: str
########
# User #
########
class UserWrapper(Schema):
user: List[SimpleUser]
class UserSchema(Schema):
affiliation: Dict[str, str] = Field(None, alias="affiliation")
defaultGroup: 'SimpleGroup' = Field(None, alias="default_group")
defaultPackage: 'SimplePackage' = Field(None, alias="default_package")
defaultSetting: 'SimpleSetting' = Field(None, alias="default_setting")
email: str = Field(None, alias="email")
forename: str = 'not specified'
groups: List['SimpleGroup'] = Field([], alias="groups")
id: str = Field(None, alias="url")
identifier: str = 'user'
link: str = 'empty'
name: str = Field(None, alias="username")
surname: str = 'not specified'
settings: List['SimpleSetting'] = Field([], alias="settings")
@staticmethod
def resolve_affiliation(obj: User):
# TODO
return {
"city": "not specified",
"country": "not specified",
"workplace": "not specified"
}
@staticmethod
def resolve_settings(obj: User):
return SettingManager.get_all_settings(obj)
@router.get("/user", response={200: UserWrapper, 403: Error})
def get_users(request, whoami: str = None):
if whoami:
return {'user': [request.user]}
else:
return {'user': User.objects.all()}
@router.get("/user/{uuid:user_uuid}", response={200: UserSchema, 403: Error})
def get_user(request, user_uuid):
try:
u = UserManager.get_user_by_id(request.user, user_uuid)
return u
except ValueError:
return 403, {
'message': f'Getting User with id {user_uuid} failed due to insufficient rights!'}
###########
# Package #
###########
class SimplePackage(SimpleObject):
identifier: str = 'package'
reviewStatus: bool = Field(None, alias="reviewed")
class PackageWrapper(Schema):
package: List['PackageSchema']
class PackageSchema(Schema):
@ -138,10 +265,6 @@ class PackageSchema(Schema):
return 'reviewed' if obj.reviewed else 'unreviewed'
class PackageWrapper(Schema):
package: List['PackageSchema']
@router.get("/package", response={200: PackageWrapper, 403: Error})
def get_packages(request):
return {'package': PackageManager.get_all_readable_packages(request.user, include_reviewed=True)}
@ -156,11 +279,58 @@ def get_package(request, package_uuid):
'message': f'Getting Package with id {package_uuid} failed due to insufficient rights!'}
@router.post("/package")
def create_packages(request, packageName: Form[str], packageDescription: Optional[str] = Form(None)):
try:
if packageName.strip() == '':
raise ValueError('Package name cannot be empty!')
new_pacakge = PackageManager.create_package(request.user, packageName, packageDescription)
return redirect(new_pacakge.url)
except ValueError as e:
return 400, {'message': str(e)}
@router.post("/package/{uuid:package_uuid}", response={200: PackageSchema | Any , 400: Error})
def update_package(
request,
package_uuid,
packageDescription: Optional[str] = Form(None),
hiddenMethod: Optional[str] = Form(None),
exportAsJson: Optional[str] = Form(None),
permissions: Optional[str] = Form(None),
ppsURI: Optional[str] = Form(None),
read: Optional[str] = Form(None),
write: Optional[str] = Form(None),
):
try:
p = PackageManager.get_package_by_id(request.user, package_uuid)
if hiddenMethod:
if hiddenMethod == 'DELETE':
p.delete()
elif packageDescription and packageDescription.strip() != '':
p.description = packageDescription
p.save()
return
elif exportAsJson == 'true':
pack_json = PackageManager.export_package(p, include_models=False, include_external_identifiers=False)
return pack_json
elif all([permissions, ppsURI, read]):
PackageManager.update_permissions
elif all([permissions, ppsURI, write]):
pass
except ValueError as e:
return 400, {'message': str(e)}
################################
# Compound / CompoundStructure #
################################
class SimpleCompound(SimpleObject):
identifier: str = 'compound'
class CompoundWrapper(Schema):
compound: List['SimpleCompound']
class CompoundPathwayScenario(Schema):
@ -218,15 +388,6 @@ class CompoundSchema(Schema):
]
class CompoundWrapper(Schema):
compound: List['SimpleCompound']
class SimpleCompoundStructure(SimpleObject):
identifier: str = 'structure'
reviewStatus: bool = Field(None, alias="compound.package.reviewed")
class CompoundStructureSchema(Schema):
InChI: str = Field(None, alias="inchi")
aliases: List[str] = Field([], alias="aliases")
@ -260,8 +421,6 @@ class CompoundStructureSchema(Schema):
@staticmethod
def resolve_charge(obj: CompoundStructure):
print(obj.smiles)
print(FormatConverter.charge(obj.smiles))
return FormatConverter.charge(obj.smiles)
@staticmethod
@ -302,10 +461,11 @@ class CompoundStructureWrapper(Schema):
@router.get("/compound", response={200: CompoundWrapper, 403: Error})
def get_compounds(request):
qs = Compound.objects.none()
for p in PackageManager.get_reviewed_packages():
qs |= Compound.objects.filter(package=p)
return {'compound': qs}
return {
'compound': Compound.objects.filter(
package__in=PackageManager.get_reviewed_packages()
).prefetch_related('package')
}
@router.get("/package/{uuid:package_uuid}/compound", response={200: CompoundWrapper, 403: Error})
@ -354,14 +514,6 @@ def get_package_compound_structure(request, package_uuid, compound_uuid, structu
#########
# Rules #
#########
class SimpleRule(SimpleObject):
identifier: str = 'rule'
@staticmethod
def resolve_url(obj: Rule):
return obj.url.replace('-ambit-', '-').replace('-rdkit-', '-')
class RuleWrapper(Schema):
rule: List['SimpleRule']
@ -469,10 +621,11 @@ class CompositeRuleSchema(Schema):
@router.get("/rule", response={200: RuleWrapper, 403: Error})
def get_rules(request):
qs = Rule.objects.none()
for p in PackageManager.get_reviewed_packages():
qs |= Rule.objects.filter(package=p)
return {'rule': qs}
return {
'rule': Rule.objects.filter(
package__in=PackageManager.get_reviewed_packages()
).prefetch_related('package')
}
@router.get("/package/{uuid:package_uuid}/rule", response={200: RuleWrapper, 403: Error})
@ -556,10 +709,6 @@ def _post_package_rule(request, package_uuid, rule_uuid, compound: Form[str]):
############
# Reaction #
############
class SimpleReaction(SimpleObject):
identifier: str = 'reaction'
class ReactionWrapper(Schema):
reaction: List['SimpleReaction']
@ -613,10 +762,11 @@ class ReactionSchema(Schema):
@router.get("/reaction", response={200: ReactionWrapper, 403: Error})
def get_reactions(request):
qs = Reaction.objects.none()
for p in PackageManager.get_reviewed_packages():
qs |= Reaction.objects.filter(package=p)
return {'reaction': qs}
return {
'reaction': Reaction.objects.filter(
package__in=PackageManager.get_reviewed_packages()
).prefetch_related('package')
}
@router.get("/package/{uuid:package_uuid}/reaction", response={200: ReactionWrapper, 403: Error})
@ -642,10 +792,6 @@ def get_package_reaction(request, package_uuid, reaction_uuid):
############
# Scenario #
############
class SimpleScenario(SimpleObject):
identifier: str = 'scenario'
class ScenarioWrapper(Schema):
scenario: List['SimpleScenario']
@ -676,10 +822,11 @@ class ScenarioSchema(Schema):
@router.get("/scenario", response={200: ScenarioWrapper, 403: Error})
def get_scenarios(request):
qs = Scenario.objects.none()
for p in PackageManager.get_reviewed_packages():
qs |= Scenario.objects.filter(package=p)
return {'scenario': qs}
return {
'scenario': Scenario.objects.filter(
package__in=PackageManager.get_reviewed_packages()
).prefetch_related('package')
}
@router.get("/package/{uuid:package_uuid}/scenario", response={200: ScenarioWrapper, 403: Error})
@ -705,18 +852,112 @@ def get_package_scenario(request, package_uuid, scenario_uuid):
###########
# Pathway #
###########
class SimplePathway(SimpleObject):
identifier: str = 'pathway'
class PathwayWrapper(Schema):
pathway: List['SimplePathway']
class PathwayEdge(Schema):
ecNumbers: List[str] = Field([], alias="ec_numbers")
id: str = Field(None, alias="url")
idreaction: str = Field(None, alias="edge_label.url")
multstep: bool = Field(None, alias="edge_label.multi_step")
name: str = Field(None, alias="name")
pseudo: bool = False
rule: Optional[str] = Field(None, alias="rule")
scenarios: List[SimpleScenario] = Field([], alias="scenarios")
source: int = -1
target: int = -1
@staticmethod
def resolve_rule(obj: Edge):
if obj.edge_label.rules.all().exists():
r = obj.edge_label.rules.all()[0]
if isinstance(r, SimpleAmbitRule):
return r.smirks
return None
class PathwayNode(Schema):
atomCount: int = Field(None, alias="atom_count")
depth: int = Field(None, alias="depth")
dt50s: List[Dict[str, str]] = Field([], alias="dt50s")
engineeredIntermediate: bool = Field(None, alias="engineered_intermediate")
id: str = Field(None, alias="url")
idcomp: str = Field(None, alias="default_node_label.url")
idreact: str = Field(None, alias="default_node_label.url")
image: str = Field(None, alias="image")
imageSize: int = Field(None, alias="image_size")
name: str = Field(None, alias="name")
proposed: List[Dict[str, str]] = Field([], alias="proposed_intermediate")
smiles: str = Field(None, alias="default_node_label.smiles")
@staticmethod
def resolve_atom_count(obj: Node):
from rdkit import Chem
return Chem.MolFromSmiles(obj.default_node_label.smiles).GetNumAtoms()
@staticmethod
def resolve_dt50s(obj: Node):
# TODO
return []
@staticmethod
def resolve_engineered_intermediate(obj: Node):
# TODO
return False
@staticmethod
def resolve_image(obj: Node):
return f"{obj.default_node_label.url}?image=svg"
@staticmethod
def resolve_image_size(obj: Node):
return 400
@staticmethod
def resolve_proposed_intermediate(obj: Node):
# TODO
return []
class PathwaySchema(Schema):
aliases: List[str] = Field([], alias="aliases")
completed: str = Field(None, alias="completed")
description: str = Field(None, alias="description")
id: str = Field(None, alias="url")
isIncremental: bool = Field(None, alias="is_incremental")
isPredicted: bool = Field(None, alias="is_predicted")
lastModified: int = Field(None, alias="last_modified")
links: List[PathwayEdge] = Field([], alias="edges")
name: str = Field(None, alias="name")
nodes: List[PathwayNode] = Field([], alias="nodes")
pathwayName: str = Field(None, alias="name")
reviewStatus: str = Field(None, alias="review_status")
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
upToDate: bool = Field(None, alias="up_to_date")
@staticmethod
def resolve_review_status(obj: Pathway):
return 'reviewed' if obj.package.reviewed else 'unreviewed'
@staticmethod
def resolve_completed(obj: Pathway):
return str(obj.completed()).lower()
@staticmethod
def resolve_up_to_date(obj: Pathway):
return "true"
@staticmethod
def resolve_last_modified(obj: Pathway):
return int(obj.modified.timestamp())
@router.get("/pathway", response={200: PathwayWrapper, 403: Error})
def get_pathways(request):
qs = Pathway.objects.none()
for p in PackageManager.get_reviewed_packages():
qs |= Pathway.objects.filter(package=p)
return {'pathway': qs}
return {
'pathway': Pathway.objects.filter(
package__in=PackageManager.get_reviewed_packages()
).prefetch_related('package')
}
@router.get("/package/{uuid:package_uuid}/pathway", response={200: PathwayWrapper, 403: Error})
@ -729,11 +970,217 @@ def get_package_pathways(request, package_uuid):
'message': f'Getting Pathways for Package with id {package_uuid} failed due to insufficient rights!'}
# @router.get("/package/{uuid:package_uuid}/pathway/{uuid:pathway_uuid}", response={200: Pathway, 403: Error})
# def get_package_pathway(request, package_uuid, pathway_uuid):
# try:
# p = PackageManager.get_package_by_id(request.user, package_uuid)
# return Pathway.objects.get(package=p, uuid=pathway_uuid)
# except ValueError:
# return 403, {
# 'message': f'Getting Pathway with id {pathway_uuid} failed due to insufficient rights!'}
@router.get("/package/{uuid:package_uuid}/pathway/{uuid:pathway_uuid}", response={200: PathwaySchema, 403: Error})
def get_package_pathway(request, package_uuid, pathway_uuid):
try:
p = PackageManager.get_package_by_id(request.user, package_uuid)
return Pathway.objects.get(package=p, uuid=pathway_uuid)
except ValueError:
return 403, {
'message': f'Getting Pathway with id {pathway_uuid} failed due to insufficient rights!'}
@router.post("/package/{uuid:package_uuid}/pathway")
def create_pathway(
request,
package_uuid,
smilesinput: Form[str],
name: Optional[str] = Form(None),
description: Optional[str] = Form(None),
rootOnly: Optional[str] = Form(None),
selectedSetting: Optional[str] = Form(None)
):
try:
p = PackageManager.get_package_by_id(request.user, package_uuid)
stand_smiles = FormatConverter.standardize(smilesinput.strip())
pw = Pathway.create(p, stand_smiles, name=name, description=description)
pw_mode = 'predict'
if rootOnly and rootOnly == 'true':
pw_mode = 'build'
pw.kv.update({'mode': pw_mode})
pw.save()
if pw_mode == 'predict':
setting = request.user.prediction_settings()
if selectedSetting:
setting = SettingManager.get_setting_by_url(request.user, selectedSetting)
pw.setting = setting
pw.save()
from .tasks import predict
predict.delay(pw.pk, setting.pk, limit=-1)
return redirect(pw.url)
except ValueError as e:
print(e)
########
# Node #
########
class NodeWrapper(Schema):
node: List['SimpleNode']
class NodeCompoundStructure(Schema):
id: str = Field(None, alias="url")
image: str = Field(None, alias="image")
smiles: str = Field(None, alias="smiles")
name:str =Field(None, alias="name")
@staticmethod
def resolve_image(obj: CompoundStructure):
return f"{obj.url}?image=svg"
class NodeSchema(Schema):
aliases: List[str] = Field([], alias="aliases")
confidenceScenarios: List[SimpleScenario] = Field([], alias="confidence_scenarios")
defaultStructure: NodeCompoundStructure = Field(None, alias="default_node_label")
depth: int = Field(None, alias="depth")
description: str = Field(None, alias="description")
engineeredIntermediate:bool = Field(None, alias="engineered_intermediate")
halflifes: Dict[str, str] = Field({}, alias="halflife")
id: str = Field(None, alias="url")
identifier: str = 'node'
image: str = Field(None, alias="image")
name: str = Field(None, alias="name")
proposedValues: List[Dict[str, str]] = Field([], alias="proposed_values")
reviewStatus: str = Field(None, alias="review_status")
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
smiles: str = Field(None, alias="default_node_label.smiles")
structures: List['CompoundStructureSchema'] = Field([], alias="node_labels")
@staticmethod
def resolve_engineered_intermediate(obj: Node):
# TODO
return False
@staticmethod
def resolve_image(obj: Node):
return f"{obj.default_node_label.url}?image=svg"
@staticmethod
def resolve_proposed_values(obj: Node):
# TODO
# {
# "scenarioId": "https://envipath.org/package/5882df9c-dae1-4d80-a40e-db4724271456/scenario/1f1a0b67-ce57-4f5a-929c-4fabf3a791fd",
# "scenarioName": "Annex IIA 7.1.1 a), Brumhard (2003) - (00000)"
# }
return []
@staticmethod
def resolve_review_status(obj: Node):
return 'reviewed' if obj.pathway.package.reviewed else 'unreviewed'
@router.get("/package/{uuid:package_uuid}/pathway/{uuid:pathway_uuid}/node", response={200: NodeWrapper, 403: Error})
def get_package_pathway_nodes(request, package_uuid, pathway_uuid):
try:
p = PackageManager.get_package_by_id(request.user, package_uuid)
return {'node': Pathway.objects.get(package=p, uuid=pathway_uuid).nodes}
except ValueError:
return 403, {
'message': f'Getting Nodes for Pathway with id {pathway_uuid} failed due to insufficient rights!'}
@router.get("/package/{uuid:package_uuid}/pathway/{uuid:pathway_uuid}/node/{uuid:node_uuid}", response={200: NodeSchema, 403: Error})
def get_package_pathway_node(request, package_uuid, pathway_uuid, node_uuid):
try:
p = PackageManager.get_package_by_id(request.user, package_uuid)
pw = Pathway.objects.get(package=p, uuid=pathway_uuid)
return Node.objects.get(pathway=pw, uuid=node_uuid)
except ValueError:
return 403, {
'message': f'Getting Node with id {node_uuid} failed due to insufficient rights!'}
########
# Edge #
########
class EdgeWrapper(Schema):
edge: List['SimpleEdge']
class EdgeNode(SimpleNode):
image: str = Field(None, alias="image")
@staticmethod
def resolve_image(obj: Node):
return f"{obj.default_node_label.url}?image=svg"
class EdgeSchema(Schema):
aliases: List[str] = Field([], alias="aliases")
description: str = Field(None, alias="description")
ecNumbers: List[str] = Field([], alias="ec_numbers")
endNodes: List['EdgeNode'] = Field([], alias="end_nodes")
id: str = Field(None, alias="url")
identifier: str = 'edge'
name: str = Field(None, alias="name")
reactionName: str = Field(None, alias="edge_label.name")
reactionURI: str = Field(None, alias="edge_label.url")
reviewStatus: str = Field(None, alias="review_status")
scenarios: List['SimpleScenario'] = Field([], alias="scenarios")
startNodes: List['EdgeNode'] = Field([], alias="start_nodes")
@staticmethod
def resolve_review_status(obj: Node):
return 'reviewed' if obj.pathway.package.reviewed else 'unreviewed'
@router.get("/package/{uuid:package_uuid}/pathway/{uuid:pathway_uuid}/edge", response={200: EdgeWrapper, 403: Error})
def get_package_pathway_edges(request, package_uuid, pathway_uuid):
try:
p = PackageManager.get_package_by_id(request.user, package_uuid)
return {'edge': Pathway.objects.get(package=p, uuid=pathway_uuid).edges}
except ValueError:
return 403, {
'message': f'Getting Edges for Pathway with id {pathway_uuid} failed due to insufficient rights!'}
@router.get("/package/{uuid:package_uuid}/pathway/{uuid:pathway_uuid}/edge/{uuid:edge_uuid}", response={200: EdgeSchema, 403: Error})
def get_package_pathway_edge(request, package_uuid, pathway_uuid, edge_uuid):
try:
p = PackageManager.get_package_by_id(request.user, package_uuid)
pw = Pathway.objects.get(package=p, uuid=pathway_uuid)
return Edge.objects.get(pathway=pw, uuid=edge_uuid)
except ValueError:
return 403, {
'message': f'Getting Edge with id {edge_uuid} failed due to insufficient rights!'}
###########
# Setting #
###########
class SettingWrapper(Schema):
setting: List['SimpleSetting']
class SettingSchema(Schema):
duplicationHash: int = -1
id: str = Field(None, alias="url")
identifier: str = 'setting'
includedPackages: List['SimplePackage'] = Field([], alias="rule_packages")
isPublic: bool = Field(None, alias="public")
name: str = Field(None, alias="name")
normalizationRules: List['SimpleRule']= Field([], alias="normalization_rules")
propertyPlugins: List[str] = Field([], alias="property_plugins")
truncationstrategy: Optional[str] = Field(None, alias="truncation_strategy")
@router.get('/setting', response={200: SettingWrapper, 403: Error})
def get_settings(request):
return {
'setting': SettingManager.get_all_settings(request.user)
}
@router.get('/setting/{uuid:setting_uuid}', response={200: SettingSchema, 403: Error})
def get_setting(request, setting_uuid):
try:
return SettingManager.get_setting_by_id(request.user, setting_uuid)
except ValueError:
return 403, {
'message': f'Getting Setting with id {setting_uuid} failed due to insufficient rights!'}

6
epdb/models.py
View File

@ -1402,13 +1402,13 @@ class Pathway(EnviPathModel, AliasMixin, ScenarioMixin):
# Mode
def is_built(self):
return self.kv.get('mode', 'build') == 'predicted'
return self.kv.get('mode', 'build') == 'build'
def is_predicted(self):
return self.kv.get('mode', 'build') == 'predicted'
def is_predicted(self):
return self.kv.get('mode', 'build') == 'predicted'
def is_incremental(self):
return self.kv.get('mode', 'build') == 'incremental'
# Status
def status(self):