-
Notifications
You must be signed in to change notification settings - Fork 0
/
model_system.py
70 lines (57 loc) · 2.58 KB
/
model_system.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
import os,re, sys
from byo.track import Track, load_track
from byo.io.genome_accessor import GenomeCache, RemoteCache
from byo.io.annotation import AnnotationAccessor
#from byo.io.lazytables import NamedTupleImporter as Importer
import byo.config
import logging
class LazyTranscriptLoader(object):
def __init__(self,system = None):
self.transcripts = None
self.system = system
self.logger = logging.getLogger("LazyTranscriptLoader(system={system.name})".format(system=self.system) )
def __getitem__(self,txname):
if not self.transcripts:
self.transcripts = self.load_transcript_catalogs()
return self.transcripts[txname]
def load_transcript_catalogs(self):
from byo.gene_model import transcripts_from_UCSC
import glob
path = os.path.join(self.system.root, "annotation", self.system.name, "*.ucsc.gz")
sources = glob.glob(path)
if sources:
self.logger.debug('loading {0}'.format(sources[0]))
T = transcripts_from_UCSC(sources[0],system = self.system)
for s in sources[1:]:
self.logger.debug('loading {0}'.format(s))
T.load(s)
self.logger.info("loaded {0} transcript models from {1} source(s)".format(len(T), len(sources)))
return T
else:
self.logger.error("no transcript models found in path '{0}'".format(path))
return {}
class ModelSystem(object):
def __init__(self, name, genome = None, transcript_models = None, root = byo.config.system_root):
self.name = name
self.root = root
if genome == None:
if getattr(byo.config,"genome_server",None):
self.genome = RemoteCache(byo.config.genome_server)[name]
else:
self.genome = GenomeCache(os.path.join(root,"genomes"))[name]
else:
self.genome = genome
if transcript_models == None:
self.transcript_models = LazyTranscriptLoader(system=self)
else:
self.transcript_models = transcript_models
# Fails before first access due to lazy loading of genome
#self.chr_sizes = self.genome.data.chrom_stats
def get_annotations_track(path="",accessor=AnnotationAccessor,**kwargs):
if not path:
path = os.path.join(self.root,"annotation",self.name,"compiled")
return Track(path,accessor,**kwargs)
def load_track(self, path, **kwargs):
return load_track(path, system=self, **kwargs)
def get_refGenes(self):
return self.transcript_models